Made zmq send & receive thread safe

Author: renukamanavalan

common/events.cpp

@@ -215,10 +215,8 @@ EventSubscriber::~EventSubscriber()
             internal_event_t evt_data;
 
             rc = zmq_message_read(m_socket, ZMQ_DONTWAIT, source, evt_data);
-            if (rc == -1) {
-                if (zerrno == EAGAIN) {
-                    rc = 0;
-                }
+            if (rc != 0) {
+                /* Break on any failure, including EAGAIN */
                 break;
             }
 
@@ -344,7 +342,7 @@ EventSubscriber::event_receive(string &key, event_params_t &params, int &missed_
             /* Read from SUBS channel */
             string evt_source;
             rc = zmq_message_read(m_socket, 0, evt_source, event_data);
-            RET_ON_ERR(rc == 0, "Failed to read message from sock");
+            RET_ON_ERR(rc == 0, "Failed to read message from sock rc=%d", rc);
         }
 
         /* Find any missed events for this runtime ID */
@@ -441,8 +439,3 @@ event_receive_wrap(event_handle_t handle)
     return op;
 }
 
-int event_last_error()
-{
-    return zerrno;
-}
-

common/events.h

@@ -96,8 +96,9 @@ typedef std::map<std::string, std::string> event_params_t;
  *      e.g. "2022-08-17T02:39:21.286611Z"
  *
  * return:
- *  0   -   On success
- *  -1  -   On failure.
+ *  0   - On success
+ *  > 0 - On failure, returns zmq_errno, if failure is zmq socket related.
+ *  < 0 - For all other failures
  */
 int event_publish(event_handle_t handle, const std::string event_tag,
         const event_params_t *params=NULL);
@@ -187,8 +188,9 @@ void events_deinit_subscriber(event_handle_t handle);
  *      missed count from all received events will give the total missed.
  *
  * return:
- *  0 - On success
- * -1 - On failure. The handle is not valid or upon receive timeout.
+ *  0   - On success
+ *  > 0 - On failure, returns zmq_errno, if failure is zmq socket related.
+ *  < 0 - For all other failures
  *
  */
 int event_receive(event_handle_t handle, std::string &key,
@@ -219,14 +221,8 @@ typedef struct {
 
 event_receive_op_t event_receive_wrap(event_handle_t handle);
 
-
-/*
- *  Get error code for last receive
- *
- *  Set to EAGAIN on timeout
- *  Any other value implies fatal error.
- *
- */
-int event_last_error();
+/* Non ZMQ Error codes */
+#define ERR_MESSAGE_INVALID -2
+#define ERR_OTHER -1
 
 #endif /* !_EVENTS_H */ 

common/events_common.cpp

@@ -1,6 +1,5 @@
 #include "events_common.h"
 
-int zerrno = 0;
 int running_ut = 0;
 
 /*