#include <iostream>
#include <string>
#include <chrono>
#include <fmt/format.h>
#include <spdlog/spdlog.h>
#include "ref-memory.hpp"
#include <maque_system.h>
#include <maque_video_enc.h>
#include <maque_time.h>
#include <evpacket.h>
#include "common.h"
#include "ntp.h"
#include <time.h>
#include "raw_tcp.h"
#include "smart.h"
#include "motion.h"
#include <thread>
#include "mqttmgr.hpp"
#include "utils.hpp"
#include <zmq.h>

using namespace std;
using namespace jsoncons;
using namespace evutils;

///
#define NUM_IFRAME_PICK 2
#define NUM_MAX_QUEQUE_SIZE 60
#define NUM_MAX_PACKET_BYTES 1200
#define INTERVAL_LOG_QFULL 3    // its about 60/15 *3 secs

/// device info
static char dev_sn[13] = {0};

/// default video gateway info
static int videoQuality = MAQUE_IMG_QUALITY_BETTER;


static uint32_t frameCntTotal = 0;
static uint32_t frameCntIframe = 0;
static uint32_t frameCntPframe = 0;
static uint64_t packetId = 0;
static spdlog::level::level_enum logLevel = spdlog::level::info;
static bool bPFrameAvail = false;
static bool bNeedRetry = false;
static int readFaildCnt = 0;
static bool enablePush = false;
static bool enableRecord = false;
static bool bCanRecord = false;
static bool bConnected = false;

static queue<DataItem> _frameQueue;
static mutex _mutFrame;
static condition_variable _condFrame;
static Notifier _notiFrame = {&_mutFrame, &_condFrame};
static CallBackArg args = {nullptr, &_notiFrame, &_frameQueue};
static int raw_socket_ = 0;
static bool bGotTime = false;

// zmq
static void *pPubCtx = nullptr;
static void *pPub = nullptr;
static const string strPubUrl = "inproc://frame";

enum EV_MSG_ERROR_CODE {
    EV_MSG_ERROR_NONE,
    EV_MSG_ERROR_EXCEPTION,
    EV_MSG_ERROR_UNSUPPORTED_CMD,
    EV_MSG_ERROR_CONTENT_SYNTAX,
    EV_MSG_ERROR_NUM
};

void clean_up(int sig)
{
    spdlog::warn("clearn up called, with sig: {}", sig);
    if (args.recFD) {
        fclose(args.recFD);
    }

    if (raw_socket_) {
        ::close(raw_socket_);
    }

    // while (args.dataq->size() > 0) {
    //     DataItem elem = args.dataq->front();
    //     args.dataq->pop();
    //     MaQueVideoEncFrameInfo_s *pMem = (MaQueVideoEncFrameInfo_s *)elem.ud;
    //     MaQue_Demo_Mem_release(pMem->handleMem);
    // }

    //stop_md_bd();
    LibXmMaQue_System_destroy();
    exit(1);
}

bool is_big_endian(void)
{
    union {
        uint32_t i;
        char c[4];
    } bint = {0x01020304};

    return bint.c[0] == 1;
}

MaQueVideoEncodeCfg_s *init_stream_cfg(MaQueVideoEncodeCfg_s *cfg)
{
    cfg->eVidComp = (MaQueVideoCompress_e)-1;
    cfg->eCapSize = (MaQueCaptureSize_e)-1;
    cfg->eBitrateCtrl = (MaQueBitrateCtrl_e)-1;
    cfg->eImageQuality = (MaQueImageQuality_e)-1;
    cfg->nFps = -1;
    cfg->nBitRate = -1;
    cfg->nIFrameInterval = -1;
    cfg->iQpMin = -1;
    cfg->iQpMax = -1;

    return cfg;
}

