big refacting of communitation architect

opencv-motion-detect/evdaemon.cpp

@@ -37,6 +37,7 @@ class EvDaemon{
     thread thRouter;
     json peerData;
     bool bReload = true;
+    bool bBootstrap = true;
     // peerData["status"];
     // peerData["pids"];
     // peerData["config"];
@@ -91,6 +92,7 @@ class EvDaemon{
                     this->peerData["config"][peerId] = v;
                     if(this->peerData["status"].count(peerId) == 0||this->peerData["status"][peerId] == 0) {
                         this->peerData["status"][peerId] = 0;
+                        if(bBootstrap) {
                             ret = zmqhelper::forkSubsystem(devSn, peerId, portRouter, pid);
                             if(ret != 0) {
                                 spdlog::error("evdaemon {} failed to fork subsystem: {}", devSn, peerId);
@@ -99,6 +101,7 @@ class EvDaemon{
                             }
                             this->peerData["pids"][peerId] = pid;
                             spdlog::info("evdaemon {} created subsystem {}", devSn, peerId); 
+                        }  
                     }else{
                         // TODO:
                     }         
@@ -127,6 +130,7 @@ class EvDaemon{
 
                                 if(this->peerData["status"].count(peerId) == 0||this->peerData["status"][peerId] == 0) {
                                     this->peerData["status"][peerId] = 0;
+                                    if(bBootstrap) {
                                         ret = zmqhelper::forkSubsystem(devSn, peerId, portRouter, pid);
                                         if(ret != 0) {
                                             spdlog::error("evdaemon {} failed to fork subsystem: {}", devSn, peerId);
@@ -135,6 +139,7 @@ class EvDaemon{
                                         }
                                         this->peerData["pids"][peerId] = pid;
                                         spdlog::info("evdaemon {} created subsystem {}", devSn, peerId); 
+                                    }
                                 }else{
                                     // TODO:
                                 }
@@ -165,9 +170,11 @@ class EvDaemon{
             if(this->peerData["config"].count(k) != 0){
                 this->peerData["config"].erase(k);
             }
-            
+            if(this->peerData["pids"].count(k) != 0) {
                 this->peerData["pids"].erase(k);
             }
+            
+        }
     }
 
     void setupSubSystems() {
@@ -271,8 +278,13 @@ class EvDaemon{
             }
             else {
                 peerData["status"][selfId] = 0;
+                if(peerData["pids"].count(selfId) != 0) {
                     peerData["pids"].erase(selfId);
+                }
+                if(peerData["config"].count(selfId) != 0) {
                     peerData["config"].erase(selfId);
+                }
+                
                 spdlog::warn("evdaemon {} peer disconnected: {}", devSn, selfId);
                 // restart this module
                 startSubModule(selfId);
@@ -456,16 +468,21 @@ class EvDaemon{
 
     EvDaemon(){
         int ret = 0;
+        char* strEnv = getenv("BOOTSTRAP");
+        if(strEnv != NULL && memcmp(strEnv, "false", 5) == 0) {
+            bBootstrap = false;
+        }
+
         // http port
-        char* strPort = getenv("DAEMON_PORT");
-        if(strPort != NULL) {
-            port = stoi(strPort);
+        strEnv = getenv("DAEMON_PORT");
+        if(strEnv != NULL) {
+            port = stoi(strEnv);
         }
 
         // zmq router port
-        strPort = getenv("ROUTER_PORT");
-        if(strPort != NULL) {
-            portRouter = stoi(strPort);
+        strEnv = getenv("ROUTER_PORT");
+        if(strEnv != NULL) {
+            portRouter = stoi(strEnv);
         }
 
         string addr = string("tcp://*:") + to_string(portRouter);

opencv-motion-detect/evmgr.cpp

@@ -47,6 +47,7 @@ private:
     queue<string> eventQue;
     mutex eventQLock;
     time_t tsLastBoot,  tsUpdateTime;
+    string drport = "5549";
 
     //
     void init()
@@ -276,16 +277,12 @@ public:
     EvMgr& operator=(EvMgr &&) = delete;
     EvMgr()
     {
-        string drport;
         const char *strEnv = getenv("DR_PORT");
         if(strEnv != NULL) {
             drport = strEnv;
-        }else{
-            spdlog::error("evmgr failed to start. no DR_PORT set");
-            exit(1);
         }
 
-        strEnv = getenv("GID");
+        strEnv = getenv("PEERID");
         if(strEnv != NULL) {
             ident = strEnv;
             auto v = strutils::split(ident, ':');

opencv-motion-detect/evmlmotion.cpp

@@ -84,6 +84,7 @@ private:
     json config;
     thread thPing;
     thread thEvent;
+    string drport = "5549";
     //
 
     int init()
@@ -634,16 +635,12 @@ public:
     EvMLMotion(queue<string> *queue)
     {
         evtQueue = queue;
-        string drport;
         const char *strEnv = getenv("DR_PORT");
         if(strEnv != NULL) {
             drport = strEnv;
-        }else{
-            spdlog::error("evmlmotion failed to start. no DR_PORT set");
-            exit(1);
         }
 
-        strEnv = getenv("GID");
+        strEnv = getenv("PEERID");
         if(strEnv != NULL) {
             selfId = strEnv;
             auto v = strutils::split(selfId, ':');

opencv-motion-detect/evpuller.cpp

@@ -43,23 +43,6 @@ private:
     const char * bytes;
     int len;
     void *pDealer=NULL;
-    thread thPing;
-
-    int ping()
-    {
-        int ret = 0;
-        vector<vector<uint8_t> >body;
-        // since identity is auto set
-        body.push_back(str2body(mgrSn + ":0:0"));
-        body.push_back(str2body(EV_MSG_META_PING));
-        body.push_back(str2body(MSG_HELLO));
-
-        ret = z_send_multiple(pDealer, body);
-        if(ret < 0) {
-            spdlog::error("evpuller {} failed to send multiple: {}", selfId, zmq_strerror(zmq_errno()));
-        }
-        return ret;
-    }
 
     int handleMsg(vector<vector<uint8_t> > v)
     {
@@ -99,18 +82,6 @@ protected:
     {
         int ret = 0;
         bool bStopSig = false;
-        // declare ready to router
-        ping();
-
-        // TODO: don't need this anymore, since I've used the draft feature of ZOUTER_NOTIFICATION instead
-        // thPing = thread([&,this]() {
-        //     while(true) {
-        //         this_thread::sleep_for(chrono::seconds(EV_HEARTBEAT_SECONDS-2));
-        //         ping();
-        //     }
-        // });
-
-        // thPing.detach();
 
         // init response msg
         while (true) {
@@ -155,6 +126,19 @@ private:
     int *streamList = NULL, numStreams = 0, iid;
     time_t tsLastBoot, tsUpdateTime;
     json config;
+    string drport = "5549";
+
+    int ping()
+    {
+        int ret = 0;
+        vector<vector<uint8_t> >body = {str2body(mgrSn + ":0:0"), str2body(EV_MSG_META_PING), str2body(MSG_HELLO)};
+
+        ret = z_send_multiple(pDealer, body);
+        if(ret < 0) {
+            spdlog::error("evpuller {} failed to send multiple: {}", selfId, zmq_strerror(zmq_errno()));
+        }
+        return ret;
+    }
 
     int init()
     {
@@ -242,6 +226,8 @@ private:
                 spdlog::error("evpuller {} failed to connect to router {}", selfId, urlDealer);
                 exit(1);
             }
+
+            ping();
         }
         catch(exception &e) {
             this_thread::sleep_for(chrono::seconds(3));
@@ -356,16 +342,12 @@ protected:
 public:
     EvPuller()
     {
-        string drport;
         const char *strEnv = getenv("DR_PORT");
         if(strEnv != NULL) {
             drport = strEnv;
-        }else{
-            spdlog::error("evpusher failed to start. no DR_PORT set");
-            exit(1);
         }
 
-        strEnv = getenv("GID");
+        strEnv = getenv("PEERID");
         if(strEnv != NULL) {
             selfId = strEnv;
             auto v = strutils::split(selfId, ':');

opencv-motion-detect/evpusher.cpp

@@ -48,6 +48,7 @@ private:
     time_t tsLastBoot, tsUpdateTime;
     json config;
     thread thPing;
+    string drport = "5549";
 
     int init()
     {
@@ -403,16 +404,12 @@ protected:
 public:
     EvPusher()
     {
-        string drport;
         const char *strEnv = getenv("DR_PORT");
         if(strEnv != NULL) {
             drport = strEnv;
-        }else{
-            spdlog::error("evpusher failed to start. no DR_PORT set");
-            exit(1);
         }
 
-        strEnv = getenv("GID");
+        strEnv = getenv("PEERID");
         if(strEnv != NULL) {
             selfId = strEnv;
             auto v = strutils::split(selfId, ':');

opencv-motion-detect/evslicer.cpp

@@ -56,6 +56,7 @@ private:
     time_t tsLastBoot, tsUpdateTime;
     json config;
     thread thPing;
+    string drport = "5549";
 
     int init()
     {
@@ -440,16 +441,12 @@ protected:
 public:
     EvSlicer()
     {
-        string drport;
         const char *strEnv = getenv("DR_PORT");
         if(strEnv != NULL) {
             drport = strEnv;
-        }else{
-            spdlog::error("evslicer failed to start. no DR_PORT set");
-            exit(1);
         }
 
-        strEnv = getenv("GID");
+        strEnv = getenv("PEERID");
         if(strEnv != NULL) {
             selfId = strEnv;
             auto v = strutils::split(selfId, ':');

opencv-motion-detect/inc/zmqhelper.hpp

@@ -200,7 +200,7 @@ int forkSubsystem(string devSn, string peerId, int drPort, pid_t &pid){
         spdlog::error("evdamon {} failed to fork subsytem - evmgr", devSn);
         return -1;
     }else if(pid == 0) {
-        ret += setenv("GID", peerId.c_str(), 1);
+        ret += setenv("PEERID", peerId.c_str(), 1);
         ret += setenv("DR_PORT", to_string(drPort).c_str(), 1);
         if(ret < 0) {
             spdlog::error("evdaemon {} failed to set env", devSn);