minor: name change for a typedef

Author: renukamanavalan

common/events.cpp

@@ -197,7 +197,7 @@ EventSubscriber::~EventSubscriber()
     int rc = 0;
 
     if (m_event_service.is_active()) {
-        events_data_lst_t events;
+        event_serialized_lst_t events;
 
         rc = m_event_service.cache_init();
         RET_ON_ERR(rc == 0, "Failed to init the cache");
@@ -212,7 +212,8 @@ EventSubscriber::~EventSubscriber()
          */
         chrono::steady_clock::time_point start = chrono::steady_clock::now();
         while(true) {
-            string source, evt_str;
+            string source;
+            event_serialized_t evt_str;
             internal_event_t evt_data;
 
             rc = zmq_message_read(m_socket, ZMQ_DONTWAIT, source, evt_data);
@@ -335,7 +336,7 @@ EventSubscriber::event_receive(string &key, event_params_t &params, int &missed_
 
         if (!m_from_cache.empty()) {
 
-            events_data_lst_t::iterator it = m_from_cache.begin();
+            event_serialized_lst_t::iterator it = m_from_cache.begin();
             rc = deserialize(*it, event_data);
             m_from_cache.erase(it);
             RET_ON_ERR(rc == 0, "Failed to deserialize message from cache");

common/events_common.h

@@ -261,9 +261,11 @@ typedef string runtime_id_t;
  *    { EVENT_SEQUENCE, "" } };
  */
 
+typedef vector<internal_event_t> internal_events_lst_t;
+
 /* Cache maintains the part 2 of an event as serialized string. */
-typedef string events_data_type_t;
-typedef vector<events_data_type_t> events_data_lst_t;
+typedef string event_serialized_t;  // events_data_type_t;
+typedef vector<event_serialized_t> event_serialized_lst_t; // events_data_lst_t;
 
 
 sequence_t str_to_seq(const string s);
@@ -278,9 +280,11 @@ 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) {
+    if (rc == -1) {
+        recv_last_err = zerrno;
+    }
+    else {
+        recv_last_err = 0;
         size_t more_size = sizeof (more);
 
         rc = zmsg_to_map(msg, data);
@@ -350,4 +354,25 @@ zmq_message_read(void *sock, int flag, P1 &pt1, P2 &pt2)
     return rc;
 }
 
+/*
+ *  Cache drain timeout.
+ *
+ *  When subscriber's de-init is called, it calls start cache service.
+ *  When subscriber init is called, it calls cache stop service.
+ *
+ *  In either scenario, an entity stops reading and let other start.
+ *  The entity that stops may have to read little longer to drain any
+ *  events in local ZMQ cache.
+ *
+ *  This timeout helps with that.
+ *  
+ *  In case of subscriber de-init, the events read during this period
+ *  is given to cache as start-up or initial stock.
+ *  In case of init where cache service reads for this period, gives
+ *  those as part of cache read and subscriber service will be diligent
+ *  about reading the same event from the channel, hence duplicate
+ *  for next one second.
+ */
+#define CACHE_DRAIN_IN_MILLISECS 1000
+
 #endif /* !_EVENTS_COMMON_H */ 

common/events_pi.h

@@ -122,23 +122,10 @@ class EventSubscriber : public events_base
          * List of cached events.
          * Only part 2 / internal_event_t is cached as serialized string.
          */
-        events_data_lst_t m_from_cache;
+        event_serialized_lst_t m_from_cache;
 
 };
 
-/*
- *  Cache drain timeout.
- *
- *  When de-init is called, it calls stop cache service.
- *  But before this point, there could be events received in zmq's
- *  local cache pending read and those that arrived since last read.
- *  These events will not be seen by cache service.
- *  So read those off and give it to cache service as starting stock.
- *  As we don't have a clue on count in zmq's cache, read in non-block
- *  mode for a period.
- */
-#define CACHE_DRAIN_IN_MILLISECS 1000
-
 /*
  * The uuid_unparse() function converts the supplied UUID uu from
  * the binary representation into a 36-byte string (plus trailing

common/events_service.cpp

@@ -73,7 +73,7 @@ event_service::init_server(void *zmq_ctx, int block_ms)
 int
 event_service::echo_send(const string s)
 {
-    events_data_lst_t l = { s };
+    event_serialized_lst_t l = { s };
 
     return channel_write(EVENT_ECHO, l);
 
@@ -83,7 +83,7 @@ event_service::echo_send(const string s)
 int
 event_service::echo_receive(string &outs)
 {
-    events_data_lst_t l;
+    event_serialized_lst_t l;
     int code;
 
     int rc = channel_read(code, l);
@@ -112,7 +112,7 @@ event_service::cache_init()
 
 
 int
-event_service::cache_start(const events_data_lst_t &lst)
+event_service::cache_start(const event_serialized_lst_t &lst)
 {
     int rc;
 
@@ -136,7 +136,7 @@ event_service::cache_stop()
 
 
 int
-event_service::cache_read(events_data_lst_t &lst)
+event_service::cache_read(event_serialized_lst_t &lst)
 {
     int rc;
 
@@ -148,25 +148,25 @@ event_service::cache_read(events_data_lst_t &lst)
 
 
 int
-event_service::channel_read(int &code, events_data_lst_t &data)
+event_service::channel_read(int &code, event_serialized_lst_t &data)
 {
-    events_data_lst_t().swap(data);
+    event_serialized_lst_t().swap(data);
     return zmq_message_read(m_socket, 0, code, data);
 }
 
 
 int
-event_service::channel_write(int code, const events_data_lst_t &data)
+event_service::channel_write(int code, const event_serialized_lst_t &data)
 {
     return zmq_message_send(m_socket, code, data);
 }
 
 
 int
-event_service::send_recv(int code, const events_data_lst_t *lst_in,
-        events_data_lst_t *lst_out)
+event_service::send_recv(int code, const event_serialized_lst_t *lst_in,
+        event_serialized_lst_t *lst_out)
 {
-    events_data_lst_t l;
+    event_serialized_lst_t l;
     int resp;
 
     if(lst_in == NULL) {

common/events_service.h

@@ -25,7 +25,7 @@
  * In case of data, it comes in seccond part.
  * 
  * Type of data is one or multiple strings, which is sent as serialized vector
- * of strings. events_data_lst_t
+ * of strings. event_serialized_lst_t
  *
  * In case of echo, part2 is the vector of single string as provided in the request.
  *
@@ -133,7 +133,7 @@ class event_service {
          *      1   - Already running
          *      -1  - On failure.
          */
-        int cache_start(const events_data_lst_t &lst);
+        int cache_start(const event_serialized_lst_t &lst);
 
         /*
          *  Called to stop caching events
@@ -179,7 +179,7 @@ class event_service {
          *  0   - On success.
          *  -1  - On failure.
          */
-        int cache_read(events_data_lst_t &lst);
+        int cache_read(event_serialized_lst_t &lst);
 
         /*
          *  Echo send service.
@@ -231,7 +231,7 @@ class event_service {
          *  0   - On success
          *  -1  - On failure
          */
-        int channel_read(int &code, events_data_lst_t &data);
+        int channel_read(int &code, event_serialized_lst_t &data);
 
         /*
          * The under lying write for req/resp from client/server
@@ -247,12 +247,12 @@ class event_service {
          *  0   - On success
          *  -1  - On failure
          */
-        int channel_write(int code, const events_data_lst_t &data);
+        int channel_write(int code, const event_serialized_lst_t &data);
 
         /*
          * send and receive helper.
          * Writes given code & data and reads back data into
-         * provided events_data_lst_t arg and response read is
+         * provided event_serialized_lst_t arg and response read is
          * returned.
          *
          * input:
@@ -265,8 +265,8 @@ class event_service {
          * return:
          *  Any failure or response code from server.
          */
-        int send_recv(int code, const events_data_lst_t *lst_in = NULL,
-                events_data_lst_t *lst_out = NULL);
+        int send_recv(int code, const event_serialized_lst_t *lst_in = NULL,
+                event_serialized_lst_t *lst_out = NULL);
 
         /*
          * de-init/close service

tests/events_common_ut.cpp

@@ -90,6 +90,7 @@ TEST(events_common, send_recv)
         /* Direct log messages to stdout */
         string dummy, op("STDOUT");
         swss::Logger::swssOutputNotify(dummy, op);
+        swss::Logger::setMinPrio(swss::Logger::SWSS_DEBUG);
     }
 #endif
 

tests/events_service_ut.cpp

@@ -14,13 +14,13 @@ static bool do_terminate = false;
 static void *zmq_ctx = NULL;
 static event_service service_cl, service_svr;
 static int server_rd_code, server_ret;
-static events_data_lst_t server_rd_lst, server_wr_lst;
+static event_serialized_lst_t server_rd_lst, server_wr_lst;
 
 /* Mimic the eventd service that handles service requests via dedicated thread */
 void serve_commands()
 {
     int code;
-    events_data_lst_t lst;
+    event_serialized_lst_t lst;
     EXPECT_EQ(0, service_svr.init_server(zmq_ctx, 1000));
     while(!do_terminate) {
         if (0 != service_svr.channel_read(code, lst)) {
@@ -65,13 +65,14 @@ void serve_commands()
 
 TEST(events_common, cache_cmds)
 {
-    events_data_lst_t lst_start, lst;
+    event_serialized_lst_t lst_start, lst;
 
 #if 0
     {
         /* Direct log messages to stdout */
         string dummy, op("STDOUT");
         swss::Logger::swssOutputNotify(dummy, op);
+        swss::Logger::setMinPrio(swss::Logger::SWSS_DEBUG);
     }
 #endif
 
@@ -93,7 +94,7 @@ TEST(events_common, cache_cmds)
      * Bunch of serialized internal_event_t as strings
      * Sending random for test purpose 
      */
-    lst_start = events_data_lst_t(
+    lst_start = event_serialized_lst_t(
             { "hello", "world", "ok" });
     EXPECT_EQ(0, service_cl.cache_start(lst_start));
     EXPECT_EQ(EVENT_CACHE_START, server_rd_code);

tests/events_ut.cpp

@@ -18,7 +18,7 @@ int last_svc_code = -1;
 
 void events_validate_ts(const string s);
 
-events_data_lst_t lst_cache;
+event_serialized_lst_t lst_cache;
 
 #define ARRAY_SIZE(d) (sizeof(d) / sizeof((d)[0]))
 
@@ -34,7 +34,7 @@ void pub_serve_commands()
     EXPECT_EQ(0, service_svr.init_server(zmq_ctx, 1000));
     while(!terminate_svc) {
         int code, resp;
-        events_data_lst_t lst;
+        event_serialized_lst_t lst;
 
         if (0 != service_svr.channel_read(code, lst)) {
             /* check client service status, before blocking on read */
@@ -74,7 +74,7 @@ void pub_serve_commands()
     EXPECT_FALSE(service_svr.is_active());
     terminate_svc = false;
     last_svc_code = -1;
-    events_data_lst_t().swap(lst_cache);
+    event_serialized_lst_t().swap(lst_cache);
 }
 
 
@@ -167,6 +167,7 @@ TEST(events, publish)
         /* Direct log messages to stdout */
         string dummy, op("STDOUT");
         swss::Logger::swssOutputNotify(dummy, op);
+        swss::Logger::setMinPrio(swss::Logger::SWSS_DEBUG);
     }
 
     string evt_source0("sonic-events-bgp");
@@ -476,6 +477,7 @@ TEST(events, subscribe)
         /* Direct log messages to stdout */
         string dummy, op("STDOUT");
         swss::Logger::swssOutputNotify(dummy, op);
+        swss::Logger::setMinPrio(swss::Logger::SWSS_DEBUG);
     }
 #endif