int32_t configure_stream(MaQueStreamChannel_e eStreamChn, MaQueVideoEncodeCfg_s *cfg)
{
    int32_t res;
    MaQueVideoEncodeCfg_s oldCfg;

    if (LibXmMaQue_VideoEnc_getCfg(0, eStreamChn, &oldCfg)) {
        spdlog::error("LibXmMaQue_VideoEncode_getCfg error");
        return -1;
    }

    if (cfg->eVidComp != -1) {
        oldCfg.eVidComp = cfg->eVidComp;
    }
    if (cfg->eCapSize != -1) {
        oldCfg.eCapSize = cfg->eCapSize;
    }
    if (cfg->eBitrateCtrl != -1) {
        oldCfg.eBitrateCtrl = cfg->eBitrateCtrl;
    }

    if (cfg->eImageQuality != -1) {
        oldCfg.eImageQuality = cfg->eImageQuality;
    }

    if (cfg->nFps != -1) {
        oldCfg.nFps = cfg->nFps;
    }
    if (cfg->nBitRate != -1) {
        oldCfg.nBitRate = cfg->nBitRate;
    }

    if (cfg->nIFrameInterval != -1) {
        oldCfg.nIFrameInterval = cfg->nIFrameInterval;
    }

    if (cfg->iQpMin > 0 && cfg->iQpMin <= 51 && cfg->iQpMax > 0 && cfg->iQpMax <= 51) {
        oldCfg.iQpMin = cfg->iQpMin;
        oldCfg.iQpMax = cfg->iQpMax;
        oldCfg.bUseUserQp = 1;
    }

    spdlog::info("main: fps={}", oldCfg.nFps);
    res = LibXmMaQue_VideoEnc_setCfg(0, eStreamChn, &oldCfg);
    if (!res) {
        memcpy(cfg, &oldCfg, sizeof(oldCfg));
    }

    return res;
}

//
XM_S32 cb_frame_proc(XM_VOID *pUserArg, MaQueVideoEncFrameInfo_s *frame)
{
    MaQueSystemTime_s *pstSysTime = &frame->stTimeStamp.stSysTime;
    CallBackArg *args = (CallBackArg *)pUserArg;
    struct timeval stTimeVal;
    int ret;

    frameCntTotal++;
    if(logLevel <= spdlog::level::debug) {
        printf("\n\nnew frame: %u, p0 h:%08X\n", frameCntTotal, (int)frame->handleMem);
    }

    if (bCanRecord) {
        auto nWrite = fwrite(frame->pData, 1, frame->nDataLen, args->recFD);
        if(frame->nDataLen != nWrite) {
            spdlog::error("failed to write file");
            if(args->recFD) {
                fclose(args->recFD);
            }
        }
    }

    if(enablePush && bConnected == false && frameCntTotal % (NUM_MAX_QUEQUE_SIZE * 3) == 0) {
        bNeedRetry = true;
        bConnected = false;
        if(raw_socket_) {
            ::close(raw_socket_);
            raw_socket_ = 0;
        }
    }

    //printf("\n==== prepare frame with enablePush:%d, dataQ: %08hhX, frametype: %d\n", enablePush, (int)args->dataq, frame->eEncodeType);
    if (enablePush && args->dataq && frame->eEncodeType == MAQUE_ENCODE_TYPE_H264) {
        unique_lock<mutex> lock(*args->noti->mut);
        // check delayed packets for about 12s. sufficiently high
        if (args->dataq->size() >= NUM_MAX_QUEQUE_SIZE) {
            spdlog::warn("dataq full");
            volatile MaQueVideoEncFrameInfo_s * p1 = (MaQueVideoEncFrameInfo_s *)args->dataq->front().ud;
            if(logLevel == spdlog::level::debug) {
                printf("%u, p1 h:%08X\n", frameCntTotal, (int)p1->handleMem);
            }
            args->dataq->pop();
            if(p1 == frame) {
                spdlog::error("BUG!!! new frame address same as what in Queue");
            }
            MaQue_Demo_Mem_release(p1->handleMem);
            p1 = nullptr;
        }

        if (frame->pData && frame->nDataLen > 0) {
            timeval tv;
            //::gettimeofday(&tv,NULL);
            DataItem dt = {(char *)frame->pData - sizeof(evpacket_t), frame->nDataLen + sizeof(evpacket_t), (void *)frame};
            evpacket_t *pkt = (evpacket_t *)dt.buf;

            memset(pkt, 0, sizeof(*pkt));
            pkt->meta.magic[0] = 0xBE;
            pkt->meta.magic[1] = 0xEF;
            memcpy(pkt->meta.sn, dev_sn, sizeof(dev_sn));
            /// NOTES: since both ends use little-endian, we don't use htonl, htons to save clocks
            pkt->vpara.frame_type = frame->eSubType + 1;
            pkt->vpara.packet_type = frame->eEncodeType;
            pkt->vpara.res.width = frame->nWidth;
            pkt->vpara.res.height = frame->nHeight;
            /// TODO:
            pkt->vpara.fps = 15;
            //pkt->vpara.ts = frame->stTimeStamp;
            pkt->length = frame->nDataLen;
            pkt->meta.packet_id = packetId;
            pkt->meta.crc = 0;
            pkt->meta.crc = crc16((unsigned char*)pkt, sizeof(evpacket_t));
            if (frame->eSubType == MAQUE_FRAME_SUBTYPE_I) {
                frameCntIframe++;
            }
            else if(frame->eSubType == MAQUE_FRAME_SUBTYPE_P) {
                frameCntPframe++;
            }

            MaQue_Demo_Mem_addRef(frame->handleMem);
            args->dataq->push(dt);
            bPFrameAvail = true;
            spdlog::debug("frame meter ic:{}, pc:{}, tc:{}, len:{}, cid:{}", frameCntIframe, frameCntPframe,frameCntTotal, frame->nDataLen, packetId);
            packetId++;
        }
        else {
            spdlog::error("invalid frame");
        }

        lock.unlock();
        args->noti->cond->notify_one();
    }
    else {
        //spdlog::warn("pusher is disabled {}:{}", __FILE__, __LINE__);
    }

    if(logLevel <= spdlog::level::debug) {
        printf("%u, p2 h:%08X\n", frameCntTotal, (int)frame->handleMem);
    }

    MaQue_Demo_Mem_release(frame->handleMem);
    return XM_SUCCESS;
}

