#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[TERMINAL_SN_SIZE] = {0};

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


static uint64_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";

/// flags used to stop modules
/// 0: stopped, 1: requested to stop, 2: runningstatic int

typedef struct ev_region_t {
    float minx;
    float miny;
    float maxx;
    float maxy;
} ev_region_t;
typedef struct ev_module_config_t {
    int stat_current;
    ev_region_t region;
    bool enabled;
    union {
        struct {
            int interval; // in seconds
        } record;
        struct {
            float face_thresh; //[0,1]
            float human_thresh;
        } ai;
        struct motion {
            int level; // 1 - 6
        } motion;
    } module;
} ev_module_config_t;

static ev_module_config_t gConfigRecord = {0};
static ev_module_config_t gConfigAI = {0};
static ev_module_config_t gConfigMotion = {0};

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)
{
    if(sig == SIGPIPE) {
        // vgw connection error
        bConnected = false;
        bNeedRetry = true;
        return;
    }

    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;
}

//
void frame_free(void *data, void*hint)
{
    MaQueVideoEncFrameInfo_s *frame = (MaQueVideoEncFrameInfo_s *)hint;
    MaQue_Demo_Mem_release(frame->handleMem);
}

//
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++;
    zmq_msg_t msg;
    void *hint = NULL;
    if(frame->eEncodeType == MAQUE_ENCODE_TYPE_H264 && frame->eFrameType == MAQUE_FRAME_TYPE_VIDEO && frame->nDataLen > 0) {
        evpacket_t* pkt = (evpacket_t*)((char*)frame->pData - sizeof(evpacket_t));
        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++;
        }
        zmq_msg_init_data(&msg, (char*)pkt, frame->nDataLen + sizeof(evpacket_t), nullptr, nullptr);
        zmq_msg_send(&msg, pPub, 0);
        packetId++;
    }

    MaQue_Demo_Mem_release(frame->handleMem);

    return XM_SUCCESS;
}

void frame_send_entry(void *args)
{
    int ret = 0;
    // setup sub
    void *pSub = zmq_socket(pPubCtx, ZMQ_SUB);
    if(pSub == nullptr) {
        spdlog::error("failed to create frame sending socket");
        exit(1);
    }
    // ret = 1;
    // zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
    // ret = 2;
    // zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
    // zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
    ret = zmq_connect(pSub, strPubUrl.c_str());
    if(ret != 0) {
        spdlog::error("failed connect frame pub: {}", strPubUrl);
        exit(1);
    }
    ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
    spdlog::info("zmq sub successed (for frame sending)");

    zmq_msg_t msg;
    while (1) {
        if(!bConnected||raw_socket_ <= 0) {
            if(frameCntTotal % (NUM_MAX_QUEQUE_SIZE * 3) == 0) {
                bNeedRetry = true;
                if(raw_socket_) {
                    ::close(raw_socket_);
                    raw_socket_ = 0;
                }
            }
            if(bNeedRetry == true) {
                // try reconect
                bNeedRetry = false;
                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
                }
            }
            continue;
        }

        zmq_msg_init(&msg);
        ret = zmq_msg_recv(&msg, pSub, 0);
        if(ret < 0) {
            spdlog::error("failed to recv zmq msg: {}", zmq_strerror(ret));
            continue;
        }

        ssize_t sz = zmq_msg_size(&msg);
        char *ptr = (char*)zmq_msg_data(&msg);

        ssize_t sent = 0;
        ssize_t totalSent = 0;
        while (sz > 0) {
            ptr += sent;
            auto tosent = sz > NUM_MAX_PACKET_BYTES? NUM_MAX_PACKET_BYTES:sz;
            sent = ::send(raw_socket_, ptr, tosent, (tosent > sz? MSG_MORE:0)|MSG_NOSIGNAL);
            if (sent <= 0) {
                break;
            }
            sz -= sent;
            totalSent += sent;
        }

        if (sent <= 0) {
            spdlog::error("failed to send, peer down: {}", sz);
            /// TODO: reconnection
            bConnected = false;
            bNeedRetry = true;
        }

        zmq_msg_close(&msg);
    }
    zmq_close (pSub);
}
using namespace chrono_literals;
void record_video_entry(void *args)
{
    int ret = 0;
    // this thread will never exit
    while(1) {
        if(!is_sdcard_avail()) {
            spdlog::error("no sd card valial. can't record local videos");
            this_thread::sleep_for(10s);
            continue;
        }

        // setup sub
        void *pSub = zmq_socket(pPubCtx, ZMQ_SUB);
        // ret = 1;
        // zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
        // ret = 2;
        // zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
        // zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
        ret = zmq_connect(pSub, strPubUrl.c_str());
        if(ret != 0) {
            spdlog::error("failed connect frame pub: {}", strPubUrl);
            exit(1);
        }
        ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
        spdlog::info("zmq sub successed (record)");

        zmq_msg_t msg;
        long long preTs = 0;
        long long nowTs = 0;
        uint32_t cnt = 0;
        fstream * fp = nullptr;
        while (1) {
            zmq_msg_init(&msg);
            ret = zmq_msg_recv(&msg, pSub, 0);
            if(ret < 0) {
                spdlog::error("failed to recv zmq msg: {}", zmq_strerror(ret));
                continue;
            }
            ssize_t sz = zmq_msg_size(&msg);
            char *ptr = (char*)zmq_msg_data(&msg);
            /// we calc timestamps about every 2s
            if(cnt % 30 == 0) {
                nowTs = chrono::system_clock::now().time_since_epoch().count()/1000;
            }
            if(nowTs - preTs > gConfigRecord.module.record.interval*1000) {
                if(fp != nullptr) {
                    if(fp->is_open())
                        fp->close();
                    delete fp;
                }
                fp = new fstream(recFilePath+to_string(nowTs), std::ios::out|std::ios::binary);
                if(!fp->is_open()) {
                    spdlog::error("failed to open record file");
                    delete fp;
                    fp = nullptr;
                }
                preTs = nowTs;
            }

            if(fp != nullptr) {
                fp->write(ptr, sz);
            }
            cnt++;
            zmq_msg_close(&msg);
        }

        zmq_close (pSub);
        /// DO NOT! do this
        //zmq_ctx_destroy(pPubCtx);
    }

}

/// 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)
{
    try {
        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);
        }
    }
    catch(exception &e) {
        spdlog::error("failed to send message: {}, {}, {}:{}", e.what(), message, __FILE__, __LINE__);
    }
}

/// 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 = fmt::format("exception {}: {}, {}:{}", e.what(), msg, __FILE__, __LINE__);
        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

    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);
    signal(SIGPIPE, SIG_IGN);

    // 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=true) {
        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();
}