events 75% covered by UT

Author: renukamanavalan

common/events.cpp

@@ -16,9 +16,17 @@
  *
  * The unit is in milliseconds in sync with ZMQ_RCVTIMEO of
  * zmq_setsockopt.
+ *
+ * Publisher uses more to shadow async connectivity from PUB to 
+ * XSUB end point of eventd's proxy. Hene have a less value.
+ *
+ * Subscriber uses it for cache management and here we need a
+ * longer timeout, to handle slow proxy. This timeout value's only
+ * impact could be subscriber process trying to terminate.
  */
 
-#define EVENTS_SERVICE_TIMEOUT_MILLISECS 200
+#define EVENTS_SERVICE_TIMEOUT_MS_PUB 200
+#define EVENTS_SERVICE_TIMEOUT_MS_SUB 2000
 
 /*
  *  Track created publishers to avoid duplicates
@@ -50,7 +58,7 @@ int EventPublisher::init(const string event_source)
      * Event service could be down. So have a timeout.
      * 
      */
-    rc = m_event_service.init_client(m_zmq_ctx, EVENTS_SERVICE_TIMEOUT_MILLISECS);
+    rc = m_event_service.init_client(m_zmq_ctx, EVENTS_SERVICE_TIMEOUT_MS_PUB);
     RET_ON_ERR (rc == 0, "Failed to init event service");
 
     rc = m_event_service.echo_send("hello");
@@ -87,7 +95,7 @@ EventPublisher::publish(const string tag, const event_params_t *params)
         string s;
 
         /* Failure is no-op; The eventd service my be down
-         * NOTE: This call atmost blocks for EVENTS_SERVICE_TIMEOUT_MILLISECS
+         * NOTE: This call atmost blocks for EVENTS_SERVICE_TIMEOUT_MS_PUB
          * as provided in publisher init.
          */
         m_event_service.echo_receive(s);
@@ -240,7 +248,8 @@ EventSubscriber::~EventSubscriber()
 
 
 int
-EventSubscriber::init(bool use_cache, const event_subscribe_sources_t *subs_sources)
+EventSubscriber::init(bool use_cache, int recv_timeout,
+        const event_subscribe_sources_t *subs_sources)
 {
     /*
      *  Initiate SUBS connection to XPUB end point.
@@ -268,8 +277,13 @@ EventSubscriber::init(bool use_cache, const event_subscribe_sources_t *subs_sour
         }
     }
 
+    if (recv_timeout != -1) {
+        rc = zmq_setsockopt (m_socket, ZMQ_RCVTIMEO, &recv_timeout, sizeof (recv_timeout));
+        RET_ON_ERR(rc == 0, "Failed to ZMQ_RCVTIMEO to %d", recv_timeout);
+    }
+
     if (use_cache) {
-        rc = m_event_service.init_client(m_zmq_ctx, EVENTS_SERVICE_TIMEOUT_MILLISECS);
+        rc = m_event_service.init_client(m_zmq_ctx, EVENTS_SERVICE_TIMEOUT_MS_SUB);
         RET_ON_ERR(rc == 0, "Fails to init the service");
 
         if (m_event_service.cache_stop() == 0) {
@@ -379,12 +393,13 @@ EventSubscriber::event_receive(string &key, event_params_t &params, int &missed_
 static EventSubscriber *s_subscriber = NULL;
 
 event_handle_t
-events_init_subscriber(bool use_cache, const event_subscribe_sources_t *sources)
+events_init_subscriber(bool use_cache, int recv_timeout,
+        const event_subscribe_sources_t *sources)
 {
     if (s_subscriber == NULL) {
         EventSubscriber *p = new EventSubscriber();
 
-        RET_ON_ERR(p->init(use_cache, sources) == 0,
+        RET_ON_ERR(p->init(use_cache, recv_timeout, sources) == 0,
                 "Failed to init subscriber");
 
         s_subscriber = p;
@@ -416,4 +431,8 @@ event_receive(event_handle_t handle, string &key,
     return -1;
 }
 
+int event_last_error()
+{
+    return recv_last_err;
+}
 

common/events.h

@@ -113,6 +113,13 @@ typedef std::vector<std::string> event_subscribe_sources_t;
  *      The cache service caches events during session down time. The deinit
  *      start the caching and init call stops the caching.
  *
+ * recv_timeout
+ *      Read blocks by default until an event is available for read.
+ *      0  - Returns immediately, with or w/o event
+ *      -1 - Default; Blocks until an event is available for read
+ *      N  - Count ofd milliseconds to wait for an event.
+ *
+ *
  *  lst_subscribe_sources_t
  *      List of subscription sources of interest.
  *      The source value is the corresponding YANG module name.
@@ -123,6 +130,7 @@ typedef std::vector<std::string> event_subscribe_sources_t;
  *  NULL on failure
  */
 event_handle_t events_init_subscriber(bool use_cache=false,
+        int recv_timeout = -1,
         const event_subscribe_sources_t *sources=NULL);
 
 /*
@@ -174,13 +182,23 @@ void events_deinit_subscriber(event_handle_t &handle);
  *
  * return:
  *  0 - On success
- * -1 - On failure. The handle is not valid.
+ * -1 - On failure. The handle is not valid or upon receive timeout.
  *
  */
 int event_receive(event_handle_t handle, std::string &key,
         event_params_t &params, int &missed_cnt);
 
 
+/*
+ *  Get error code for last receive
+ *
+ *  Set to EAGAIN on timeout
+ *  Any other value implies fatal error.
+ *
+ */
+int event_last_error();
+
+
 /*
  *  Cache drain timeout.
  *

common/events_common.cpp

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

common/events_common.h

@@ -25,6 +25,7 @@ using namespace chrono;
 
 extern int errno;
 extern int zerrno;
+extern int recv_last_err;
 
 /*
  * Max count of possible concurrent event publishers
@@ -287,6 +288,7 @@ zmq_read_part(void *sock, int flag, int &more, DT &data)
     more = 0;
     zmq_msg_init(&msg);
     int rc = zmq_msg_recv(&msg, sock, flag);
+    recv_last_err = zerrno;
 
     if (rc != -1) {
         size_t more_size = sizeof (more);

common/events_pi.h

@@ -93,6 +93,7 @@ class EventSubscriber : public events_base
         EventSubscriber();
 
         int init(bool use_cache=false,
+                int recv_timeout= -1,
                 const event_subscribe_sources_t *subs_sources=NULL);
 
         virtual ~EventSubscriber();