vcpe/srcs/vcpe_agent.c

//
// Created by xajhuang on 2022/6/10.
//
#include <uv.h>
#include <zlog.h>
#include <zmq.h>
#include <string.h>
#include "task_manager.h"
#include "user_errno.h"
#include "config.h"
#include "misc.h"
#include "init.h"

static void *g_pContext  = NULL;
static void *g_pResponse = NULL;

_Noreturn static void mqServerCb(void *UNUSED(pArg)) {
    while (TRUE) {
        zmq_msg_t msg;
        zmq_msg_init(&msg);

        if (zmq_msg_recv(&msg, g_pResponse, 0) != -1) {
            printf("Data channel receive(%zu): %s\n", zmq_msg_size(&msg), (const char *)zmq_msg_data(&msg));
            zmq_msg_close(&msg);
#if 0
            zmq_msg_close(&msg);

            zmq_msg_init_size(&msg, 5);

            memcpy(zmq_msg_data(&msg), "hello", 5);

            zmq_msg_send(&msg, g_pResponse, 0);
#endif
        }

        uv_sleep(10);
    }
}

static int data_mq_init(void) {
    static uv_thread_t uvThread;

    char buf[1024];

    g_pContext = zmq_ctx_new();

    if (g_pContext == NULL) {
        return -ERR_MQ_CREATE_MQ;
    }

    g_pResponse = zmq_socket(g_pContext, ZMQ_PAIR);

    if (g_pResponse == NULL) {
        zmq_ctx_destroy(g_pContext);
        return -ERR_MQ_CREATE_REP;
    }

    memset(buf, 0, 1024);

    sprintf(buf, "ipc:///tmp/msg_fifo0");
    printf("Start message data channel server: ipc:///tmp/msg_fifo0\n");


    if (zmq_bind(g_pResponse, buf) != 0) {
        zmq_close(g_pResponse);
        zmq_ctx_destroy(g_pContext);
        return -ERR_MQ_CONN_SERVER;
    }

    uv_thread_create(&uvThread, mqServerCb, NULL);

    return ERR_SUCCESS;
}

int main(int argc, char **argv) {
    uv_setup_args(argc, argv);

    data_mq_init();
    task_manager_run();

    return 0;
}