zeroMQrr wrapper for Matlab: allow to specify an url in GetResponses command

Author: jonaskn

Resources/Matlab/compile_matlab_zeroMQrr.m

@@ -1,4 +1,3 @@
-% 
 % zeroMQrr provides a simple wrapper around the zeromq library with a
 % request->response scenario in mind
 %
@@ -12,9 +11,11 @@
 %             When blocking is 0 (default) zeroMQrr adds the request to a queue and returns the time the request was added to that queue. See 'GetResponses'
 %             [timeRequestAdded]=zeroMQrr('Send',url, request, 0);
 %             
-%      dialogue = zeroMQrr('GetResponses', [wairForEmptyQueue=1])   
-%             Retirieves responses collected for requests send in non blocking mode. If wairForEmptyQueue>0, the function waits for the last queued request to get a response.
-%             dialogue is a struct with the fields: request, response, timeRequestAdded, timeRequestSent, timeResponeRecieved
+%      dialogue = zeroMQrr('GetResponses' [, url] [,wairForEmptyQueue=1])
+%             Retirieves responses collected for requests send in non blocking mode. If an url is provided
+%             only responses for that url will be returend. If wairForEmptyQueue>0, the function waits for
+%             the last queued request to get a response.
+%             Dialogue is a struct with the fields: request, response, timeRequestAdded, timeRequestSent, timeResponeRecieved
 %             At the moment this cannot get filtered by the url of the request yet.
 %             
 %      zeroMQrr('CloseAll')   
@@ -36,4 +37,4 @@
 % it under the terms of the GNU General Public License as published by
 % the Free Software Foundation (see GPL.txt)
 %
-% Changes to allow receiving responses by Jonas Kn?ll, 2016
+% Changes to allow receiving responses by Jonas Knoell, 2016

Resources/Matlab/compile_matlab_zeroMQwrapper.m

@@ -86,7 +86,8 @@ std::mutex socketMap_mutex;
 std::queue<dialogue> InDialogues;
 std::mutex InDialogues_mutex;
 
-std::queue<dialogue> OutDialogues;
+typedef std::list<dialogue> dlist;
+dlist OutDialogues;
 std::mutex OutDialogues_mutex;
 
 typedef std::list<ThreadData*> lst;
@@ -111,7 +112,7 @@ void CloseContext(void)
             (*i)->thread_running=false;
         }
         threadList_mutex.unlock();
-        
+
         bool threadListEmpty=false;
         while(!threadListEmpty){
             threadList_mutex.lock();
@@ -124,7 +125,7 @@ void CloseContext(void)
                 do_sleep(ms10);
             }
         }
-            
+
 		for (mp::iterator i= socketMap.begin(); i != socketMap.end(); i++) {
             int toms=0;
             zmq_setsockopt(i->second.zmqSocket,ZMQ_LINGER,&toms, sizeof(toms));
@@ -136,6 +137,25 @@ void CloseContext(void)
 	}
 }
 
