Switched to shared_ptr per review comments

Author: renukamanavalan

common/events.cpp

@@ -7,14 +7,58 @@
  *  duplicates helps reduce load.
  */
 
-typedef map <string, EventPublisher *> lst_publishers_t;
-lst_publishers_t s_publishers;
+lst_publishers_t EventPublisher::s_publishers;
+
+event_handle_t
+EventPublisher::get_publisher(const string event_source)
+{
+    event_handle_t ret = NULL;
+    lst_publishers_t::const_iterator itc = s_publishers.find(event_source);
+    if (itc != s_publishers.end()) {
+        // Pre-exists
+        ret = itc->second;
+    }
+    else {
+        EventPublisher_ptr_t p(n.get()ew EventPublisher());
+
+        int rc = p->init(event_source);
+
+        if (rc == 0) {
+            ret = p.get();
+            s_publishers[event_source] = p;
+        }
+    }
+    return ret;
+}
+
+void
+EventPublisher::drop_publisher(event_handle_t handle)
+{
+    lst_publishers_t::iterator it;
+
+    for(it=s_publishers.begin(); it != s_publishers.end(); ++it) {
+        if (it->second.get() == handle) {
+            s_publishers.erase(it);
+            break;
+        }
+    }
+}
 
 EventPublisher::EventPublisher() :
     m_zmq_ctx(NULL), m_socket(NULL), m_sequence(0)
 {
 }
 
+static string
+get_uuid()
+{
+    uuid_t id;
+    char uuid_str[UUID_STR_SIZE];
+    uuid_generate(id);
+    uuid_unparse(id, uuid_str);
+    return string(uuid_str);
+}
+
 int EventPublisher::init(const string event_source)
 {
     m_zmq_ctx = zmq_ctx_new();
@@ -43,13 +87,7 @@ int EventPublisher::init(const string event_source)
 
     m_event_source = event_source;
 
-    {
-    uuid_t id;
-    char uuid_str[UUID_STR_SIZE];
-    uuid_generate(id);
-    uuid_unparse(id, uuid_str);
-    m_runtime_id = string(uuid_str);
-    }
+    m_runtime_id = get_uuid();
 
     m_socket = sock;
 out:
@@ -144,40 +182,13 @@ EventPublisher::publish(const string tag, const event_params_t *params)
 event_handle_t
 events_init_publisher(const string event_source)
 {
-    event_handle_t ret = NULL;
-    lst_publishers_t::const_iterator itc = s_publishers.find(event_source);
-    if (itc != s_publishers.end()) {
-        // Pre-exists
-        ret = itc->second;
-    }
-    else {
-        EventPublisher *p =  new EventPublisher();
-
-        int rc = p->init(event_source);
-
-        if (rc != 0) {
-            delete p;
-        }
-        else {
-            ret = p;
-            s_publishers[event_source] = p;
-        }
-    }
-    return ret;
+    return EventPublisher::get_publisher(event_source);
 }
 
 void
 events_deinit_publisher(event_handle_t handle)
 {
-    lst_publishers_t::iterator it;
-
-    for(it=s_publishers.begin(); it != s_publishers.end(); ++it) {
-        if (it->second == handle) {
-            delete it->second;
-            s_publishers.erase(it);
-            break;
-        }
-    }
+    EventPublisher::drop_publisher(handle);
 }
 
 int
@@ -193,6 +204,34 @@ event_publish(event_handle_t handle, const string tag, const event_params_t *par
 }
 
 
+event_handle_t
+EventSubscriber::get_subscriber(bool use_cache, int recv_timeout,
+        const event_subscribe_sources_t *sources)
+{
+
+    if (s_subscriber == NULL) {
+        EventSubscriber_ptr_t sub(new EventSubscriber());
+
+        RET_ON_ERR(sub->init(use_cache, recv_timeout, sources) == 0,
+                "Failed to init subscriber");
+
+        s_subscriber = sub;
+    }
+out:
+    return s_subscriber.get();
+}
+
+
+void
+EventSubscriber::drop_subscriber(event_handle_t handle)
+{
+    if ((handle == s_subscriber) && (s_subscriber != NULL)) {
+        delete s_subscriber;
+        s_subscriber = NULL;
+    }
+}
+
+
 EventSubscriber::EventSubscriber() : m_zmq_ctx(NULL), m_socket(NULL),
     m_cache_read(false)
 {};
@@ -407,41 +446,21 @@ EventSubscriber::event_receive(string &key, event_params_t &params, int &missed_
 
 
 /* Expect only one subscriber per process */
-static EventSubscriber *s_subscriber = NULL;
+EventSubscriber *EventSubscriber::s_subscriber = NULL;
 
 event_handle_t
 events_init_subscriber(bool use_cache, int recv_timeout,
         const event_subscribe_sources_t *sources)
 {
-    EventSubscriber *sub = NULL;
-
-    if (s_subscriber == NULL) {
-        sub = new EventSubscriber();
-
-        RET_ON_ERR(sub->init(use_cache, recv_timeout, sources) == 0,
-                "Failed to init subscriber");
-
-        s_subscriber = sub;
-        sub = NULL;
-    }
-out:
-    if (sub != NULL) {
-        delete sub;
-    }
-    return s_subscriber;
+    return EventSubscriber::get_subscriber(use_cache, recv_timeout, sources);
 }
 
-
 void
 events_deinit_subscriber(event_handle_t handle)
 {
-    if ((handle == s_subscriber) && (s_subscriber != NULL)) {
-        delete s_subscriber;
-        s_subscriber = NULL;
-    }
+    EventSubscriber::drop_subscriber(handle);
 }
 
-
 event_receive_op_t
 event_receive(event_handle_t handle)
 {

common/events_pi.h

@@ -35,19 +35,29 @@ class events_base
  * There can be multiple sender processes/threads for a single source.
  *
  */
+class EventPublisher;
+typedef shared_ptr<EventPublisher> EventPublisher_ptr_t;
+typedef map <string, EventPublisher_ptr_t> lst_publishers_t;
+
 
 class EventPublisher : public events_base
 {
-    public:
-        EventPublisher();
+    static lst_publishers_t s_publishers;
 
+    public:
         virtual ~EventPublisher();
 
         int init(const string event_source);
 
         int publish(const string event_tag,
                 const event_params_t *params);
+
+        static event_handle_t get_publisher(const string event_source);
+        static void drop_publisher((event_handle_t handle);
+
     private:
+        EventPublisher();
+
 
         void *m_zmq_ctx;
         void *m_socket;
@@ -72,12 +82,14 @@ class EventPublisher : public events_base
  *  would have nearly no impact on others.
  *
  */
+class EventSubscriber;
+typedef shared_ptr<EventSubscriber> EventSubscriber_ptr_t;
 
 class EventSubscriber : public events_base
 {
+    static EventSubscriber_ptr_t s_subscriber;
+
     public:
-        EventSubscriber();
-        
         int init(bool use_cache=false,
                 int recv_timeout= -1,
                 const event_subscribe_sources_t *subs_sources=NULL);
@@ -86,7 +98,14 @@ class EventSubscriber : public events_base
 
         int event_receive(string &key, event_params_t &params, int &missed_cnt);
 
+        static event_handle_t get_subscriber(bool use_cache, int recv_timeout,
+                const event_subscribe_sources_t *sources);
+
+        static void drop_subscriber(event_handle_t handle);
+
     private:
+        EventSubscriber();
+        
         void *m_zmq_ctx;
         void *m_socket;