init

opencv-motion-detect/Makefile

@@ -7,7 +7,7 @@ LIBOPENCV = `pkg-config opencv --cflags --libs`
 LIBFFMPEG = `pkg-config libavformat libavutil libavcodec libswscale --cflags --libs`
 LIBS=-lpthread
 SQLITE=vendor/sqlite/sqlite3.c
-HEADERS=-Iinc
+HEADERS=-Iinc -Ivendor/include
 
 .PHONY: libzmq
 all: evmgr evpuller evpusher evslicer evmlmotion

opencv-motion-detect/evmgr.cpp

@@ -13,12 +13,13 @@
 namespace fs = std::filesystem;
 #endif
 
-#include "vendor/include/zmq.h"
+#include "zmqhelper.hpp"
 #include "tinythread.hpp"
 #include "common.hpp"
 #include "database.h"
 
 using namespace std;
+using namespace zmqhelper;
 
 /**
  *  functions:
@@ -83,31 +84,28 @@ private:
         return ret;
     }
 
-    void handleMsg(vector<string> body) {
+    int handleMsg(vector<vector<uint8_t> > &body) {
+        int ret = 0;
         zmq_msg_t msg;
-        // dump
-        string dump;
-        for(auto &i: body) {
-             dump += i + ";";
-        }
-        cout <<endl;
 
+        // ID_SENDER, ID_TARGET, MSG
         if(body.size() != 3) {
-            spdlog::error("evmgr {} illegal message received: {}", devSn, dump);
+            spdlog::error("evmgr {} illegal message received, frame num: {}", devSn, body.size());
+            return -1;
         }
 
-        if(body[0] != devSn) {
-            for(int i =0; i < 3; i++) {
-                spdlog::info("evmgr {}, msg body is {}; {}; {}; forwarding...", devSn, body[0], body[1], body[2]);
-                zmq_msg_init(&msg);
-                zmq_msg_init_data(&msg, (void*)body[i].c_str(), body[i].size(), NULL, NULL);
-                mqErrorMsg("evmgr", devSn, "failed to send zmq msg", zmq_send_const(pRouter, zmq_msg_data(&msg), body[i].size(), i ==2?0:ZMQ_SNDMORE));
-                zmq_msg_close(&msg);
+        // if need forward
+        if(memcmp((void*)(body[1].data()), devSn.data(), body[1].size()) != 0) {
+            ret = z_send_multiple(pRouter, body);
+            if(ret < 0) {
+                spdlog::error("evmgr {} failed to send multiple: {}", devSn, zmq_strerror(zmq_errno()));
             }
         }else{
             // TODO: report msg
-            spdlog::info("evmgr {} subsystem report msg received: {}; {}; {}", devSn, body[0], body[1], body[2]);
+            spdlog::info("evmgr {} subsystem report msg received: {}; {}; {}", devSn, zmqhelper::body2str(body[0]), zmqhelper::body2str(body[1]), zmqhelper::body2str(body[2]));
         }
+
+        return ret;
     }
 
 protected:
@@ -120,25 +118,11 @@ protected:
                 bStopSig = true;
                 break;
             }
-            vector<string>body;
-            int64_t more = 1;
-            // business logic
-            int cnt = 0;
-            char *tmp;
-
-            while(more > 0) {
-                cnt++;
-                mqErrorMsg("evmgr", devSn, "failed to init zmq msg", zmq_msg_init(&msg));
-                ret = mqErrorMsg("evmgr", devSn, "failed to recv zmq msg", zmq_recvmsg(pRouter, &msg, 0));
-                tmp = new char[ret+1];
-                memcpy(tmp, zmq_msg_data(&msg), ret);
-                tmp[ret] = 0;
-                body.push_back(string(tmp));
-                delete tmp;
-                zmq_msg_close(&msg);
-                spdlog::debug("evmgr {} received[{}]: {} ", devSn, cnt, body.back());
-                size_t more_size = sizeof (more);
-                mqErrorMsg("evmgr", devSn, "failed to get zmq sockopt", zmq_getsockopt(pRouter, ZMQ_RCVMORE, &more, &more_size));
+            vector<vector<uint8_t> >body;
+            ret = z_recv_multiple(pRouter, body);
+            if(ret < 0) {
+                spdlog::error("evmgr {} failed to receive multiple msg: {}", devSn, zmq_strerror(zmq_errno()));
+                continue;   
             }
             // full proto msg received.
             handleMsg(body);

opencv-motion-detect/evpuller.cpp

@@ -13,12 +13,13 @@
 namespace fs = std::filesystem;
 #endif
 
-#include "vendor/include/zmq.h"
+#include <zmqhelper.hpp>
 #include "tinythread.hpp"
 #include "common.hpp"
 #include "database.h"
 
 using namespace std;
+using namespace zmqhelper;
 
 int mqErrorMsg(string cls, string devSn, int iid, string extraInfo, int ret)
 {
@@ -38,42 +39,31 @@ private:
     const char * bytes;
     int len;
     void *pDealer=NULL;
-    // void *pRepCtx = NULL; // for packets REP
-    // void *pRep = NULL;
-
-    int init()
-    {
-        // int ret = 0;
-        // pRepCtx = zmq_ctx_new();
-        // pRep = zmq_socket(pRepCtx, ZMQ_REP);
-        // ret = zmq_bind(pRep, urlRep.c_str());
-        // if(ret < 0) {
-        //     spdlog::error("failed to bind rep: {}, {}", zmq_strerror(ret), urlRep.c_str());
-        //     this_thread::sleep_for(chrono::seconds(20));
-        //     return -1;
-        // }
-        return 0;
-    }
 protected:
     void run()
     {
+        int ret = 0;
         bool bStopSig = false;
 
         zmq_msg_t msg;
         zmq_msg_t msg1;
-        int ret =zmq_msg_init(&msg);
-        zmq_msg_init_data(&msg, (void*)bytes, len, NULL, NULL);
+        ret = zmq_msg_init(&msg);
+        ret += zmq_msg_init_data(&msg, (void*)bytes, len, NULL, NULL);
+        if(ret < 0) {
+            spdlog::error("evpuller {} {} failed to init msg: {}", devSn, iid, zmq_strerror(zmq_errno()));
+            return;
+        }
         // declare ready to router
-        vector<string>body;
-        body.push_back(mgrSn);
-        body.push_back("hello");
-        int cnt = 0;
-        for(auto &i:body) {
-            zmq_msg_init(&msg1);
-            zmq_msg_init_data(&msg1, (void*)i.c_str(), i.size(), NULL, NULL);
-            mqErrorMsg("evpuller", devSn,iid, "failed to send zmq msg", zmq_send_const(pDealer, zmq_msg_data(&msg1), i.size(), cnt==(body.size()-1)?0:ZMQ_SNDMORE));
-            zmq_msg_close(&msg1);
-            cnt++;
+        vector<vector<uint8_t> >body;
+        // since identity is auto set
+        body.push_back(str2body(mgrSn));
+        body.push_back(str2body("hello"));
+
+        ret = z_send_multiple(pDealer, body);
+        if(ret < 0) {
+            spdlog::error("evpuller {} {} failed to send multiple: {}", devSn, iid, zmq_strerror(zmq_errno()));
+            //TODO:
+            return;
         }
 
         while (true) {
@@ -82,7 +72,7 @@ protected:
                 break;
             }
             spdlog::info("evpuller reqSrv {} {} waiting for req", devSn, iid);
-            int ret =zmq_msg_init(&msg1);
+            ret =zmq_msg_init(&msg1);
             ret = zmq_recvmsg(pDealer, &msg1, 0);
             if(ret < 0) {
                 spdlog::error("failed to recv zmq msg: {}", zmq_strerror(ret));
@@ -97,19 +87,11 @@ public:
     RepSrv() = delete;
     RepSrv(RepSrv &) = delete;
     RepSrv(RepSrv&&) = delete;
-    RepSrv(string mgrSn, string devSn, int iid, const char* formatBytes, int len, void *pDealer):mgrSn(mgrSn),devSn(devSn), iid(iid), bytes(formatBytes), len(len), pDealer(pDealer)
-    {
-        init();
-    };
-    ~RepSrv()
-    {
-        // if(pRep != NULL) {
-        //     zmq_close(pRep);
-        // }
-        // if(pRepCtx != NULL) {
-        //     zmq_ctx_destroy(pRepCtx);
-        // }
-    };
+    RepSrv(string mgrSn, string devSn, int iid, const char* formatBytes,
+           int len, void *pDealer):mgrSn(mgrSn),devSn(devSn), iid(iid), bytes(formatBytes),
+        len(len), pDealer(pDealer) {};
+
+    ~RepSrv() {};
 };
 
 class EvPuller: public TinyThread {

opencv-motion-detect/inc/common.hpp

@@ -370,4 +370,5 @@ json registry(const char *sn, const char *scn, int iid)
 }
 } // namespace cloudutils
 
+
 #endif