zmq fixed to have additional json headers using rx_zmqjsonheader

slsDetectorSoftware/commonFiles/sls_detector_defs.h

@@ -196,7 +196,8 @@ enum networkParameter {
   RECEIVER_STREAMING_PORT,	/**< receiever streaming TCP(ZMQ) port */
   CLIENT_STREAMING_PORT,	/**< client streaming TCP(ZMQ) port */
   RECEIVER_STREAMING_SRC_IP,/**< receiever streaming TCP(ZMQ) ip */
-  CLIENT_STREAMING_SRC_IP	/**< client streaming TCP(ZMQ) ip */
+  CLIENT_STREAMING_SRC_IP,	/**< client streaming TCP(ZMQ) ip */
+  ADDITIONAL_JSON_HEADER    /**< additional json header (ZMQ) */
 };
 
 /**

slsDetectorSoftware/slsDetector/slsDetector.cpp

@@ -563,6 +563,8 @@ int slsDetector::initializeDetectorSize(detectorType type) {
     memset(thisDetector->zmqip,0,MAX_STR_LENGTH);
     /** set zmq tcp src ip address in receiver*/
     memset(thisDetector->receiver_zmqip,0,MAX_STR_LENGTH);
+    /** set additional json header in receiver*/
+    memset(thisDetector->receiver_additionalJsonHeader,0,MAX_STR_LENGTH);
 
 
     /** sets onlineFlag to OFFLINE_FLAG */
@@ -6214,6 +6216,9 @@ string slsDetector::setNetworkParameter(networkParameter index, string value) {
 		return setClientStreamingIP(value);
 	case RECEIVER_STREAMING_SRC_IP:
 		return setReceiverStreamingIP(value);
+	case ADDITIONAL_JSON_HEADER:
+	    return setAdditionalJsonHeader(value);
+
   default:
     return (char*)("unknown network parameter");
   }
@@ -6252,6 +6257,8 @@ string slsDetector::getNetworkParameter(networkParameter index) {
   	return getClientStreamingIP();
   case RECEIVER_STREAMING_SRC_IP:
 	return getReceiverStreamingIP();
+  case ADDITIONAL_JSON_HEADER:
+      return getAdditionalJsonHeader();
   default:
     return (char*)("unknown network parameter");
   }
@@ -6352,6 +6359,7 @@ string slsDetector::setReceiver(string receiverIP){
 		std::cout << "flippeddatax:" << thisDetector->flippedData[d] << endl;
 		std::cout << "10GbE:" << thisDetector->tenGigaEnable << endl << endl;
 		std::cout << "rx streaming source ip:" << thisDetector->receiver_zmqip << endl;
+        std::cout << "rx additional json header:" << thisDetector->receiver_additionalJsonHeader << endl;
 		std::cout << "enable gap pixels:" << thisDetector->gappixels << endl;
 		std::cout << "rx streaming port:" << thisDetector->receiver_zmqport << endl;
 		std::cout << "r_readfreq:" << thisDetector->receiver_read_freq << endl << endl;
@@ -6399,6 +6407,7 @@ string slsDetector::setReceiver(string receiverIP){
 			setReadReceiverFrequency(thisDetector->receiver_read_freq);
 			setReceiverStreamingPort(getReceiverStreamingPort());
 			setReceiverStreamingIP(getReceiverStreamingIP());
+			setAdditionalJsonHeader(getAdditionalJsonHeader());
 			enableDataStreamingFromReceiver(enableDataStreamingFromReceiver(-1));
 		}
 	}
