big refacting of communitation architect

opencv-motion-detect/evcloudsvc.cpp

@@ -242,7 +242,6 @@ private:
     int handleMsg(vector<vector<uint8_t> > &body)
     {
         int ret = 0;
-        zmq_msg_t msg;
         // ID_SENDER, ID_TARGET, meta ,MSG
         string selfId, peerId, meta;
         if(body.size() == 2 && body[1].size() == 0) {

opencv-motion-detect/evmgr.cpp

@@ -251,8 +251,6 @@ protected:
     {
         bool bStopSig = false;
         int ret = 0;
-        zmq_msg_t msg;
-
         while (true) {
             if(checkStop() == true) {
                 bStopSig = true;

opencv-motion-detect/evmlmotion.cpp

@@ -462,7 +462,7 @@ private:
             }
         } //end for
 
-        spdlog::debug("evmlmotion {} contours {} area {}, thresh {} hasEvent {}", selfId, cnts.size(), cnts.size() == 0? 0:cv::contourArea(cnts[i]), detPara.area, hasEvent);
+        spdlog::debug("evmlmotion {} contours {} area {}, thresh {} hasEvent {}", selfId, cnts.size(), hasEvent? cv::contourArea(cnts[i]):0, detPara.area, hasEvent);
 
         // business logic for event
         auto dura = chrono::duration_cast<chrono::seconds>(evtStartTm - evtStartTmLast).count();
@@ -579,7 +579,7 @@ protected:
                         spdlog::error("evmlmotion {} failed to send event {} to {}: {}", this->selfId, evt, this->slicerGid, zmq_strerror(zmq_errno()));
                     }
                     else {
-                        spdlog::info("evmlmotion {} send event: {}", this->selfId, evt);
+                        spdlog::info("evmlmotion {} sent event: {}", this->selfId, evt);
                     }
                 }
                 else {

opencv-motion-detect/evslicer.cpp

@@ -57,9 +57,33 @@ private:
     int *streamList = nullptr;
     time_t tsLastBoot, tsUpdateTime;
     json config;
-    thread thPing;
+    thread thMsgProcessor;
     string drport = "5549";
 
+    int handleMsg(vector<vector<uint8_t> > v){
+        int ret = 0;
+        string peerId, meta;
+        json data;
+        if(v.size() == 3) {
+            try{
+                peerId = body2str(v[0]);
+                meta = json::parse(body2str(v[1]))["type"];
+                data = json::parse(body2str(v[2]));
+                spdlog::info("evslicer {} received msg from {}, type = {}, data = {}", selfId, meta, data.dump());
+            }catch(exception &e){
+                spdlog::error("evslicer {} failed to process msg:{}", body2str(v[1]), body2str(v[2]));
+            }  
+        }else{
+            string msg;
+            for(auto &b:v) {
+                msg +=body2str(b) + ";";
+            } 
+            spdlog::error("evslicer {} get invalid msg with size {}: {}", selfId, v.size(), msg);
+        }
+
+        return ret;
+    }
+
     int init()
     {
         int ret = 0;
@@ -179,6 +203,19 @@ private:
             exit(1);
         }
 
+        thMsgProcessor = thread([this](){
+            while(true) {
+                auto body = z_recv_multiple(pDealer,false);
+                if(body.size() == 0) {
+                    spdlog::error("evslicer {} failed to receive multiple msg: {}", selfId, zmq_strerror(zmq_errno()));
+                    continue;
+                }
+                // full proto msg received.
+                handleMsg(body);
+            }
+        });
+
+        thMsgProcessor.detach();
 
         return ret;
     }

opencv-motion-detect/inc/av_common.hpp

@@ -230,9 +230,11 @@ int encode(AVFormatContext *ctx, char **bytes)
     got += sizeof(wholeSize);
     memcpy((*bytes) + got, PS_MARK_E, strlen(PS_MARK_E));
     got += strlen(PS_MARK_E);
-
+    if(wholeSize != got){
+        spdlog::error("encode wholesize: {}, should be {}", got, wholeSize);
+    }
     assert(wholeSize == got);
-    spdlog::info("encode wholesize: {}", got);
+    
     return wholeSize;
 }