+bool close_socket(std::string url){
+
+    bool success=false;
+    socketMap_mutex.lock();
+    mp::iterator it=socketMap.find(url);
+    socketData sock;
+	int toms=0;
+    if (it != socketMap.end()){
+        sock = it->second;
+        zmq_setsockopt(sock.zmqSocket,ZMQ_LINGER,&toms, sizeof(toms));
+		zmq_close (sock.zmqSocket);
+        socketMap.erase(it);
+        success = true;
+    }
+    socketMap_mutex.unlock();
+
+    return success;
+}
+
 void initialize_zmq()
 {
 	mexAtExit(CloseContext);
@@ -172,9 +192,9 @@ void SendAndReceive( dialogue* d )
 	zmq_msg_t request_z;
 	zmq_msg_init_size (&request_z, d->request.length());
 	memcpy (zmq_msg_data (&request_z), d->request.c_str() , d->request.length());
-	  
+
     zmq_msg_send (&request_z, sock.zmqSocket, 0);
-    
+
     auto time = std::chrono::high_resolution_clock::now().time_since_epoch();
     d->timeRequestSent = (std::chrono::duration_cast<std::chrono::microseconds>(time)).count();
 	zmq_msg_close (&request_z);
@@ -235,7 +255,7 @@ void SendAndReceive( dialogue* d )
             SendAndReceive( &d );
 
             OutDialogues_mutex.lock(); 
-            OutDialogues.push(d);
+            OutDialogues.push_back(d);
             OutDialogues_mutex.unlock(); 
         }else{
             InDialogues_mutex.unlock(); 
@@ -283,7 +303,7 @@ void mexFunction( int nlhs, mxArray* plhs[],
     if (nrhs < 1) {
         return;
     }
-    
+
 	char* Command = mxArrayToString(prhs[0]);
 
 	if   (strcmp(Command, "StartConnectThread") == 0)  {
@@ -313,7 +333,6 @@ void mexFunction( int nlhs, mxArray* plhs[],
 
         if(blocking){
             SendAndReceive(&d);
-            
             if (d.timeResponseReceived==-1) {
                 // Thread was killed and now we try to send things again?
                 mexPrintf("ZMQ: Failed to get a response.\n");
@@ -341,6 +360,9 @@ void mexFunction( int nlhs, mxArray* plhs[],
                 plhs[1]=mxOutStruct;
             }
         }else{ //not blocking
+            if( lastDialogueAdded >= d.timeRequestAdded){ //make sure we get a new timestamp
+                d.timeRequestAdded = lastDialogueAdded+1;
+            }
             lastDialogueAdded = d.timeRequestAdded;
             InDialogues_mutex.lock(); 
             InDialogues.push(d);
@@ -359,7 +381,7 @@ void mexFunction( int nlhs, mxArray* plhs[],
             {
                 mxArray * mxOutStruct;
                 mxOutStruct = mxCreateDoubleMatrix(1,1, mxREAL);
-                memcpy(mxOutStruct ,&(d.timeRequestAdded),sizeof(double));
+                memcpy(mxGetData(mxOutStruct) ,&(d.timeRequestAdded),sizeof(double));
                 plhs[0]=mxOutStruct;
             }
 
@@ -372,26 +394,50 @@ void mexFunction( int nlhs, mxArray* plhs[],
     }
 
 	if (strcmp(Command, "CloseThread") == 0){
-        threadList_mutex.lock();
-        for (lst::iterator i= threadList.begin(); i != threadList.end(); i++) {
-            (*i)->thread_running=false;
+
+        if(nrhs<2){
+            return;
+        }
+        close_socket(std::string(mxArrayToString(prhs[1])));
+
+        //close thread only if socketMap empty
+        socketMap_mutex.lock();
+        if(!socketMap.empty()){
+            socketMap_mutex.unlock();
+            threadList_mutex.lock();
+            for (lst::iterator i= threadList.begin(); i != threadList.end(); i++) {
+                (*i)->thread_running=false;
+            }
+            threadList_mutex.unlock();
+        }else{
+            socketMap_mutex.unlock();
         }
-        threadList_mutex.unlock();
 	}
 
-   if (strcmp(Command, "GetResponses") == 0)
-   {
-       //do we want to waint untill the current Sending Queue is emptied and all replies are in?
+   if (strcmp(Command, "GetResponses") == 0){
+
        bool wairForEmptyQueue;
-       if(nrhs<3)
-       {
-           wairForEmptyQueue = false;
-       }else
-       {
-           double Tmp = mxGetScalar(prhs[2]);
+       std::string url;
+       //if no second input or second input is no string
+       if ( nrhs<2 || !mxIsChar(prhs[1]) ){
+           url = "*";
+       }else{
+            url = (mxArrayToString(prhs[1]));
+       }
+
+       //do we want to waint untill the current Sending Queue is emptied and all replies are in?
+       if(nrhs<2 || (nrhs<3 && mxIsChar(prhs[1])) ){//if only one input or 2 and that is an url
+           wairForEmptyQueue = true;
+       }else{
+           double Tmp;
+           if (mxIsChar(prhs[1])){ //if second is the url, take the third
+            Tmp = mxGetScalar(prhs[2]);
+           }else{
+            Tmp = mxGetScalar(prhs[1]);
+           }
            wairForEmptyQueue = Tmp>0;
        }
-         
+
        double lastAddedTime = lastDialogueAdded;
        double lastFetchedTime = lastDialogueFetched;
        if(lastFetchedTime ==lastAddedTime){ //nothing to fetch
@@ -400,44 +446,57 @@ void mexFunction( int nlhs, mxArray* plhs[],
        //if so, do it now
        while(wairForEmptyQueue)
        {
-          OutDialogues_mutex.lock(); //we might still be one respone behind...         
-          if(!OutDialogues.empty() && OutDialogues.back().timeRequestAdded>=lastAddedTime)
+           OutDialogues_mutex.lock();
+         if(!OutDialogues.empty() && OutDialogues.back().timeRequestAdded>=lastAddedTime)
           {
-               wairForEmptyQueue=false;
+             wairForEmptyQueue=false;
           }
           OutDialogues_mutex.unlock();     
           do_sleep(ms10);
        }
-       
-       //now we build a struct with all replies
+
+       //find the ones we want to return
+       std::queue<dialogue> returnDialogues;
        OutDialogues_mutex.lock();
        if(!OutDialogues.empty()){
         lastDialogueFetched=OutDialogues.back().timeRequestAdded;
+
+        dlist::iterator it=OutDialogues.begin();
+        while(it!=OutDialogues.end()){
+         if(!url.compare("*") || !it->url.compare(url)){
+          dialogue d= (*it);
+          returnDialogues.push(d);
+          it=OutDialogues.erase(it);
+         }else{
+             it++;
+         }
+        }
        }
-       
-       mxArray * mxOutStruct = mxCreateStructMatrix(1,OutDialogues.size(),5,dialogueFieldnames);		
-       
+       OutDialogues_mutex.unlock();
+
+       //now we build a struct with all replies
+       mxArray * mxOutStruct = mxCreateStructMatrix(1,returnDialogues.size(),5,dialogueFieldnames);
+
        int j=0;
-       while(!OutDialogues.empty())
+       while(!returnDialogues.empty())
        {
-           dialogue d = OutDialogues.front();
-           OutDialogues.pop();
-           
+           dialogue d = returnDialogues.front();
+           returnDialogues.pop();
+
            mxSetField(mxOutStruct, j, "request",  mxCreateString( d.request.c_str() ) );
            mxSetField(mxOutStruct, j, "response",  mxCreateString( d.response.c_str() ) );
-           
+
            mxSetField(mxOutStruct,j, "timeRequestAdded", mxCreateDoubleMatrix(1,1, mxREAL));
        	memcpy(mxGetData(mxGetField(mxOutStruct,j, "timeRequestAdded")) ,&(d.timeRequestAdded),sizeof(double));
-           
+
            mxSetField(mxOutStruct,j, "timeRequestSent", mxCreateDoubleMatrix(1,1, mxREAL));
        	memcpy(mxGetData(mxGetField(mxOutStruct,j, "timeRequestSent")) ,&(d.timeRequestSent),sizeof(double));
-           
+
            mxSetField(mxOutStruct,j, "timeResponseReceived", mxCreateDoubleMatrix(1,1, mxREAL));
        	memcpy(mxGetData(mxGetField(mxOutStruct,j, "timeResponseReceived")) ,&(d.timeResponseReceived),sizeof(double));
            j++;
        }
-       OutDialogues_mutex.unlock();
-       
+
        plhs[0] = mxOutStruct;
    }
 }

Resources/Matlab/libzmq-v110-mt-3_2_2.dll

@@ -1,16 +1,56 @@
-url = 'localhost:5556'; % or, e.g., //'tcp://10.71.212.19:5556 if GUI runs on another machine...
+function zeroMQrr_example(url)
 
-% handle = zeroMQrr('StartConnectThread',url); not necessary anymore
+if nargin<1
+    url = 'tcp://localhost:5556'; % or, e.g., //'tcp://10.71.212.19:5556 if GUI runs on another machine...
+end
+
+%blocking request to get the current recording status
+[isacquiring, details]=zeroMQrr('Send',url, 'IsAcquiring', 1); 
+%also get addiional info. this has a struct as returned by getResponses for
+%that one response
+isrecording=zeroMQrr('Send',url, 'IsRecording', 1); 
+
+createNewDir=true;
+RecDir=[];
+PrependText=[];
+AppendText=[];
+if (strcmp(isacquiring, '1'))
+    message='StartAcquisition';
+%     zeroMQrr('Send',url, 'StartAcquisition', 1); 
+    %options:
+    if createNewDir
+        message=[message ' CreateNewDir=1'];
+    end
+    if ~ismepty(RecDir)
+        message=[message ' RecDir=' RecDir];
+    end
+    if ~ismepty(PrependText)
+        message=[message ' PrependText=' PrependText];
+    end
+    if ~ismepty(AppendText)
+        message=[message ' AppendText=' AppendText];
+    end
+
+%     message=sprintf('StartAcquisition CreateNewDir=%i RecDir=%s PrependText=%s AppendText=%s');
+    zeroMQrr('Send',url, message); 
+end
+if (strcmp(isrecording,'1'))
+    zeroMQrr('Send',url, 'StartRecord'); 
+end
+
+%wait for responsess
+responses=zeroMQrr('GetResponses', url); 
+
+recordingNumber=zeroMQrr('Send',url, 'getRecordingNumber',1); 
+experimentNumber=zeroMQrr('Send',url, 'getExperimentNumber',1); 
 
 zeroMQrr('Send',url ,'ClearDesign');
 zeroMQrr('Send',url ,'NewDesign nGo_Left_Right');
 zeroMQrr('Send',url ,'AddCondition Name GoRight TrialTypes 1 2 3');
 zeroMQrr('Send',url ,'AddCondition Name GoLeft TrialTypes 4 5 6');
 
-%blocking request to get the current recording status
-isrecording=zeroMQrr('Send',url, 'IsRecording', 1); 
-
-tic; while toc < 2; end;
+%wait for responsess
+responses=zeroMQrr('GetResponses', url); 
 
 for k = 1:10
     % indicate trial type number #2 has started
@@ -22,8 +62,29 @@
     tic; while toc < 1; end;
 end
 
+zeroMQrr('Send',url, 'StopRecord', 0); 
+
+zeroMQrr('Send',url, 'StopAcquisition', 0); 
+
 %get all responses to the 'Send' requests since the last call to 'GetResponses'
+%default: block untill all responses are in
 blockTillQueueEmpty=1;%default
+responses=zeroMQrr('GetResponses'); 
+responses=zeroMQrr('GetResponses',blockTillQueueEmpty); 
+%caviat: will also block if only responses in queue are for another url
+responses=zeroMQrr('GetResponses',url); 
+responses=zeroMQrr('GetResponses',url,blockTillQueueEmpty); 
+
+%%or choose only to get responses already obtained
+blockTillQueueEmpty=0;
 responses=zeroMQrr('GetResponses',blockTillQueueEmpty); 
+%caviat: will also block if only responses in queue are for another url
+responses=zeroMQrr('GetResponses',url,blockTillQueueEmpty); 
+
+zeroMQrr('CloseThread', url);
+
+zeroMQrr('CloseAll');
+
+
+end 
 
-zeroMQrr('CloseThread');