void frame_send_entry(void *args)
{
    CallBackArg *pvArg = (CallBackArg *)args;
    int freshConnect = 0;
    DataItem elem = {0};
    while (1) {
        MaQueVideoEncFrameInfo_s *pMem;
        {
            unique_lock<mutex> lk(*pvArg->noti->mut);
            pvArg->noti->cond->wait(lk, [pvArg] { return !pvArg->dataq->empty();});
            if (pvArg->dataq->empty()) {
                spdlog::warn("dataq empty");
                continue;
            }
            // /// remove old frames
            // lk.unlock();
            // while(pvArg->dataq->size() > 2) {
            //     lk.lock();
            //     elem = pvArg->dataq->front();
            //     pvArg->dataq->pop();
            //     lk.unlock();
            //     pMem = (MaQueVideoEncFrameInfo_s *)elem.ud;
            //     MaQue_Demo_Mem_release(pMem->handleMem);
            // }

            // lk.lock();
            elem = pvArg->dataq->front();
            pvArg->dataq->pop();
            lk.unlock();
            pMem = (MaQueVideoEncFrameInfo_s *)elem.ud;
        }

        if(bConnected == false) {
            if(bNeedRetry == true) {
                // try reconect
                bNeedRetry = false;
                if(raw_socket_ != 0) {
                    ::close(raw_socket_);
                    raw_socket_ = 0;
                }
                spdlog::info("try reconnecting to video gateway {}:{}", host, port);
                auto ret = raw_connect(host, port, &raw_socket_);
                if (ret < 0 || raw_socket_ <= 0) {
                    spdlog::error("failed to reconnect {}:{}", host, port);
                    bConnected = false;
                }
                else {
                    spdlog::info("successfully connected to {}:{}", host, port);
                    bConnected = true;
                    //LibXmMaQue_OSD_showTime(0, MAQUE_STREAM_CHN_MAIN, &stOsdTimeModParamMain.stOsdParam);
                    //freshConnect = pvArg->dataq->size();
                    /// pop all frames
                }
            }
        }

        if (bConnected == false || !pMem||elem.size <= 0 || elem.buf == nullptr || (elem.buf + sizeof(evpacket_t)) == nullptr) {
            //printf("invalid frame. addr: %08X, len:%u\n", (uint32_t)elem.buf, elem.size);
            string msg;
            if(elem.size <=0) {
                msg = fmt::format("invalid frame size: {}", elem.size);
                spdlog::warn(msg);
            }
            else if(elem.buf == nullptr||elem.buf + sizeof(evpacket_t) == nullptr) {
                msg = fmt::format("invalid buffer addr: {0:p}", elem.buf);
                spdlog::warn(msg);
            }
            else {
                /// bConnected is false. needn't to log again
                //msg = fmt::format("no connection to video gateway server {}:{}", host, port);
                //spdlog::warn(msg);
            }

            /// release msg
            if(pMem && pMem->handleMem) {
                if(logLevel <= spdlog::level::debug) {
                    printf("%u, p3 h:%08X\n", frameCntTotal, (int)pMem->handleMem);
                }
                MaQue_Demo_Mem_release(pMem->handleMem);
                pMem = nullptr;
            }
            continue;
        }

        /// send frame
        if (elem.size > 0) {
            auto sz = elem.size;
            char *ptr = elem.buf;
            ssize_t sent = 0;
            while (elem.size > 0) {
                ptr += sent;
                sent = ::send(raw_socket_, ptr, elem.size > NUM_MAX_PACKET_BYTES? NUM_MAX_PACKET_BYTES:elem.size, 0);
                if (sent <= 0) {
                    break;
                }
                else {
                    //spdlog::debug("sent:{}, total:{}", sent, sz);
                }
                elem.size -= sent;
            }

            if (sent <= 0) {
                spdlog::error("failed to send, peer down");
                /// TODO: reconnection
                bConnected = false;
                bNeedRetry = true;
            }
        }
        else {
            spdlog::error("frame size error");
        }
        if(logLevel <= spdlog::level::debug) {
            printf("%u, p4 h:%08X\n", frameCntTotal, (int)pMem->handleMem);
        }
        MaQue_Demo_Mem_release(pMem->handleMem);
    }
}

