OCT

REM:
1. 增加Agent接收PPPoE信息测试程序

srcs/CMakeLists.txt

@@ -1,4 +1,5 @@
 SET(PROJECT_TARGET vcpe_main)
+SET(PROJECT_TARGET_AGENT vcpe_agent)
 
 PROJECT(${PROJECT_TARGET})
 
@@ -24,8 +25,10 @@ AUX_SOURCE_DIRECTORY(pppoe VCPE_SRC)
 AUX_SOURCE_DIRECTORY(user VCPE_SRC)
 
 ADD_EXECUTABLE(${PROJECT_TARGET} ${VCPE_SRC} ${VCPE_HEADS} vcpe_main.c)
+ADD_EXECUTABLE(${PROJECT_TARGET_AGENT} ${VCPE_HEADS} vcpe_agent.c)
 
 TARGET_LINK_LIBRARIES(${PROJECT_TARGET} common lwip_linux ${COMMON_LIBS})
+TARGET_LINK_LIBRARIES(${PROJECT_TARGET_AGENT} common ${COMMON_LIBS})
 
 # 自动复制配置文件到运行路径
 ADD_CUSTOM_COMMAND(TARGET ${PROJECT_TARGET}

srcs/libs/init/init.c

@@ -91,6 +91,10 @@ int user_init(const char *pAppCfgFile, const char *pCfgDirectory, const char *pK
         dzlog_error("Message queue init error: %d\n", ret);
     }
 
+    if ((ret = mq_data_init(NULL)) != ERR_SUCCESS) {
+        dzlog_error("Message queue init error: %d\n", ret);
+    }
+
     return ERR_SUCCESS;
 }
 

srcs/libs/mq/mq_data.c

@@ -72,7 +72,7 @@ _Noreturn static void mqDataChannelCb(void *pDataCh) {
 
 int mq_data_init(DATACHNNELCB dataCb) {
     static uv_thread_t uvThread;
-    void              *pContext = get_mq_context();
+    void              *pContext = zmq_ctx_new();
 
     char buf[1024];
 
@@ -82,7 +82,7 @@ int mq_data_init(DATACHNNELCB dataCb) {
 
     g_pDataChCb = dataCb;
 
-    g_pDataCh = zmq_socket(pContext, ZMQ_PAIR);
+    g_pDataCh = zmq_socket(pContext, ZMQ_REQ);
 
     if (g_pDataCh == NULL) {
         zmq_ctx_destroy(g_pDataCh);
@@ -91,10 +91,10 @@ int mq_data_init(DATACHNNELCB dataCb) {
 
     memset(buf, 0, 1024);
 
-    sprintf(buf, "tcp://*:%d", cfg_get_zero_mq_data_channel());
-    dzlog_info("Start message queue server: tcp://*:%d\n", cfg_get_zero_mq_data_channel());
+    sprintf(buf, "tcp://127.0.0.1:%d", cfg_get_zero_mq_data_channel());
+    dzlog_info("Start message queue connect: tcp://127.0.0.1:%d\n", cfg_get_zero_mq_data_channel());
 
-    if (zmq_bind(g_pDataCh, buf) != 0) {
+    if (zmq_connect(g_pDataCh, buf) != 0) {
         zmq_close(g_pDataCh);
         zmq_ctx_destroy(g_pDataCh);
         return -ERR_MQ_BIND_SOCKET;

srcs/vcpe_agent.c

@@ -0,0 +1,82 @@
+//
+// 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_REP);
+
+    if (g_pResponse == NULL) {
+        zmq_ctx_destroy(g_pContext);
+        return -ERR_MQ_CREATE_REP;
+    }
+
+    memset(buf, 0, 1024);
+
+    sprintf(buf, "tcp://*:6279");
+    printf("Start message data channel server: tcp://*:6279\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;
+}

srcs/vcpe_main.c

@@ -8,10 +8,8 @@
 #include "task_manager.h"
 #include "init.h"
 #include "user_info.h"
-#include "pppoe_session.h"
 #include "lwip/sys.h"
 #include "lwip/tcpip.h"
-#include "netif/rawif.h"
 
 static void test_init(void *arg) { /* remove compiler warning */
     sys_sem_t *init_sem;