init

opencv-motion-detect/evmgr.cpp

@@ -94,8 +94,8 @@ private:
         }
         cout <<endl;
 
-        // ID_SENDER, ID_TARGET, MSG
-        if(body.size() != 3) {
+        // ID_SENDER, ID_TARGET, meta ,MSG
+        if(body.size() != 4) {
             spdlog::warn("evmgr {} dropped a message, since its size is incorrect: {}", devSn, body.size());
             return 0;
         }
@@ -107,6 +107,7 @@ private:
             v.push_back(body[1]);
             v.push_back(body[0]);
             v.push_back(body[2]);
+            v.push_back(body[3]);
             ret = z_send_multiple(pRouter, v);
             if(ret < 0) {
                 spdlog::error("evmgr {} failed to send multiple: {}", devSn, zmq_strerror(zmq_errno()));

opencv-motion-detect/evpuller.cpp

@@ -48,6 +48,7 @@ protected:
         vector<vector<uint8_t> >body;
         // since identity is auto set
         body.push_back(str2body(mgrSn + ":0:0"));
+        body.push_back(str2body("")); // blank meta
         body.push_back(str2body(MSG_HELLO));
 
         ret = z_send_multiple(pDealer, body);
@@ -58,8 +59,7 @@ protected:
         }
 
         // init response msg
-        vector<uint8_t> msgBody;
-        msgBody.insert(msgBody.end(), (uint8_t *)bytes, (uint8_t *)bytes+len);
+        auto msgBody = data2body(const_cast<char*>(bytes), len);
         while (true) {
             if(checkStop() == true) {
                 bStopSig = true;
@@ -67,38 +67,37 @@ protected:
             }
 
             spdlog::info("evpuller repSrv {} {} waiting for req", devSn, iid);
-            // proto: [sender_id] [body]
+            // proto: [sender_id] [meta] [body]
             auto v = z_recv_multiple(pDealer);
-            if(v.size() == 0) {
+            if(v.size() != 3) {
                 //TODO:
+                spdlog::error("evpuller {} {},  repSrv received invalid message: {}", devSn, iid, v.size());
+                continue;
             }
             cout << endl<<endl;
             for(auto&j:v) {
                     cout <<body2str(j) << "; ";
             }
             cout << endl;
-            if(ret < 0) {
-                spdlog::error("evpuller {} {},  repSrv failed to recv msg: {}, {}", devSn, iid, v.size(), zmq_strerror(zmq_errno()));
-                continue;
-            }else if(v.size() != 2) {
-                spdlog::error("evpuller {} {},  repSrv received invalid msg, size: {}", devSn, iid, v.size());
-                continue;
-            }else{
-                //
+            try{
+                // rep framectx
+                // TODO: verify sender id
+                auto meta = json::parse(body2str(v[1]));
+                if(meta["type"].get<string>() == EV_PACKET_TYPE_AVFORMATCTX) {
+                    vector<vector<uint8_t> > rep;
+                    rep.push_back(v[0]);
+                    rep.push_back(v[1]);
+                    rep.push_back(msgBody);
+                    ret = z_send_multiple(pDealer, rep);
+                    if(ret < 0) {
+                        spdlog::error("evpuller {} {} failed send rep to requester {}: {}", devSn, iid, body2str(v[0]), zmq_strerror(zmq_errno()));
+                    }
+                }else{
+                    spdlog::error("evpuller {} {} unknown meta from {}: {}", devSn, iid, body2str(v[0]), body2str(v[1]));
+                }
+            }catch(exception &e) {
+                spdlog::error("evpuller {} {} excpetion parse request from {}: {}", devSn, iid, body2str(v[0]), body2str(v[1]));
             }
-
-
-            // vector<uint8_t> v;
-
-            // ret =zmq_msg_init(&msg1);
-            // ret = zmq_recvmsg(pDealer, &msg1, 0);
-            // if(ret < 0) {
-            //     spdlog::error("failed to recv zmq msg: {}", zmq_strerror(ret));
-            //     continue;
-            // }
-            // zmq_msg_close(&msg1);
-            // spdlog::info("evpuller {} {} reveived req", devSn, iid);
-            // zmq_send_const(pDealer, zmq_msg_data(&data), len, 0);
         }
     }
 public:

opencv-motion-detect/evpusher.cpp

@@ -144,6 +144,7 @@ private:
         vector<vector<uint8_t> >body;
         // since identity is auto set
         body.push_back(str2body(mgrSn+":0:0"));
+        body.push_back(str2body("")); // blank meta
         body.push_back(str2body(MSG_HELLO));
 
         ret = z_send_multiple(pDealer, body);
@@ -169,24 +170,40 @@ private:
         spdlog::info("evpusher {} {} send hello to puller: {}", devSn, iid, pullerGid);
         vector<vector<uint8_t> > body;
         body.push_back(str2body(pullerGid));
+        json meta;
+        meta["type"] = EV_PACKET_TYPE_AVFORMATCTX;
+        body.push_back(str2body(meta.dump()));
         body.push_back(str2body(MSG_HELLO));
-        ret = z_send_multiple(pDealer, body);
-        if(ret < 0) {
-            spdlog::error("evpusher {} {}, failed to send hello to puller: {}", devSn, iid, zmq_strerror(zmq_errno()));
-            return ret;
-        }
-        spdlog::info("evpusher {} {} success send hello", devSn, iid);
-
-        // expect response with avformatctx
-        auto v = z_recv_multiple(pDealer);
-        if(v.size() != 3) {
-            spdlog::error("evpusher {} {}, failed to receive avformatctx: {},{}", devSn, iid, v.size(), zmq_strerror(zmq_errno()));
-            return ret;
+        bool gotFormat = false;
+        while(!gotFormat) {
+            ret = z_send_multiple(pDealer, body);
+            if(ret < 0) {
+                spdlog::error("evpusher {} {}, failed to send hello to puller: {}", devSn, iid, zmq_strerror(zmq_errno()));
+                continue;
+            }
+            spdlog::info("evpusher {} {} success send hello", devSn, iid);
+
+            // expect response with avformatctx
+            auto v = z_recv_multiple(pDealer);
+            if(v.size() != 3) {
+                spdlog::error("evpusher {} {}, received bad size zmq msg for avformatctx: {}", devSn, iid, v.size());
+            }else if(body2str(v[0]) != pullerGid) {
+                spdlog::error("evpusher {} {}, invalid sender for avformatctx: {}, should be: {}", devSn, iid, body2str(v[0]), pullerGid);
+            }else{
+                try{
+                    auto cmd = json::parse(body2str(v[1]));
+                    if(cmd["type"].get<string>() == EV_PACKET_TYPE_AVFORMATCTX){
+                        pAVFormatInput = (AVFormatContext *)malloc(sizeof(AVFormatContext));
+                        AVFormatCtxSerializer::decode((char *)(v[2].data()), v[2].size(), pAVFormatInput);
+                        gotFormat = true;
+                    }    
+                }catch(exception &e) {
+                    spdlog::error("evpusher {} {}, exception in parsing avformatctx packet: {}", devSn, iid, e.what());
+                }
+            }
         }
-
-        pAVFormatInput = (AVFormatContext *)malloc(sizeof(AVFormatContext));
-        AVFormatCtxSerializer::decode((char *)v.back().data(), ret, pAVFormatInput);
         
+        //
         ret = avformat_alloc_output_context2(&pAVFormatRemux, NULL, "rtsp", urlOut.c_str());
         if (ret < 0) {
             spdlog::error("evpusher {} {} failed create avformatcontext for output: %s", devSn, iid, av_err2str(ret));

opencv-motion-detect/inc/common.hpp

@@ -230,7 +230,7 @@ int decode(char *bytes, int len, AVFormatContext *pCtx)
     memcpy(&ret, bytes +len -strlen(PS_MARK_E) - sizeof(ret), sizeof(ret));
     if ((memcmp(PS_MARK_S, bytes + got, strlen(PS_MARK_S)) != 0 && memcmp(PS_MARK_E, bytes + len - strlen(PS_MARK_E), strlen(PS_MARK_E)) != 0)||ret != len)
     {
-        spdlog::error("invalid packet: {} {}", ret, len);
+        spdlog::error("invalid avformatctx: {} {}", ret, len);
         return -1;
     }
     spdlog::info("decode len: {}", ret);