getSn confliction

opencv-motion-detect/evdaemon.cpp

@@ -283,7 +283,7 @@ private:
                     if(this->peerData["contConn"].count(k) != 0 && this->peerData["contConn"][k] > 4) { // 4 times
                         json meta;
                         json data;
-                        string msg = fmt::format("evdaemon {} detects module {} is restarting frequently, slow down for 10s", this->devSn, k);
+                        string msg = fmt::format("evdaemon {} detects module {} is restarting frequently, should tabke action", this->devSn, k);
                         spdlog::error(msg);
                         data["msg"] = msg;
                         data["modId"] = k;
@@ -295,7 +295,7 @@ private:
                         meta["type"] = EV_MSG_META_TYPE_REPORT;
                         meta["value"] = EV_MSG_META_VALUE_REPORT_LEVEL_ERROR;
                         z_send(pDealer, "evcloudsvc", meta.dump(), data.dump());
-                        this_thread::sleep_for(chrono::seconds(10));
+                        // this_thread::sleep_for(chrono::seconds(10));
                     }
                     this->peerData["status"][k] = 0;
                     ret = zmqhelper::forkSubsystem(devSn, k, portRouter, pid);
@@ -771,6 +771,8 @@ private:
                             if(v.size() == 1) {
                                 if(data["metaValue"] == EV_MSG_META_VALUE_CMD_REVESETUN) {
                                     manageReverseTun(true, data["data"]);
+                                }else if(data["metaValue"] == EV_MSG_META_VALUE_CMD_UPDATE){
+                                    //
                                 }
                                 else {
                                     spdlog::info("evdaemon {} received msg {} from cloud to itself. but has no implementation for", devSn, data.dump());

opencv-motion-detect/evslicer.cpp

@@ -72,7 +72,7 @@ private:
         int ret = 0;
         string peerId, meta;
         json data;
-        string msg;
+        string msg, metaType, metaValue;
         for(auto &b:v) {
             msg +=body2str(b) + ";";
         }
@@ -81,23 +81,35 @@ private:
         if(v.size() == 3) {
             try {
                 peerId = body2str(v[0]);
-                meta = json::parse(body2str(v[1]))["type"];
-                if(meta == EV_MSG_META_AVFORMATCTX) {
-                    lock_guard<mutex> lock(this->mutMsg);
-                    if(pAVFormatInput == nullptr) {
-                        pAVFormatInput = (AVFormatContext *)malloc(sizeof(AVFormatContext));
-                        AVFormatCtxSerializer::decode((char *)(v[2].data()), v[2].size(), pAVFormatInput);
-                        gotFormat = true;
-                        cvMsg.notify_one();
-                        spdlog::info("evslicer {} got avformat from {}", selfId, peerId);
-                    }
-                    else {
-                        spdlog::warn("evslicer {} received avformatctx msg from {}, but already proceessed before, restarting", selfId, peerId);
-                        spdlog::error("evslicer {} restart since reinit", selfId);
-                        exit(0);
+                json meta = json::parse(body2str(v[1]));
+                metaType = meta["type"];
+                if(meta.count("value") != 0) {
+                    metaValue = meta["value"];
+                }
+
+                // msg from cluster mgr
+                string clusterMgrId = this->mgrSn + ":evmgr:0";
+                if(peerId == clusterMgrId) {
+                    //
+                }
+                else if(peerId == pullerGid) {
+                    if(metaType == EV_MSG_META_AVFORMATCTX) {
+                        lock_guard<mutex> lock(this->mutMsg);
+                        if(pAVFormatInput == nullptr) {
+                            pAVFormatInput = (AVFormatContext *)malloc(sizeof(AVFormatContext));
+                            AVFormatCtxSerializer::decode((char *)(v[2].data()), v[2].size(), pAVFormatInput);
+                            gotFormat = true;
+                            cvMsg.notify_one();
+                            spdlog::info("evslicer {} got avformat from {}", selfId, peerId);
+                        }
+                        else {
+                            spdlog::warn("evslicer {} received avformatctx msg from {}, but already proceessed before, restarting", selfId, peerId);
+                            spdlog::error("evslicer {} restart since reinit", selfId);
+                            exit(0);
+                        }
                     }
                 }
-                else if(meta == EV_MSG_META_EVENT) {
+                else if(metaType == EV_MSG_META_EVENT) {
                     data = json::parse(body2str(v[2]));
 
                     /// evslicer has two msg interfaces to subsystems on edge side
@@ -107,7 +119,7 @@ private:
                         spdlog::info("evslicer {} received invalid msg from {}: {}", selfId, peerId, msg);
                     }
                     else {
-                        spdlog::info("evslicer {} received msg from {}, type = {}, data = {}", selfId, peerId, meta, data.dump());
+                        spdlog::info("evslicer {} received msg from {}, type = {}, data = {}", selfId, peerId, metaType, data.dump());
                         if(data["type"] == "event") {
                             lock_guard<mutex> lock(this->mutEvent);
                             eventQueue.push(data.dump());

opencv-motion-detect/inc/zmqhelper.cpp

@@ -171,33 +171,33 @@ int setupDealer(void **ctx, void **s, string addr, string ident, int timeout) {
 /// @return 0 success, otherwise failed.
 int recvConfigMsg(void *s, json &config, string addr, string ident){
     bool bConfigGot = false;
-    auto t = thread([&](){
-        this_thread::sleep_for(chrono::seconds(5));
-        if(bConfigGot == false){
-            spdlog::error("{} failed receive config from evdaemon", ident);
-            exit(1);
-        }
-    });
-    auto v = zmqhelper::z_recv_multiple(s);
-    if(v.size() != 3) {
+    // auto t = thread([&](){
+    //     this_thread::sleep_for(chrono::seconds(5));
+    //     if(bConfigGot == false){
+    //         spdlog::error("{} failed receive config from evdaemon", ident);
+    //         exit(1);
+    //     }
+    // });
+    while(!bConfigGot) {
+        auto v = zmqhelper::z_recv_multiple(s);
+        if(v.size() != 3) {
             spdlog::error("{} received invalid msg from evdaemon", ident);
-            exit(1);
-    }
+            continue;
+        }
 
-    spdlog::debug("{} configuration msg received: {} {} {}", ident, body2str(v[0]), body2str(v[1]), body2str(v[2]));
-    try{
-        string sMeta = json::parse(body2str(v[1]))["type"];
-        if(sMeta != EV_MSG_META_CONFIG) {
-            throw StrException("meta type is:" + sMeta + ", but expecting " + EV_MSG_META_CONFIG);
+        spdlog::debug("{} configuration msg received: {} {} {}", ident, body2str(v[0]), body2str(v[1]), body2str(v[2]));
+        try{
+            string sMeta = json::parse(body2str(v[1]))["type"];
+            if(sMeta != EV_MSG_META_CONFIG) {
+                throw StrException("meta type is:" + sMeta + ", but expecting " + EV_MSG_META_CONFIG);
+            }
+            config = json::parse(body2str(v[2]));
+            bConfigGot = true;
+        }catch(exception &e) {
+            spdlog::error("{} invalid config msg from daemon {}, {}", ident, addr, e.what());
         }
-        config = json::parse(body2str(v[2]));
-        bConfigGot = true;
-    }catch(exception &e) {
-        spdlog::error("{} invalid config msg from daemon {}, {}", ident, addr, e.what());
     }
 
-    t.join();
-
     return bConfigGot? 0: -1;
 }