111 lines
2.3 KiB
C
111 lines
2.3 KiB
C
|
//
|
||
|
|
// Created by xajhu on 2021/7/15 0015.
|
||
|
|
//
|
||
|
|
#include <stdlib.h>
|
||
|
|
#include <zmq.h>
|
||
|
|
#include <uv.h>
|
||
|
|
#include <string.h>
|
||
|
|
#include <zlog.h>
|
||
|
|
|
||
|
|
#include "msg_queue.h"
|
||
|
|
#include "config.h"
|
||
|
|
#include "user_errno.h"
|
||
|
|
#include "misc.h"
|
||
|
|
|
||
|
|
static int g_mqExit = FALSE;
|
||
|
|
static void *g_pContext = NULL;
|
||
|
|
static void *g_pResponse = NULL;
|
||
|
|
|
||
|
|
static void process_msg(zmq_msg_t *pMsg) {
|
||
|
|
const char *pResp;
|
||
|
|
zmq_msg_t msg;
|
||
|
|
const char *pRecMsg = strdup((const char *)zmq_msg_data(pMsg));
|
||
|
|
|
||
|
|
dzlog_info("receive(%zu): %s\n", zmq_msg_size(pMsg), pRecMsg);
|
||
|
|
zmq_msg_close(pMsg);
|
||
|
|
|
||
|
|
pResp = on_msg_cmd(pRecMsg);
|
||
|
|
|
||
|
|
if (pResp != NULL && strlen(pResp) > 0) {
|
||
|
|
zmq_msg_init_size(&msg, strlen(pResp));
|
||
|
|
memcpy(zmq_msg_data(&msg), pResp, strlen(pResp));
|
||
|
|
zmq_msg_send(&msg, g_pResponse, 0);
|
||
|
|
free((void *)pResp);
|
||
|
|
} else {
|
||
|
|
zmq_msg_init_size(&msg, 5);
|
||
|
|
memcpy(zmq_msg_data(&msg), "error", 5);
|
||
|
|
zmq_msg_send(&msg, g_pResponse, 0);
|
||
|
|
}
|
||
|
|
|
||
|
|
zmq_msg_close(&msg);
|
||
|
|
free((void *)pRecMsg);
|
||
|
|
}
|
||
|
|
|
||
|
|
void mqServerCb(void *UNUSED(pArg)) {
|
||
|
|
while (TRUE) {
|
||
|
|
zmq_msg_t msg;
|
||
|
|
zmq_msg_init(&msg);
|
||
|
|
|
||
|
|
if (zmq_msg_recv(&msg, g_pResponse, 0) != -1) {
|
||
|
|
process_msg(&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
|
||
|
|
}
|
||
|
|
|
||
|
|
if (g_mqExit) {
|
||
|
|
break;
|
||
|
|
}
|
||
|
|
|
||
|
|
uv_sleep(10);
|
||
|
|
}
|
||
|
|
|
||
|
|
zmq_close(g_pResponse);
|
||
|
|
zmq_ctx_destroy(g_pContext);
|
||
|
|
}
|
||
|
|
|
||
|
|
int 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_REP);
|
||
|
|
|
||
|
|
if (g_pResponse == NULL) {
|
||
|
|
zmq_ctx_destroy(g_pContext);
|
||
|
|
return -ERR_MQ_CREATE_REP;
|
||
|
|
}
|
||
|
|
|
||
|
|
memset(buf, 0, 1024);
|
||
|
|
|
||
|
|
sprintf(buf, "tcp://*:%d", cfg_get_zero_mq_port());
|
||
|
|
dzlog_info("Start message queue server: tcp://*:%d\n", cfg_get_zero_mq_port());
|
||
|
|
|
||
|
|
if (zmq_bind(g_pResponse, buf) != 0) {
|
||
|
|
zmq_close(g_pResponse);
|
||
|
|
zmq_ctx_destroy(g_pContext);
|
||
|
|
return -ERR_MQ_BIND_SOCKET;
|
||
|
|
}
|
||
|
|
|
||
|
|
uv_thread_create(&uvThread, mqServerCb, NULL);
|
||
|
|
|
||
|
|
return ERR_OK;
|
||
|
|
}
|
||
|
|
|
||
|
|
void mq_uninit(void) {
|
||
|
|
g_mqExit = TRUE;
|
||
|
|
}
|