/// verify request message
string verify_request(json &js)
{
    string ret;
    string check[] = {kMsgCmd,kMsgData, kMsgRid, kMsgTime, kMsgRid};
    for(auto &k: check) {
        if(!js.contains(k)) {
            ret += k + ",";
        }
    }

    if(!ret.empty()) {
        ret = "missing fields in request: " + ret;
    }

    return ret;
}

/// low level api for report and response
void _report_response(string topic, string message)
{
    auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
    if(client == nullptr) {
        spdlog::error("failed to create mqtt instance to {}", mqtt_url);
        return;
    }
    auto rc = (*client)[topic].pub(message.c_str(), message.size(), 2, false);
    if(rc < 0) {
        spdlog::error("failed to pub mqtt message on {}: {}", topic, message);
    }
    else {
        spdlog::info("successfully pub message at {}:{}", topic, message);
    }
}

/// having data field along with code and msg
/// signature {"rid", "code", "msg", "cata",  "data":{}}
void report_response_json(string topic, json& message)
{
    // time
    auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
    message.insert_or_assign(kMsgTime, ts);
    // rid
    auto rid = message["rid"].as<string>();
    message.insert_or_assign(kMsgRid, rid);
    // type
    if(rid.empty()) {
        message.insert_or_assign(kMsgType, kMsgTypeReport);
    }
    else {
        message.insert_or_assign(kMsgType, kMsgTypeResponse);
    }

    // sn
    message.insert_or_assign(kMsgSn, string(dev_sn));

    _report_response(topic, message.to_string());
}

/// no data field. only code and msg field
void report_response_args(string topic, int code, string message, string cata, string rid, json &&data)
{
    json js;
    /// current timestamp
    auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
    js.insert_or_assign(kMsgTime, ts);
    /// rid
    js.insert_or_assign(kMsgRid, rid);
    /// type
    if(rid.empty()) {
        js.insert_or_assign(kMsgType, kMsgTypeReport);
    }
    else {
        js.insert_or_assign(kMsgType, kMsgTypeResponse);
    }
    js.insert_or_assign(kMsgCata, cata);
    ///
    js.insert_or_assign(kMsgCode, code);
    js.insert_or_assign(kMsgMsg, message);
    js.insert_or_assign(kMsgSn, string(dev_sn));

    _report_response(topic, js.to_string());
}

int apply_config(json &js)
{
    auto hasSD = is_sdcard_avail();
    /// verify parameters
    string str, strWarn;
    if(!js.contains(kMsgData)) {
        str = fmt::format("no data object in config: {}", js.to_string());
    }
    else {
        json data = js[kMsgData];
        string veri[] = {kMsgConfigVgw, kMsgConfigMqtt, kMsgConfigUpload};
        for(auto &k:veri) {
            auto value = msg_field(data, k);
            if(value.empty()) {
                str = fmt::format("no {} field in data: {}", k, js.to_string());
                break;
            }
            else {
                // parse host and port
                auto uri = httplib::Uri::Parse(value);
                if(uri.Host.empty()|| uri.Port.empty()) {
                    str = fmt::format("host addr invalid {}:{}, {}", k, value, js.to_string());
                    break;
                }
                // check network connections
                int socket = 0;
                auto rc = raw_connect(uri.Host, uri.Port, &socket);
                if(rc < 0 || socket == 0) {
                    str = fmt::format("failed to connect to {}:{}, {}", k, value, js.to_string());
                    break;
                }
                else {
                    // we must close the test socket
                    ::close(socket);
                }
            }
        }
        // check features field
        json features = data.get_value_or<json>(kMsgConfigFeatures, json());
        string feats[]= {kMsgConfigMotion, kMsgConfigRecord, kMsgConfigAi};
        for(auto &k:feats) {
            auto f = features.get_value_or<json>(k, json());
            if(f.size() == 0) {
                strWarn = fmt::format("{}, ", k);
            }
            if(!strWarn.empty()) {
                strWarn = "no below feature in config, default configuration is applied: " + strWarn;
            }
        }
    }
    if(!str.empty()) {
        report_response_args(pub_topic, EV_MSG_ERROR_CONTENT_SYNTAX, str, kMsgCmdConfig, msg_field(js, kMsgRid), json());
        spdlog::error(str);
        return -EV_MSG_ERROR_CONTENT_SYNTAX;
    }

    // try to apply

}

