init

opencv-motion-detect/Makefile

@@ -5,16 +5,20 @@ CFLAGS = -g -Wall
 
 LIBOPENCV = `pkg-config opencv --cflags --libs`
 LIBFFMPEG = `pkg-config libavformat libavutil libavcodec --cflags --libs`
+LIBS=
 
 .PHONY: all
-all: libzmq rtspr
+all: libzmq evmgr evpusher
 
 .PHONY: libzmq
 libzmq:
-	cd vendor/libzmq && [ -f $(CURDIR)/vendor/lib/pkgconfig/libzmq.pc ] || ./autogen.sh && ./configure --prefix=$(CURDIR)/vendor
+	cd vendor/libzmq && [ ! -f $(CURDIR)/vendor/lib/pkgconfig/libzmq.pc ] || ./autogen.sh && ./configure --prefix=$(CURDIR)/vendor
 	cd vendor/libzmq && make -j 4 && make install
-rtspr: rtsp-relay.cpp
-	$(CPP) $(CPPFLAGS) -o rtspr rtsp-relay.cpp $(LIBFFMPEG) `pkg-config --cflags --libs vendor/lib/pkgconfig/libzmq.pc`
+evmgr: evmgr.cpp inc/common.hpp inc/tinythread.hpp
+	$(CPP) $(CPPFLAGS) -o evmgr evmgr.cpp $(LIBFFMPEG) `pkg-config --cflags --libs vendor/lib/pkgconfig/libzmq.pc` $(LIBS)
+
+evpusher: evpusher.cpp inc/common.hpp inc/tinythread.hpp
+	$(CPP) $(CPPFLAGS) -o evpusher evpusher.cpp $(LIBFFMPEG) `pkg-config --cflags --libs vendor/lib/pkgconfig/libzmq.pc` $(LIBS)
 	
 rtsp: rtsp.cpp
 	$(CPP) $(CFLAGS) -o rtsp rtsp.cpp $(LIBFFMPEG)
@@ -26,4 +30,7 @@ mux: demuxing_decoding.c
 	$(CC) $(CFLAGS) -o mux demuxing_decoding.c $(LIBFFMPEG)
 
 clean:
