Fix issue where unsubscribing in push_events led to API_EventTimeout … (#699)

Subscribing or unsubscribing events in an event callback when this callback
was called during a subscribe_event() phase was leading to events and
heartbeat events no longer received by the event client.
A bug was fixed when several ZMQ_DELAY_EVENT commands were received
consecutively by the ZMQ control socket.
Stop delaying events only if there is no other ZMQ_DELAY_EVENT command currently requested

Author: bourtemb

cppapi/client/zmqeventconsumer.cpp

@@ -698,6 +698,9 @@ bool ZmqEventConsumer::process_ctrl(zmq::message_t &received_ctrl,zmq::pollitem_
 {
     bool ret = false;
 
+    // static variable to count the number of ZMQ_DELAY_EVENT requests currently in progress:
+    static int nb_current_delay_event_requests = 0;
+
 //
 // For debug and logging purposes
 //
@@ -1075,14 +1078,32 @@ bool ZmqEventConsumer::process_ctrl(zmq::message_t &received_ctrl,zmq::pollitem_
 
         case ZMQ_DELAY_EVENT:
         {
-            old_poll_nb = poll_nb;
-            poll_nb = 1;
+            // If poll_nb == 1, then we are already in a situation where events are being delayed
+            // and we are currently only taking care of messages received on the control socket
+            // No need to update old_poll_nb in this case because it is already correct
+            // otherwise this would lead to issues like https://github.com/tango-controls/cppTango/issues/686
+            // where events would no longer be received if someone subscribes or unsubscribes to events in
+            // an event callback and when the callback is executed during a subscribe_event call
+            if (poll_nb != 1)
+            {
+                old_poll_nb = poll_nb;
+                poll_nb = 1;
+            }
+            nb_current_delay_event_requests++;
         }
         break;
 
         case ZMQ_RELEASE_EVENT:
         {
-            poll_nb = old_poll_nb;
+            if(nb_current_delay_event_requests >= 1)
+            {
+                nb_current_delay_event_requests--;
+            }
+            if(nb_current_delay_event_requests == 0)
+            {
+                // Stop delaying events only if there is no other ZMQ_DELAY_EVENT command requested
+                poll_nb = old_poll_nb;
+            }
         }
         break;