partial compile

Author: renukamanavalan

common/Makefile.am

@@ -29,6 +29,8 @@ endif
 
 libswsscommon_la_SOURCES = \
     events_common.cpp         \
+    events_service.cpp        \
+    events.cpp                \
     logger.cpp                \
     redisreply.cpp            \
     configdb.cpp              \

common/events.cpp

@@ -1,19 +1,32 @@
 #include "events_pi.h"
 
+/*
+ * Publisher use echo service and subscriber uses cache service
+ * The eventd process runs this service, which could be down
+ * All service interactions being async, a timeout is required
+ * not to block forever on read.
+ *
+ * The unit is in milliseconds (in sync with ZMQ_RCVTIMEO of
+ * zmq_setsockopt.
+ */
+
+#define EVENTS_SERVICE_TIMEOUT_MILLISECS 200
+
 /*
  *  Track created publishers to avoid duplicates
  *  As we track missed event count by publishing instances, avoiding
  *  duplicates helps reduce load.
  */
+
 typedef map <string, EventPublisher> lst_publishers_t;
 lst_publishers_t s_publishers;
 
-EventPublisher::EventPublisher(const string source) :
+EventPublisher::EventPublisher() :
     m_zmq_ctx(NULL), m_socket(NULL), m_sequence(0), m_init(false)
 {
 }
 
-int EventPublisher::init(const string event_source, int block_ms)
+int EventPublisher::init(const string event_source)
 {
     m_zmq_ctx = zmq_ctx_new();
     m_socket = zmq_socket (m_zmq_ctx, ZMQ_PUB);
@@ -26,7 +39,11 @@ int EventPublisher::init(const string event_source, int block_ms)
     // Any message published before connection establishment is dropped.
     //
     event_service m_event_svc;
-    rc = m_event_svc.init_client(block_ms);
+    /*
+     * Event service could be down. So have a timeout.
+     * 
+     */
+    rc = m_event_svc.init_client(m_zmq_ctx, EVENTS_SERVICE_TIMEOUT_MILLISECS);
     RET_ON_ERR (rc == 0, "Failed to init event service");
 
     rc = m_event_svc.echo_send("hello");
@@ -57,7 +74,7 @@ EventPublisher::event_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 block_ms milliseconds
+         * NOTE: This call atmost blocks for EVENTS_SERVICE_TIMEOUT_MILLISECS
          * as provided in publisher init.
          */
         m_event_service.echo_receive(s);
@@ -82,11 +99,11 @@ EventPublisher::event_publish(const string tag, const event_params_t *params)
     event_data[EVENT_RUNTIME_ID] = m_runtime_id;
     event_data[EVENT_SEQUENCE] = seq_to_str(m_sequence);
 
-    return zmq_message_send(m_event_source, event_data);
+    return zmq_message_send(m_socket, m_event_source, event_data);
 }
 
 event_handle_t