@@ -6644,6 +6653,39 @@ string slsDetector::setReceiverStreamingIP(string sourceIP) {
 
 
 
+string slsDetector::setAdditionalJsonHeader(string jsonheader) {
+
+    int fnum=F_ADDITIONAL_JSON_HEADER;
+    int ret = FAIL;
+    char arg[MAX_STR_LENGTH];
+    memset(arg,0,sizeof(arg));
+    char retval[MAX_STR_LENGTH];
+    memset(retval,0, sizeof(retval));
+
+    strcpy(arg, jsonheader.c_str());
+
+    if(thisDetector->receiverOnlineFlag==ONLINE_FLAG){
+#ifdef VERBOSE
+        std::cout << "Sending additional json header " << arg << std::endl;
+#endif
+        if (connectData() == OK){
+            ret=thisReceiver->sendString(fnum,retval,arg);
+            disconnectData();
+        }
+        if(ret==FAIL) {
+              setErrorMask((getErrorMask())|(COULDNOT_SET_NETWORK_PARAMETER));
+              std::cout << "Warning: Could not set additional json header" << std::endl;
+        } else
+            strcpy(thisDetector->receiver_additionalJsonHeader, retval);
+        if(ret==FORCE_UPDATE)
+            updateReceiver();
+    }
+
+    return getAdditionalJsonHeader();
+}
+
+
+
 string slsDetector::setDetectorNetworkParameter(networkParameter index, int delay){
 	int fnum = F_SET_NETWORK_PARAMETER;
 	int ret = FAIL;
@@ -8819,6 +8861,10 @@ int slsDetector::updateReceiverNoWait() {
   n += 	dataSocket->ReceiveDataOnly(path,MAX_STR_LENGTH);
   strcpy(thisDetector->receiver_zmqip, path);
 
+  // additional json header
+  n +=  dataSocket->ReceiveDataOnly(path,MAX_STR_LENGTH);
+  strcpy(thisDetector->receiver_additionalJsonHeader, path);
+
   // gap pixels
   n += dataSocket->ReceiveDataOnly(&ind,sizeof(ind));
   thisDetector->gappixels = ind;

slsDetectorSoftware/slsDetector/slsDetector.h

@@ -285,6 +285,8 @@ class slsDetector : public slsDetectorUtils, public energyConversion {
     int nGappixels[2];
     /** data bytes including gap pixels */
     int dataBytesInclGapPixels;
+    /** additional json header */
+    char receiver_additionalJsonHeader[MAX_STR_LENGTH];
 
 
 
@@ -1824,6 +1826,8 @@ class slsDetector : public slsDetectorUtils, public energyConversion {
   string getClientStreamingIP(){return string(thisDetector->zmqip);};
   /** gets the zmq source ip in receiver, returns "none" if default setting and no custom ip set*/
   string getReceiverStreamingIP(){return string(thisDetector->receiver_zmqip);};
+  /** gets the additional json header, returns "none" if default setting and no custom set*/
+  string getAdditionalJsonHeader(){return string(thisDetector->receiver_additionalJsonHeader);};
 
   /** validates the format of detector MAC address and sets it \sa sharedSlsDetector  */
   string setDetectorMAC(string detectorMAC);
@@ -1847,6 +1851,8 @@ class slsDetector : public slsDetectorUtils, public energyConversion {
   string setClientStreamingIP(string sourceIP);
   /** sets the zmq source ip in receiver. if empty, uses rx_hostname*/
   string setReceiverStreamingIP(string sourceIP);
+  /** additional json header, returns "none" if default setting and no custom set */
+  string setAdditionalJsonHeader(string jsonheader);
 
   /** sets the transmission delay for left or right port or for an entire frame*/
   string setDetectorNetworkParameter(networkParameter index, int delay);

slsDetectorSoftware/slsDetector/slsDetectorCommand.cpp

@@ -2034,6 +2034,13 @@ slsDetectorCommand::slsDetectorCommand(slsDetectorUtils *det)  {
 	descrToFuncMap[i].m_pFuncPtr=&slsDetectorCommand::cmdNetworkParameter;
 	i++;
 
+    /*! \page network
+   - <b>rx_jsonaddheader [t]</b> sets/gets additional json header to be streamed out with the zmq from receiver. Default is empty. \c t must be in the format "\"label1\":\"value1\",\"label2\":\"value2\"" etc. Use only if it needs to be processed by an intermediate process. \c Returns \c (string)
+     */
+    descrToFuncMap[i].m_pFuncName="rx_jsonaddheader"; //
+    descrToFuncMap[i].m_pFuncPtr=&slsDetectorCommand::cmdNetworkParameter;
+    i++;
+
 	/*! \page network
    - <b>configuremac [i]</b> configures the MAC of the detector with these parameters: detectorip, detectormac, rx_udpip, rx_udpmac, rx_udpport, rx_udpport2 (if applicable). This command is already included in \c rx_hsotname. Only put!. \c Returns \c (int)
 	 */
@@ -4089,6 +4096,8 @@ string slsDetectorCommand::cmdNetworkParameter(int narg, char *args[], int actio
 		// if streaming, switch it off
 		prev_streaming = myDet->enableDataStreamingFromReceiver();
 		if (prev_streaming) myDet->enableDataStreamingFromReceiver(0);
+	} else if (cmd=="rx_jsonaddheader") {
+	    t=ADDITIONAL_JSON_HEADER;
 	}
 	else return ("unknown network parameter")+cmd;
 
@@ -4132,6 +4141,9 @@ string slsDetectorCommand::helpNetworkParameter(int narg, char *args[], int acti
 		os << "rx_zmqip ip \n sets/gets the 0MQ (TCP) ip of the receiver from where data is streamed from (eg. to GUI or another process for further processing). "
 				"Default is ip of rx_hostname and works for GUI. This is usually used to stream out to an external process for further processing."
 				"restarts streaming in receiver with new port" << std::endl;
+		os << "rx_jsonaddheader [t]\n sets additional json header to be streamed "
+		        "out with the zmq from receiver. Default is empty. \c t must be in the format '\"label1\":\"value1\",\"label2\":\"value2\"' etc."
+		        "Use only if it needs to be processed by an intermediate process." << std::endl;
 	}
 	if (action==GET_ACTION || action==HELP_ACTION) {
 		os << "detectormac \n gets detector mac "<< std::endl;
@@ -4148,6 +4160,8 @@ string slsDetectorCommand::helpNetworkParameter(int narg, char *args[], int acti
 		os << "rx_zmqport \n gets the 0MQ (TCP) port of the receiver from where data is streamed from"<< std::endl;
 		os << "zmqip \n gets the 0MQ (TCP) ip of the client to where final data is streamed to.If no custom ip, empty until first time connect to receiver" << std::endl;
 		os << "rx_zmqip \n gets/gets the 0MQ (TCP) ip of the receiver from where data is streamed from. If no custom ip, empty until first time connect to receiver" << std::endl;
+        os << "rx_jsonaddheader \n gets additional json header to be streamed "
+                "out with the zmq from receiver." << std::endl;
 	}
 	return os.str();
 

slsReceiverSoftware/include/DataStreamer.h

@@ -25,8 +25,9 @@ class DataStreamer : private virtual slsReceiverDefs, public ThreadObject {
 	 * @param sEnable pointer to short frame enable
 	 * @param fi pointer to file index
 	 * @param fd flipped data enable for x and y dimensions
+	 * @param ajh additional json header
 	 */
-	DataStreamer(Fifo*& f, uint32_t* dr, int* sEnable, uint64_t* fi, int* fd);
+	DataStreamer(Fifo*& f, uint32_t* dr, int* sEnable, uint64_t* fi, int* fd, char* ajh);
 
 	/**
 	 * Destructor
@@ -233,5 +234,8 @@ class DataStreamer : private virtual slsReceiverDefs, public ThreadObject {
 
 	/** flipped data across both dimensions enable */
 	int* flippedData;
+
+	/** additional json header */
+	char* additionJsonHeader;
 };
 

slsReceiverSoftware/include/UDPBaseImplementation.h

@@ -271,6 +271,12 @@ class UDPBaseImplementation : protected virtual slsReceiverDefs, public UDPInter
 	 */
 	char *getStreamingSourceIP() const;
 
+    /**
+     * Get additional json header
+     * @return additional json header
+     */
+    char *getAdditionalJsonHeader() const;
+
 
 	/*************************************************************************
 	 * Setters ***************************************************************
@@ -555,6 +561,11 @@ class UDPBaseImplementation : protected virtual slsReceiverDefs, public UDPInter
 	 */
 	void setStreamingSourceIP(const char* c);
 
+    /**
+     * Set additional json header
+     */
+    void setAdditionalJsonHeader(const char* c);
+
 	/*
 	 * Restream stop dummy packet from receiver
 	 * @return OK or FAIL
@@ -712,6 +723,8 @@ class UDPBaseImplementation : protected virtual slsReceiverDefs, public UDPInter
 	uint32_t streamingPort;
 	/** streaming port */
 	char streamingSrcIP[MAX_STR_LENGTH];
+	/** additional json header */
+	char additionalJsonHeader[MAX_STR_LENGTH];
 
 	//***receiver parameters***
 	uint32_t silentMode;

slsReceiverSoftware/include/UDPInterface.h

@@ -51,6 +51,7 @@ class UDPInterface {
 	 *	-setGapPixelsEnable
 	 *	-setStreamingPort
 	 *	-setStreamingSourceIP
+	 *	-setAdditionalJsonHeader
 	 *	-setDataStreamEnable
 	 *
 	 *
@@ -357,6 +358,12 @@ class UDPInterface {
 	 */
 	virtual char *getStreamingSourceIP() const = 0;
 
+    /**
+     * Get additional json header
+     * @return additional json header
+     */
+    virtual char *getAdditionalJsonHeader() const = 0;
+
 
 	/*************************************************************************
 	 * Setters ***************************************************************
@@ -643,6 +650,11 @@ class UDPInterface {
 	 */
 	virtual void setStreamingSourceIP(const char* c) = 0;
 
+    /**
+     * Set additional json header
+     */
+    virtual void setAdditionalJsonHeader(const char* c) = 0;
+
 	/*
 	 * Restream stop dummy packet from receiver
 	 * @return OK or FAIL

slsReceiverSoftware/include/ZmqSocket.h

@@ -276,7 +276,8 @@ public:
 			uint64_t bunchId = 0, uint64_t timestamp = 0,
 			uint16_t modId = 0, uint16_t xCoord = 0, uint16_t yCoord = 0, uint16_t zCoord = 0,
 			uint32_t debug = 0, uint16_t roundRNumber = 0,
-			uint8_t detType = 0, uint8_t version = 0, int* flippedData = 0) {
+			uint8_t detType = 0, uint8_t version = 0, int* flippedData = 0,
+			char* additionalJsonHeader = 0) {
 
 
 		char buf[MAX_STR_LENGTH] = "";
@@ -310,7 +311,7 @@ public:
 		        //additional stuff
 		        "\"flippedDataX\":%u"
 
-				"}\n\0";
+				;//"}\n\0";
 		int length = sprintf(buf, jsonHeaderFormat,
 				jsonversion, dynamicrange, fileIndex, npixelsx, npixelsy, imageSize,
 				acqIndex, fIndex, (fname == NULL)? "":fname, dummy?0:1,
@@ -322,9 +323,15 @@ public:
 						//additional stuff
 						((flippedData == 0 ) ? 0 :flippedData[0])
 		);
+		if (additionalJsonHeader && strlen(additionalJsonHeader)) {
+		    length = sprintf(buf, "%s, %s}\n%c", buf, additionalJsonHeader, '\0');
+		} else {
+		    length = sprintf(buf, "%s}\n%c", buf, '\0');
+		}
+
 #ifdef VERBOSE
 		//if(!index)
-			FILE_LOG(logINFO) << index << ": Streamer: buf:" << buf;
+			cprintf(BLUE,"%d : STreamer: buf: %s\n", index, buf);
 #endif
 
 		if(zmq_send (socketDescriptor, buf, length, dummy?0:ZMQ_SNDMORE) < 0) {

slsReceiverSoftware/include/slsReceiverTCPIPInterface.h

@@ -297,6 +297,9 @@ class slsReceiverTCPIPInterface : private virtual slsReceiverDefs {
 	/** restream stop packet */
 	int restream_stop();
 
+    /** set additional json header */
+    int set_additional_json_header();
+
 
 
 	/** detector type */

slsReceiverSoftware/include/sls_receiver_funcs.h

@@ -63,6 +63,8 @@ enum recFuncs{
 	F_SET_RECEIVER_SILENT_MODE,		/** < sets the receiver silent mode */
 	F_ENABLE_GAPPIXELS_IN_RECEIVER,	/** < sets gap pixels in the receiver */
 	F_RESTREAM_STOP_FROM_RECEIVER,	/** < restream stop from receiver */
+	F_ADDITIONAL_JSON_HEADER,       /** < additional json header */
+
 
 	/* Always append functions hereafter!!! */
 

slsReceiverSoftware/src/DataStreamer.cpp

@@ -26,7 +26,7 @@ pthread_mutex_t DataStreamer::Mutex = PTHREAD_MUTEX_INITIALIZER;
 bool DataStreamer::SilentMode(false);
 
 
-DataStreamer::DataStreamer(Fifo*& f, uint32_t* dr, int* sEnable, uint64_t* fi, int* fd) :
+DataStreamer::DataStreamer(Fifo*& f, uint32_t* dr, int* sEnable, uint64_t* fi, int* fd, char* ajh) :
 		ThreadObject(NumberofDataStreamers),
 		generalData(0),
 		fifo(f),
@@ -39,7 +39,8 @@ DataStreamer::DataStreamer(Fifo*& f, uint32_t* dr, int* sEnable, uint64_t* fi, i
 		firstAcquisitionIndex(0),
 		firstMeasurementIndex(0),
 		completeBuffer(0),
-		flippedData(fd)
+		flippedData(fd),
+		additionJsonHeader(ajh)
 {
 	if(ThreadObject::CreateThread()){
 		pthread_mutex_lock(&Mutex);
@@ -287,8 +288,9 @@ int DataStreamer::SendHeader(sls_detector_header* header, uint32_t size, uint32_
 			header->modId, header->xCoord, header->yCoord, header->zCoord,
 			header->debug, header->roundRNumber,
 			header->detType, header->version,
-			flippedData
+			flippedData,
-	);
+			additionJsonHeader
+			);
 }
 
 

slsReceiverSoftware/src/UDPBaseImplementation.cpp

@@ -83,6 +83,7 @@ void UDPBaseImplementation::initializeMembers(){
 	dataStreamEnable = false;
 	streamingPort = 0;
 	memset(streamingSrcIP, 0, sizeof(streamingSrcIP));
+    memset(additionalJsonHeader, 0, sizeof(additionalJsonHeader));
 
 	//***receiver parameters***
 	silentMode = 0;
@@ -232,6 +233,16 @@ char *UDPBaseImplementation::getStreamingSourceIP() const{
 	return output;
 }
 
+char *UDPBaseImplementation::getAdditionalJsonHeader() const{
+    FILE_LOG(logDEBUG) << __AT__ << " starting";
+
+    char* output = new char[MAX_STR_LENGTH]();
+    memset(output, 0, MAX_STR_LENGTH);
+    strcpy(output,additionalJsonHeader);
+    //freed by calling function
+    return output;
+}
+
 /*************************************************************************
  * Setters ***************************************************************
  * They modify the local cache of configuration or detector parameters ***
@@ -597,6 +608,11 @@ void UDPBaseImplementation::setStreamingSourceIP(const char c[]){
 	FILE_LOG(logINFO) << "Streaming Source IP: " << streamingSrcIP;
 }
 
+void UDPBaseImplementation::setAdditionalJsonHeader(const char c[]){
+    FILE_LOG(logDEBUG) << __AT__ << " starting";
+    strcpy(additionalJsonHeader, c);
+    FILE_LOG(logINFO) << "Additional JSON Header: " << additionalJsonHeader;
+}
 
 int UDPBaseImplementation::restreamStop() {
 	FILE_LOG(logERROR) << __AT__ << " doing nothing...";

slsReceiverSoftware/src/UDPStandardImplementation.cpp

@@ -218,7 +218,7 @@ int UDPStandardImplementation::setDataStreamEnable(const bool enable) {
 		if (enable) {
 			bool error = false;
 			for ( int i = 0; i < numThreads; ++i ) {
-				dataStreamer.push_back(new DataStreamer(fifo[i], &dynamicRange, &shortFrameEnable, &fileIndex, flippedData));
+				dataStreamer.push_back(new DataStreamer(fifo[i], &dynamicRange, &shortFrameEnable, &fileIndex, flippedData, additionalJsonHeader));
 				dataStreamer[i]->SetGeneralData(generalData);
 				if (dataStreamer[i]->CreateZmqSockets(&numThreads, streamingPort, streamingSrcIP) == FAIL) {
 					error = true;

slsReceiverSoftware/src/slsReceiverTCPIPInterface.cpp

@@ -293,6 +293,7 @@ const char* slsReceiverTCPIPInterface::getFunctionName(enum recFuncs func) {
 	case F_RECEIVER_STREAMING_SRC_IP: 	return "F_RECEIVER_STREAMING_SRC_IP";
 	case F_ENABLE_GAPPIXELS_IN_RECEIVER:return "F_ENABLE_GAPPIXELS_IN_RECEIVER";
 	case F_RESTREAM_STOP_FROM_RECEIVER:	return "F_RESTREAM_STOP_FROM_RECEIVER";
+    case F_ADDITIONAL_JSON_HEADER:      return "F_ADDITIONAL_JSON_HEADER";
 	default:							return "Unknown Function";
 	}
 }
@@ -341,6 +342,7 @@ int slsReceiverTCPIPInterface::function_table(){
 	flist[F_RECEIVER_STREAMING_SRC_IP]		= 	&slsReceiverTCPIPInterface::set_streaming_source_ip;
 	flist[F_ENABLE_GAPPIXELS_IN_RECEIVER]	=	&slsReceiverTCPIPInterface::enable_gap_pixels;
 	flist[F_RESTREAM_STOP_FROM_RECEIVER]	= 	&slsReceiverTCPIPInterface::restream_stop;
+	flist[F_ADDITIONAL_JSON_HEADER]         =   &slsReceiverTCPIPInterface::set_additional_json_header;
 
 #ifdef VERYVERBOSE
 	for (int i = 0; i < NUM_REC_FUNCTIONS ; i++) {
@@ -716,6 +718,14 @@ int slsReceiverTCPIPInterface::send_update() {
 	if (path != NULL)
 		delete[] path;
 
+    // additional json header
+#ifdef SLS_RECEIVER_UDP_FUNCTIONS
+    path = receiverBase->getAdditionalJsonHeader();
+#endif
+    mySock->SendDataOnly(path,MAX_STR_LENGTH);
+    if (path != NULL)
+        delete[] path;
+
 	// gap pixels enable
 #ifdef SLS_RECEIVER_UDP_FUNCTIONS
 	ind = (int)receiverBase->getGapPixelsEnable();
@@ -2549,3 +2559,52 @@ int slsReceiverTCPIPInterface::restream_stop(){
 	// return ok/fail
 	return ret;
 }
+
+
+
+int slsReceiverTCPIPInterface::set_additional_json_header() {
+    ret = OK;
+    memset(mess, 0, sizeof(mess));
+    char arg[MAX_STR_LENGTH];
+    memset(arg, 0, sizeof(arg));
+    char* retval=NULL;
+
+    // receive arguments
+    if (mySock->ReceiveDataOnly(arg,MAX_STR_LENGTH) < 0 )
+        return printSocketReadError();
+
+    // execute action
+#ifdef SLS_RECEIVER_UDP_FUNCTIONS
+    if (receiverBase == NULL)
+        invalidReceiverObject();
+    else {
+        // set
+        if (mySock->differentClients && lockStatus)
+            receiverlocked();
+        else if (receiverBase->getStatus() != IDLE)
+            receiverNotIdle();
+        else {
+                receiverBase->setAdditionalJsonHeader(arg);
+        }
+
+        //get
+        retval = receiverBase->getAdditionalJsonHeader();
+    }
+#endif
+#ifdef VERYVERBOSE
+    FILE_LOG(logDEBUG1) << "additional json header:" << retval;
+#endif
+
+    if (ret == OK && mySock->differentClients)
+        ret = FORCE_UPDATE;
+
+    // send answer
+    mySock->SendDataOnly(&ret,sizeof(ret));
+    if (ret == FAIL)
+        mySock->SendDataOnly(mess,sizeof(mess));
+    mySock->SendDataOnly(retval,MAX_STR_LENGTH);
+    delete[] retval;
+
+    // return ok/fail
+    return ret;
+}