-	rm -fr rtsp cvprog mux rtspr *.dSYM
+	mkdir -p vendor/lib/pkgconfig/bak
\ No newline at end of file
+	mv vendor/lib/pkgconfig/*.pc vendor/lib/pkgconfig/bak
+	rm -fr evmgr evpusher *.dSYM vendor/lib/pkgconfig/*.pc
+	cd vendor/libzmq && make clean
\ No newline at end of file

opencv-motion-detect/evpusher.cpp

+#pragma GCC diagnostic ignored "-Wunused-private-field"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+
+
+#include <stdlib.h>
+#include <string>
+#include <thread>
+#include <iostream>
+#include <chrono>
+#include <future>
+
+#ifdef OS_LINUX
+#include <filesystem>
+namespace fs = std::filesystem;
+#endif
+
+#include  "vendor/include/zmq.h"
+#include "inc/json.hpp"
+#include "inc/blockingconcurrentqueue.hpp"
+#include "inc/tinythread.hpp"
+#include "inc/common.hpp"
+
+using namespace std;
+using json = nlohmann::json;
+using namespace moodycamel;
+
+class PacketPusher: public TinyThread {
+private:
+// 2M
+#define MAX_ZMQ_MSG_SIZE 1204 * 1024 * 2
+    void *pSubContext = NULL; // for packets relay
+    void *pSubscriber = NULL;
+    int setupMq()
+    {
+        teardownMq();
+        int ret = 0;
+        pSubContext = zmq_ctx_new();
+        pSubscriber = zmq_socket(pSubContext, ZMQ_SUB);
+        ret = zmq_setsockopt(pSubscriber, ZMQ_SUBSCRIBE, "", 0);
+        if(ret != 0) {
+            logThrow(NULL, AV_LOG_FATAL, "failed connect to pub");
+        }
+        ret = zmq_connect(pSubscriber, "tcp://172.31.0.96:5556");
+        if(ret != 0) {
+            logThrow(NULL, AV_LOG_FATAL, "failed create sub");
+        }
+        
+        return 0;
+    }
+
+    int teardownMq()
+    {
+        if(pSubscriber != NULL) {
+            zmq_close(pSubscriber);
+        }
+        if(pSubscriber != NULL) {
+            zmq_ctx_destroy(pSubscriber);
+        }
+        return 0;
+    }
+protected:
+    void run()
+    {
+        int ret = 0;
+        bool bStopSig = false;
+        zmq_msg_t msg;
+        av_log_set_level(AV_LOG_DEBUG);
+        while (true) {
+            if(checkStop() == true) {
+                bStopSig = true;
+                break;
+            }
+            int ret =zmq_msg_init(&msg);
+            if(ret != 0) {
+                av_log(NULL, AV_LOG_ERROR, "failed to init zmq msg");
+                continue;
+            }
+            ret = zmq_recvmsg(pSubscriber, &msg, 0);
+            if(ret < 0) {
+                av_log(NULL, AV_LOG_ERROR, "failed to recv zmq msg");
+                continue;
+            }
+
+            av_log(NULL, AV_LOG_DEBUG, "msg size: %d, %d", ret, zmq_msg_size(&msg));
+            zmq_msg_close(&msg);
+        }
+        if(!bStopSig && ret < 0) {
+            //TOOD: reconnect
+            av_log(NULL, AV_LOG_ERROR, "TODO: failed, reconnecting");
+        }else {
+            av_log(NULL, AV_LOG_INFO, "exit on command");
+        }
+    }
+
+public:
+    PacketPusher()
+    {
+        setupMq();
+    }
+    ~PacketPusher()
+    {
+        teardownMq();
+    }
+};
+
+int main(int argc, char *argv[]){
+    
+    PacketPusher pusher;
+    pusher.join();
+}
\ No newline at end of file

opencv-motion-detect/inc/common.hpp

+#ifndef __COMMON_H__
+#define __COMMON_H__
+extern "C" {
+#include <libavformat/avformat.h>
+}
+#include <libavutil/timestamp.h>
+#undef av_err2str
+#define av_err2str(errnum) av_make_error_string((char*)__builtin_alloca(AV_ERROR_MAX_STRING_SIZE), AV_ERROR_MAX_STRING_SIZE, errnum)
+
+void logThrow(void * avcl, int lvl, const char *fmt, ...)
+{
+    (void) avcl;
+    (void) lvl;
+    va_list args;
+    va_start( args, fmt );
+    av_log(NULL, AV_LOG_FATAL, fmt, args);
+    va_end( args );
+    throw fmt;
+}
+
+namespace PacketSerializer {
+    int encode(AVPacket &pkt, char **bytes) {
+        int wholeSize = 4 + pkt.size;
+        if(pkt.side_data_elems != 0) {
+            for(int i = 0; i < pkt.side_data_elems; i++) {
+                wholeSize += pkt.side_data[i].size + 8;
+            }
+        }else{
+            wholeSize +=4;
+        }
+        wholeSize += 8 * 5 + 4;
+        *bytes = (char*)malloc(wholeSize);
+
+        // data
+        memcpy(*bytes, &(pkt.size), 4);
+        memcpy(*bytes, pkt.data, pkt.size);
+        //side data
+        memcpy(*bytes, &(pkt.side_data_elems), 4);
+        if(pkt.side_data_elems != 0) {
+            for(int i = 0; i < pkt.side_data_elems; i++) {
+                memcpy(*bytes, &(pkt.side_data[i].size), 4);
+                memcpy(*bytes, pkt.side_data[i].data, pkt.side_data[i].size);
+                memcpy(*bytes, &(pkt.side_data[i].type), 4);
+            }
+        }else{
+            wholeSize +=4;
+        }
+
+        // other properties
+        memcpy(*bytes, &(pkt.pts), 8);
+        memcpy(*bytes, &(pkt.dts), 8);
+        memcpy(*bytes, &(pkt.pos), 8);
+        memcpy(*bytes, &(pkt.duration), 8);
+        memcpy(*bytes, &(pkt.convergence_duration), 8);
+        memcpy(*bytes, &(pkt.flags), 4);
+        av_log_set_level(AV_LOG_DEBUG);
+        av_log(NULL, AV_LOG_DEBUG, "\n\n\npkt origin size %d, serialized size: %d, elems:%d\n\n\n", pkt.size, wholeSize, pkt.side_data_elems);
+        return wholeSize;
+    }
+
+    AVPacket *decode(char * bytes) {
+        // allocate packet mem on heap
+        AVPacket *pkt = (AVPacket*)malloc(sizeof(AVPacket));
+        int got = 0;
+        memcpy(&(pkt->size), bytes, 4);
+        got += 4;
+        av_new_packet(pkt, pkt->size);
+        memcpy(pkt->data, bytes + got, pkt->size);
+        got += pkt->size;
+        memcpy(&pkt->side_data_elems, bytes + got, 4);
+        got += 4;
+        for(int i = 0; i < pkt->side_data_elems; i++) {
+            memcpy(&(pkt->side_data[i].size), bytes+got, 4);
+            got += 4;
+            memcpy(pkt->side_data[i].data,bytes + got ,pkt->side_data[i].size);
+            got += pkt->side_data[i].size;
+            memcpy(&(pkt->side_data[i].type), bytes + got, 4);
+            got += 4;
+        }
+
+        // props
+        memcpy(&(pkt->pts), bytes + got, 8);
+        got += 8;
+        memcpy(&(pkt->dts), bytes + got, 8);
+        got += 8;
+        memcpy(&(pkt->pos), bytes + got, 8);
+        got += 8;
+        memcpy(&(pkt->duration), bytes + got, 8);
+        got += 8;
+        memcpy(&(pkt->convergence_duration), bytes + got, 8);
+        got += 8;
+        memcpy(&(pkt->flags), bytes + got, 4);
+        got += 4;
+
+        return pkt;
+    }
+}
+
+void mqPacketFree(void *data, void*hint) {
+    free(data);
+}
+
+#endif
+

opencv-motion-detect/inc/tinythread.hpp

+#ifndef __TINY_THREAD__
+#define __TINY_THREAD__
+
+#include <thread>
+#include <iostream>
+#include <chrono>
+#include <future>
+
+using namespace std;
+
+
+class TinyThread {
+    std::promise<void> exitSignal;
+    std::future<void> futureObj;
+    int state = 0;
+    thread th;
+protected:
+    // Task need to provide defination  for this function
+    // It will be called by thread function
+    virtual void run() = 0;
+public:
+    TinyThread() :
+        futureObj(exitSignal.get_future())
+    {
+
+    }
+    TinyThread(TinyThread && obj) : exitSignal(std::move(obj.exitSignal)), futureObj(std::move(obj.futureObj))
+    {
+        std::cout << "Move Constructor is called" << std::endl;
+    }
+    TinyThread & operator=(TinyThread && obj)
+    {
+        std::cout << "Move Assignment is called" << std::endl;
+        exitSignal = std::move(obj.exitSignal);
+        futureObj = std::move(obj.futureObj);
+        return *this;
+    }
+
+    // Thread function to be executed by thread
+private:
+    void _run()
+    {
+        if(state == 0) {
+            th =thread([&]() {
+                this->run();
+            });
+            state = 1;
+        }
+    }
+
+public:
+    //Checks if thread is requested to stop
+    bool checkStop()
+    {
+        // checks if value in future object is available
+        if (futureObj.wait_for(std::chrono::milliseconds(0)) == std::future_status::timeout)
+            return false;
+        return true;
+    }
+    // Request the thread to stop by setting value in promise object
+    void stop()
+    {
+        exitSignal.set_value();
+    }
+
+    void join()
+    {
+        _run();
+        if(th.joinable()) {
+            th.join();
+        }
+    }
+
+    void detach()
+    {
+        _run();
+        if(th.joinable()) {
+            th.detach();
+        }
+    }
+};
+
+#endif
\ No newline at end of file