/// handles all incoming request from cloud services
void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string topic)
{
    char msg[len];
    memcpy(msg, data, len);
    try {
        auto js = json::parse(msg);
        spdlog::info("received request on {}:\n{}", topic, (char*)msg);
        auto str = verify_request(js);
        if(!str.empty()) {
            spdlog::warn(str);
            report_response_args(pub_topic, EV_MSG_ERROR_CONTENT_SYNTAX, str, js.contains(kMsgCmd)?js[kMsgCmd].as<string>():string(""), js.contains(kMsgRid)?js[kMsgRid].as<string>():string(""), json());
            return;
        }

        /// dispatch msg
        auto cmd = js[kMsgCmd].as<string>();
        if(cmd == kMsgCmdConfig) {
            // apply config
        }
        else if(cmd == kMsgCmdUploadVideo) {
            // upload video
        }
        else if(cmd == kMsgCmdGetConfig) {
            // response config
        }
        else if(cmd == kMsgCmdStatus) {
            // response status
        }
        else if(cmd == kMsgCmdPtz) {
            // response Ptz control
        }
        else {
            auto str = fmt::format("unsupported cmd {}: {}", cmd, js.to_string());
            spdlog::warn(str);
            report_response_args(pub_topic, EV_MSG_ERROR_UNSUPPORTED_CMD, str, cmd, msg_field(js, kMsgRid), json());
        }

    }
    catch(exception &e) {
        auto str = string("exception: ") + e.what() + string(", ") + msg;
        spdlog::error(str);
        report_response_args(pub_topic, EV_MSG_ERROR_EXCEPTION, str, string(""), string(""), json());
    }
    //(*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}

///
void start_mqtt(void *arg)
{
    auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
    if(client == nullptr) {
        spdlog::error("failed to create mqtt instance to {}", mqtt_url);
        return;
    }
    client->subscribe(sub_topic + dev_sn, handle_mqtt_req, false);
}


///
int main(int argc, char *argv[])
{
    int ret = XM_SUCCESS;
    spdlog::info("test on hi3518ev300, {}", is_big_endian());

    get_mac_addr(dev_sn);

    // get env for host and port
    auto host_ = getenv("VGW_HOST");
    enablePush = true;
    if(host_) {
        host = host_;
    }
    else {
        enablePush = false;
    }

    auto port_ = getenv("VGW_PORT");
    if(port_) {
        port_ = port;
    }
    else {
        enablePush = false;
    }

    /// TODO: check sd availability
    bool bSDAvail = is_sdcard_avail();

    auto logLevel_ = getenv("LOG_LEVEL");
    if(logLevel_) {
        logLevel = spdlog::level::from_str(logLevel_);
        spdlog::set_level(logLevel);
    }
    else {
        spdlog::info("log level configured to info");
        spdlog::set_level(logLevel);
    }

    auto videoQuality_ = getenv("QUALITY");
    if(videoQuality_) {
        videoQuality = atoi(videoQuality_);
        if(videoQuality >= MAQUE_IMG_QUALITY_NR || videoQuality < 0) {
            videoQuality = MAQUE_IMG_QUALITY_BETTER;
        }
    }

    MaQue_Demo_Mem_Init();

    /// thread of utp time
    auto thNtp = thread([] {
        bool bGotTime = false;
        time_t stm;
        while(!bGotTime)
        {
            if(getNtpTime(&stm) >=0) {
                spdlog::info("ntp got time");
                ::stime(&stm);
                bGotTime = true;
            }
            else {
                spdlog::warn("failed to get ntp time");
            }
            this_thread::sleep_for(chrono::seconds(20));
        }
    });

    if(thNtp.joinable()) {
        thNtp.detach();
    }

    MaQueStartParam_s startParam = {MAQUE_VIDEO_STANDARD_PAL, {MAQUE_VIDEO_COMPRESS_H264, MAQUE_VIDEO_COMPRESS_H265}, "/mnt/sd/Config/"};
    ret = LibXmMaQue_System_startUp(&startParam);
    spdlog::info("ret: {}", ret);

    //
    MaQueMemoryApi_s memApi = {MaQue_Demo_Mem_alloc, MaQue_Demo_Mem_release, MaQue_Demo_Mem_addRef, MaQue_Demo_Mem_setLength};
    ret = LibXmMaQue_Mem_init(&memApi);
    spdlog::info("mem ret: {}", ret);

    MaQueSystemTime_s tm;
    LibXmMaQue_Time_getCurrentTime(&tm);
    LibXmMaQue_Time_print(&tm, nullptr);
    //
    MaQueCodeAbilities_s capb;
    ret = LibXmMaQue_VideoEnc_getAbilities(0, &capb);
    spdlog::info("abt ret: {}", ret);
    if (XM_SUCCESS == ret) {
        spdlog::info(fmt::format("capbilities: supported codecs {0:#b},"
                                 "max fhd/s {0:d}, max res: {0:d}, per ch mres: {0:d}, {0:d}, {0:d}, {0:d}",
                                 int(capb.videoEncTypeMask),
                                 int(capb.maxEncPowerX1080P), int(capb.eDecImageSizeMax), int(capb.astVidEncChnAbility[0].eCapSizeMax),
                                 int(capb.astVidEncChnAbility[1].eCapSizeMax), int(capb.astVidEncChnAbility[2].eCapSizeMax), int(capb.astVidEncChnAbility[3].eCapSizeMax)));
    }

    //
    MaQueVideoEncodeCfg_s cfg;
    init_stream_cfg(&cfg);
    cfg.nFps = 15;
    cfg.eIFrmIntvType = IFRAME_INTV_TYPE_TIME;
    cfg.nIFrameInterval = 10;
    cfg.eImageQuality = (MaQueImageQuality_e)videoQuality;
    cfg.eVidComp = MAQUE_VIDEO_COMPRESS_H264;
    cfg.nBitRate = 200 * 8; // 200KB
    ret = configure_stream((MaQueStreamChannel_e)0, &cfg);
    spdlog::info("cfg stream ret: {}", ret);
    ///

    /// record setting
    if(enableRecord && bSDAvail) {
        FILE *recFD = fopen(recFilePath, "w+");
        if (!recFD) {
            spdlog::error("fopen() Failed: {}", recFilePath);
            bCanRecord = false;
        }
        else {
            args.recFD = recFD;
            bCanRecord = true;
        }
    }

    ret = LibXmMaQue_VideoEnc_startStream(0, (MaQueStreamChannel_e)0, cb_frame_proc, (void *)&args);
    if (XM_SUCCESS == ret) {
        spdlog::info("created record task successfully");
    }

    signal(SIGINT, clean_up);
    // signal(SIGTERM, clean_up);
    signal(SIGKILL, clean_up);

    // setup zmq
    pPubCtx = zmq_ctx_new();
    pPub = zmq_socket(pPubCtx, ZMQ_PUB);
    ret = zmq_bind(pPub, strPubUrl.c_str());
    if(ret < 0) {
        spdlog::error("{} failed to create zmq pub topic: {}", strPubUrl);
        exit(1);
    }

    //

    if (enablePush) {
        ret = raw_connect(host, port, &raw_socket_);
        if (ret < 0 || raw_socket_ <= 0) {
            spdlog::error("failed connect to video gateway at {}:{}", host, port);
            bConnected = false;
        }
        else {
            bConnected = true;
            spdlog::info("successfully connected to video gateway at {}:{}", host, port);
        }
        thread thPush = thread(frame_send_entry, &args);
        spdlog::info("sizeof pkt header {}, sizeof tv {}", sizeof(evpacket_t), sizeof(timeval));
        thPush.detach();
    }
    else {
        spdlog::warn("push video is not enabled in environ settings");
    }

    start_md_bd(&args);
    /// subscribe to mqtt
    start_mqtt(nullptr);

    thread thSmart = thread(maq_smart_task_entry, &args);
    thSmart.join();
}