-events_init_publisher(const string event_source, int block_millisecs)
+events_init_publisher(const string event_source)
 {
     event_handle_t ret = NULL;
     lst_publishers_t::iterator it = s_publishers.find(event_source);
@@ -97,7 +114,7 @@ events_init_publisher(const string event_source, int block_millisecs)
     else {
         EventPublisher *p =  new EventPublisher();
 
-        int rc = p->init(event_source, block_millisecs);
+        int rc = p->init(event_source);
 
         if (rc != 0) {
             delete p;
@@ -132,10 +149,10 @@ event_publish(event_handle_t handle, const string tag, const event_params_t *par
 {
     for(it=s_publishers.begin(); it != s_publishers.end(); ++it) {
         if (it->second == handle) {
-            it->second->event_publish(tag, params);
-            break;
+            return it->second->event_publish(tag, params);
         }
     }
+    return -1;
 }
 
 
@@ -179,7 +196,9 @@ EventSubscriber::~EventSubscriber()
                 break;
             }
             events.push_back(evt_str);
-            if(chrono::steady_clock::now() - start > chrono::seconds(2)) 
+            chrono::steady_clock::timepoint now = chrono::steady_clock::now();
+            if (chrono::duration_cast<std::chrono::milliseconds>(now - start) >
+                    CACHE_DRAIN_IN_MILLISECS)
                 break;
         }
 
@@ -223,12 +242,9 @@ EventSubscriber::init(bool use_cache, const event_subscribe_sources_t *subs_sour
     }
 
     if (use_cache) {
-        rc = m_event_svc.init_client(m_zmq_ctx);
+        rc = m_event_svc.init_client(m_zmq_ctx, EVENTS_SERVICE_TIMEOUT_MILLISECS);
         RET_ON_ERR(rc == 0, "Fails to init the service");
 
-        /* Shadow the cache SUBS connect request, as it is async */
-        m_event_svc.send_recv(EVENT_ECHO);
-
         if (m_event_svc.cache_stop() == 0) {
             // Stopped an active cache 
             m_cache_read = true;
@@ -261,10 +277,12 @@ EventSubscriber::prune_track()
 
 
 int
-EventSubscriber::event_receive(event_str_t &event, int &missed_cnt)
+EventSubscriber::event_receive(string &&key, event_params_t &params, int &missed_cnt)
 {
-    event.clear();
-    while (event.empty()) {
+    key.clear();
+    event_params_t().swap(params);
+
+    while (key.empty()) {
         internal_event_t event_data;
         int rc = 0; 
 
@@ -303,7 +321,16 @@ EventSubscriber::event_receive(event_str_t &event, int &missed_cnt)
             }
         }
         if (missed_cnt >= 0) {
-            event = event_data[EVENT_STR_DATA];
+            map_str_str_t ev;
+
+            deserialize(event_data[EVENT_STR_DATA], ev);
+            RET_ON_ERR(ev.size() == 1, "Expect string (%s) deserialize to one key",
+                    event_data[EVENT_STR_DATA].c_str());
+            
+            key = ev.begin()->first;
+
+            deserialize(ev.begin()->second, params);
+
         }
     }
 out:
@@ -345,10 +372,11 @@ events_deinit_subscriber(event_handle_t &handle)
 
 
 int
-event_receive(event_handle_t handle, event_str_t &event, int &missed_cnt)
+event_receive(event_handle_t handle, string &key,
+        event_params_t &params, int &missed_cnt)
 {
     if ((handle == s_subscriber) && (s_subscriber != NULL)) {
-        return s_subscriber->event_receive(event, missed_cnt);
+        return s_subscriber->event_receive(key, params, missed_cnt);
     }
     return -1;
 }

common/events.h

@@ -1,3 +1,6 @@
+#ifndef _EVENTS_H
+#define _EVENTS_H
+
 /*
  * Events library 
  *
@@ -30,20 +33,12 @@ typedef void* event_handle_t;
  *      returned by this call is tagged with this source, transparently. The receiver
  *      could subscribe with this source as filter.
  *
- *  block_millisecs -
- *      Block either until publisher connects successfully or timeout
- *      whichever earlier.
- *      0   - No blocling.
- *      -1  - Block until connected.
- *      N   - Count in milli seconds to wait.
- *      NOTE: The connection is to eventd service, which could be down.
  * Return 
  *  Non NULL handle
  *  NULL on failure
  */
 
-event_handle_t events_init_publisher(std::string event_source,
-        int block_millisecs);
+event_handle_t events_init_publisher(std::string event_source);
 
 /*
  * De-init/free the publisher
@@ -65,7 +60,7 @@ typedef std::map<std::string, std::string> event_params_t;
 /*
  * timestamp param name
  */
-const string event_ts("timestamp");
+const std::string event_ts("timestamp");
 
 /*
  * Publish an event
@@ -78,7 +73,7 @@ const string event_ts("timestamp");
  *  the running instance of a publisher and other a running sequence
  *  starting from 0, which is local to this runtime-id.
  *
- *  The receiver API keep next last received number for each runtime id
+ *  The receiver API keep last received number for each runtime id
  *  and use this info to compute missed event count upon next event.
  *
  * input:
@@ -97,8 +92,11 @@ const string event_ts("timestamp");
  *      The timestamp should be per rfc3339
  *      e.g. "2022-08-17T02:39:21.286611Z"
  *
+ * return:
+ *  0   -   On success
+ *  -1  -   On failure.
  */
-void event_publish(event_handle_t handle, const std::string event_tag,
+int event_publish(event_handle_t handle, const std::string event_tag,
         const event_params_t *params=NULL);
 
 
@@ -112,8 +110,8 @@ typedef std::vector<std::string> event_subscribe_sources_t;
  * Input:
  *  use_cache
  *      When set to true, it will make use of the cache service transparently.
- *      The cache service caches events during session down time (last deinit to this
- *      init call).
+ *      The cache service caches events during session down time. The deinit
+ *      start the caching and init call stops the caching.
  *
  *  lst_subscribe_sources_t
  *      List of subscription sources of interest.
@@ -138,26 +136,37 @@ event_handle_t events_init_subscriber(bool use_cache=false,
  */
 void events_deinit_subscriber(event_handle_t &handle);
 
-/*
- * Received event as JSON string as 
- *  < YANG path of schema >: {
- *      event_params_t
- *  }
- */
-typedef std::string event_str_t;
-
 /*
  * Receive an event.
  * A blocking call.
  *
  *  This API maintains an expected sequence number and use the received
- *  sequence in event to compute missed events count.
+ *  sequence in event to compute missed events count. The missed count
+ *  set of events missed from this sender.
+ *
+ *  Received event:
+ *      It is a form of JSON struct, with a single key and
+ *      params as value. The key is <YANG schema module name>:<YANG schema tag
+ *      name> and params is as per schema description for that event.
+ *     
+ *      e.g.
+ *          { "sonic-events-bgp:bgp-state": {
+ *              "ip": "100.126.188.90",
+ *              "status": "down",
+ *              "timestamp": "2022-08-17T02:39:21.286611Z"
+ *              }
+ *          }
  *
  * input:
  *  handle - As obtained from events_init_subscriber
  *
  * output:
- *  event - Received event.
+ *  key : 
+ *      YANG path as <event source module name>:<event tag/container name
+ *      within the module that describes the event>
+ *
+ *  params:
+ *      Parms associated.
  *
  *  missed_cnt:
  *      Count of missed events from this sender, before this event. Sum of
@@ -168,5 +177,21 @@ typedef std::string event_str_t;
  * -1 - On failure. The handle is not valid.
  *
  */
-int event_receive(event_handle_t handle, event_str_t &event, int &missed_cnt);
+int event_receive(event_handle_t handle, std::string &key,
+        event_params_t &params, int &missed_cnt);
+
+
+/*
+ *  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
 
+#endif /* !_EVENTS_H */