From 9eded28ee51dd0725305c95bcec14335ac017580 Mon Sep 17 00:00:00 2001 From: Parker Liu Date: Fri, 16 Jul 2021 14:35:37 +0800 Subject: [PATCH] Add some message types and reqresp types for read data from plugin --- include/neu_adapter.h | 4 + plugins/plugin_sample.c | 33 +++-- src/adapter/adapter.c | 158 ++++++++++++++++++++---- src/adapter/adapter_internal.h | 4 + src/core/databuf.c | 39 ++++-- src/core/databuf.h | 27 ++++- src/core/message.c | 3 +- src/core/message.h | 17 ++- src/core/neu_manager.c | 213 ++++++++++++++++++++++++++------- src/core/neu_manager.h | 2 - 10 files changed, 404 insertions(+), 96 deletions(-) diff --git a/include/neu_adapter.h b/include/neu_adapter.h index 307cd6fb..6aaaa949 100644 --- a/include/neu_adapter.h +++ b/include/neu_adapter.h @@ -31,8 +31,12 @@ typedef struct neu_adapter neu_adapter_t; typedef enum neu_reqresp_type { NEU_REQRESP_NOP, + NEU_REQRESP_START_PERIODIC_READ, + NEU_REQRESP_STOP_PERIODIC_READ, NEU_REQRESP_READ, NEU_REQRESP_WRITE, + // The ownership of buffer will move into reqresp. + NEU_REQRESP_MOVE_BUF, } neu_reqresp_type_e; typedef struct neu_request { diff --git a/plugins/plugin_sample.c b/plugins/plugin_sample.c index 101f276c..d7f9390d 100644 --- a/plugins/plugin_sample.c +++ b/plugins/plugin_sample.c @@ -33,8 +33,8 @@ static neu_plugin_t* sample_plugin_open(neu_adapter_t* adapter, { neu_plugin_t* plugin; - if (adapter == NULL) { - log_error("Open plugin with NULL adapter"); + if (adapter == NULL || callbacks == NULL) { + log_error("Open plugin with NULL adapter or callbacks"); return NULL; } @@ -89,20 +89,35 @@ static int sample_plugin_request(neu_plugin_t* plugin, { int rv = 0; - if (plugin == NULL) { - log_warn("The plugin pointer is NULL"); + if (plugin == NULL || req == NULL) { + log_warn("The plugin pointer or request is NULL"); return (-1); } log_info("send request to plugin: %s", neu_plugin_module.module_name); const adapter_callbacks_t* adapter_callbacks; adapter_callbacks = plugin->common.adapter_callbacks; - if (adapter_callbacks != NULL) { - neu_response_t resp; - memset(&resp, 0, sizeof(resp)); - resp.req_id = req->req_id; - rv = adapter_callbacks->response(plugin->common.adapter, &resp); + switch (req->req_type) { + case NEU_REQRESP_READ: + { + neu_response_t resp; + static const char* resp_str = "Sample plugin read response"; + + char* resp_buf; + resp_buf = strdup(resp_str); + + memset(&resp, 0, sizeof(resp)); + resp.req_id = req->req_id; + resp.resp_type = NEU_REQRESP_MOVE_BUF; + resp.buf_len = sizeof(strlen(resp_buf) + 1); + resp.buf = resp_buf; + rv = adapter_callbacks->response(plugin->common.adapter, &resp); + break; + } + + default: + break; } return rv; } diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index c1e43bc2..cb9ecd2a 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -45,6 +45,7 @@ typedef enum adapter_state { struct neu_adapter { uint32_t id; adapter_type_e type; + nng_mtx* mtx; adapter_state_e state; bool stop; char* name; @@ -52,6 +53,7 @@ struct neu_adapter { nng_pipe pipe; nng_socket sock; nng_thread* thrd; + uint32_t new_req_id; char* plugin_lib_name; void* plugin_lib; // handle of dynamic lib neu_plugin_module_t* plugin_module; @@ -89,17 +91,20 @@ static int unload_plugin_library(void* lib_handle) return rv; } +static uint32_t adapter_get_req_id(neu_adapter_t* adapter) +{ + uint32_t req_id; + + req_id = adapter->new_req_id++; + return req_id; +} + static void adapter_loop(void* arg) { int rv; neu_adapter_t* adapter; adapter = (neu_adapter_t*)arg; - rv = nng_pair1_open_poly(&adapter->sock); - if (rv != 0) { - neu_panic("The adapter(%s) can't open pipe", adapter->name); - } - const char* manager_url; manager_url = neu_manager_get_url(adapter->manager); rv = nng_dial(adapter->sock, manager_url, NULL, 0); @@ -107,15 +112,19 @@ static void adapter_loop(void* arg) neu_panic("The adapter can't dial to %s", manager_url); } + nng_mtx_lock(adapter->mtx); + adapter->state = ADAPTER_STATE_IDLE; + nng_mtx_unlock(adapter->mtx); + const char* adapter_str = "adapter started"; nng_msg* out_msg; size_t msg_size; msg_size = msg_inplace_data_get_size(strlen(adapter_str) + 1); rv = nng_msg_alloc(&out_msg, msg_size); if (rv == 0) { - void* msg_ptr; + message_t* msg_ptr; char* buf_ptr; - msg_ptr = nng_msg_body(out_msg); + msg_ptr = (message_t*)nng_msg_body(out_msg); msg_inplace_data_init(msg_ptr, MSG_EVENT_STATUS_STRING, msg_size); buf_ptr = msg_get_buf_ptr(msg_ptr); @@ -127,29 +136,69 @@ static void adapter_loop(void* arg) while (1) { nng_msg* msg; + nng_mtx_lock(adapter->mtx); + if (adapter->stop) { + adapter->state = ADAPTER_STATE_NULL; + nng_mtx_unlock(adapter->mtx); + log_info("Exit loop of the adapter(%s)", adapter->name); + break; + } + nng_mtx_unlock(adapter->mtx); + rv = nng_recvmsg(adapter->sock, &msg, 0); if (rv != 0) { - if (adapter->stop) { - break; - } else { - log_warn("Manage pipe no message received"); - continue; - } + log_warn("Manage pipe no message received"); + continue; } - message_t* neu_msg; - neu_msg = nng_msg_body(msg); - switch (msg_get_type(neu_msg)) { + message_t* pay_msg; + pay_msg = nng_msg_body(msg); + switch (msg_get_type(pay_msg)) { case MSG_CONFIG_INFO_STRING: { char* buf_ptr; - buf_ptr = msg_get_buf_ptr(neu_msg); + buf_ptr = msg_get_buf_ptr(pay_msg); log_info("Recieve string: %s", buf_ptr); break; } + case MSG_CMD_START_READ: + { + const neu_plugin_intf_funs_t* intf_funs; + neu_request_t req; + uint32_t req_code; + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = adapter_get_req_id(adapter); + req.req_type = NEU_REQRESP_READ; + req.buf_len = sizeof(uint32_t); + req_code = 1; + req.buf = (char*)&req_code; + intf_funs->request(adapter->plugin, &req); + break; + } + + case MSG_CMD_STOP_READ: + { + break; + } + + case MSG_CMD_EXIT_LOOP: + { + uint32_t exit_code; + + exit_code = *(uint32_t*)msg_get_buf_ptr(pay_msg); + log_info("adapter(%s) exit loop by exit_code=%d", + adapter->name, exit_code); + nng_mtx_lock(adapter->mtx); + adapter->state = ADAPTER_STATE_NULL; + adapter->stop = true; + nng_mtx_unlock(adapter->mtx); + break; + } + default: - log_warn("Receive a not handled message(type: %d)", msg_get_type(neu_msg)); + log_warn("Receive a not supported message(type: %d)", + msg_get_type(pay_msg)); break; } @@ -163,11 +212,44 @@ static int adapter_response(neu_adapter_t* adapter, neu_response_t* resp) { int rv = 0; + if (adapter == NULL || resp == NULL) { + log_warn("The adapter or response is NULL"); + return (-1); + } + log_info("Get response from plugin"); + switch (resp->resp_type) { + case NEU_REQRESP_MOVE_BUF: + { + core_databuf_t* databuf; + databuf = core_databuf_new_with_buf(resp->buf, resp->buf_len); + // for debug + log_debug("Get respose buf: %s", core_databuf_get_ptr(databuf)); + + nng_msg* msg; + size_t msg_size; + msg_size = msg_external_data_get_size(); + rv = nng_msg_alloc(&msg, msg_size); + if (rv == 0) { + message_t* pay_msg; + + pay_msg = (message_t*)nng_msg_body(msg); + msg_external_data_init(pay_msg, MSG_DATA_NEURON_DATABUF, + databuf); + nng_sendmsg(adapter->sock, msg, 0); + } + core_databuf_put(databuf); + break; + } + + default: + break; + } return rv; } -static int adapter_event_notify(neu_adapter_t* adapter, neu_event_notify_t* event) +static int adapter_event_notify(neu_adapter_t* adapter, + neu_event_notify_t* event) { int rv = 0; @@ -190,12 +272,20 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info) return NULL; } + int rv; + adapter->state = ADAPTER_STATE_NULL; + adapter->stop = false; + if ((rv = nng_mtx_alloc(&adapter->mtx)) != 0) { + log_error("Can't allocate mutex for adapter"); + free(adapter); + return NULL; + } + adapter->id = info->id; adapter->type = info->type; adapter->name = strdup(info->name); adapter->plugin_lib_name = strdup(info->plugin_lib_name); - adapter->state = ADAPTER_STATE_IDLE; - adapter->stop = false; + adapter->new_req_id = 0; if ( adapter->name == NULL || adapter->plugin_lib_name == NULL) { @@ -205,6 +295,7 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info) if (adapter->plugin_lib_name != NULL) { free(adapter->plugin_lib_name); } + nng_mtx_free(adapter->mtx); free(adapter); log_error("Failed duplicate string for create adapter"); return NULL; @@ -228,6 +319,11 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info) neu_panic("Can't to open plugin(%s)", plugin_module->module_name); } + rv = nng_pair1_open(&adapter->sock); + if (rv != 0) { + neu_panic("The adapter(%s) can't open pipe", adapter->name); + } + adapter->plugin = plugin; return adapter; } @@ -238,6 +334,7 @@ void neu_adapter_destroy(neu_adapter_t* adapter) neu_panic("Destroied adapter is NULL"); } + nng_close(adapter->sock); adapter->plugin_module->intf_funs->close(adapter->plugin); unload_plugin_library(adapter->plugin_lib); if (adapter->name != NULL) { @@ -246,6 +343,7 @@ void neu_adapter_destroy(neu_adapter_t* adapter) if (adapter->plugin_lib_name != NULL) { free(adapter->plugin_lib_name); } + nng_mtx_free(adapter->mtx); free(adapter); return; } @@ -272,6 +370,10 @@ int neu_adapter_stop(neu_adapter_t* adapter, neu_manager_t* manager) { int rv = 0; + log_info("Stop the adapter(%s)", adapter->name); + nng_mtx_lock(adapter->mtx); + adapter->stop = true; + nng_mtx_unlock(adapter->mtx); nng_thread_destroy(adapter->thrd); const neu_plugin_intf_funs_t* intf_funs; @@ -281,3 +383,17 @@ int neu_adapter_stop(neu_adapter_t* adapter, neu_manager_t* manager) return rv; } +const char* neu_adapter_get_name(neu_adapter_t* adapter) +{ + return (const char*)adapter->name; +} + +neu_manager_t* neu_adapter_get_manager(neu_adapter_t* adapter) +{ + return (neu_manager_t*)adapter->manager; +} + +nng_socket neu_adapter_get_sock(neu_adapter_t* adapter) +{ + return adapter->sock; +} diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 0f0b100e..cb8bb85b 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -20,6 +20,7 @@ #ifndef ADAPTER_INTERNAL_H #define ADAPTER_INTERNAL_H +#include "core/databuf.h" #include "core/neu_manager.h" typedef enum adapter_type { @@ -42,5 +43,8 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info); void neu_adapter_destroy(neu_adapter_t* adapter); int neu_adapter_start(neu_adapter_t* adapter, neu_manager_t* manager); int neu_adapter_stop(neu_adapter_t* adapter, neu_manager_t* manager); +const char* neu_adapter_get_name(neu_adapter_t* adapter); +neu_manager_t* neu_adapter_get_manager(neu_adapter_t* adapter); +nng_socket neu_adapter_get_sock(neu_adapter_t* adapter); #endif diff --git a/src/core/databuf.c b/src/core/databuf.c index b0e4f19a..9abd10cd 100644 --- a/src/core/databuf.c +++ b/src/core/databuf.c @@ -25,21 +25,23 @@ struct core_databuf { neu_atomic_int ref_count; - size_t len; - char buf[0]; + size_t len; + void* buf; }; -core_databuf_t* core_databuf_alloc(size_t size) +core_databuf_t* core_databuf_new_with_buf(void* buf, size_t len) { core_databuf_t* databuf; - databuf = (core_databuf_t*)malloc(sizeof(core_databuf_t) + size); + databuf = (core_databuf_t*)malloc(sizeof(core_databuf_t)); if (databuf == NULL) { return NULL; } neu_atomic_init(&databuf->ref_count); neu_atomic_set(&databuf->ref_count, 1); + databuf->len = len; + databuf->buf = buf; return databuf; } @@ -53,16 +55,21 @@ core_databuf_t* core_databuf_get(core_databuf_t* databuf) return databuf; } -void core_databuf_put(core_databuf_t* databuf) +core_databuf_t* core_databuf_put(core_databuf_t* databuf) { if (databuf == NULL) { - return; + return NULL; } if (neu_atomic_dec_nv(&databuf->ref_count) == 0) { + if (databuf->buf != NULL) { + free(databuf->buf); + } free(databuf); + return NULL; } - return; + + return databuf; } size_t core_databuf_get_len(core_databuf_t* databuf) @@ -74,7 +81,23 @@ size_t core_databuf_get_len(core_databuf_t* databuf) return databuf->len; } -void* core_databuf_peek_ptr(core_databuf_t* databuf) +core_databuf_t* core_databuf_set_buf(core_databuf_t* databuf, + void* buf, size_t len) +{ + if (databuf == NULL) { + return NULL; + } + + if (databuf->buf != NULL) { + free(databuf->buf); + } + + databuf->len = len; + databuf->buf = buf; + return databuf; +} + +void* core_databuf_get_ptr(core_databuf_t* databuf) { if (databuf == NULL) { return NULL; diff --git a/src/core/databuf.h b/src/core/databuf.h index 3ee36306..be26e808 100644 --- a/src/core/databuf.h +++ b/src/core/databuf.h @@ -22,17 +22,38 @@ typedef struct core_databuf core_databuf_t; +/** + * Create a shared databuf with provided buffer, the ownership of buffer will + * move into shared databuf. + * The caller don't free the buffer. + */ +core_databuf_t* core_databuf_new_with_buf(void* buf, size_t len); + /** * Get a shared databuf, it will increase reference count of databuf */ core_databuf_t* core_databuf_get(core_databuf_t* databuf); /** - * Put a shared databuf, it will decrease reference count of databuf + * Put a shared databuf, it will decrease reference count of databuf. + * If the reference count of databuf is 0, then free databuf and buffer in it. */ -void core_databuf_put(core_databuf_t* databuf); +core_databuf_t* core_databuf_put(core_databuf_t* databuf); +/** + * Get length of the buffer in databuf + */ size_t core_databuf_get_len(core_databuf_t* databuf); -void* core_databuf_peek_ptr(core_databuf_t* databuf); + +/** + * Replace original buffer pointer in databuf with new buffer pointer, the + * original buffer should be free. + */ +core_databuf_t* core_databuf_set_buf(core_databuf_t* databuf, + void* buf, size_t len); +/** + * Get a buffer pointer in databuf + */ +void* core_databuf_get_ptr(core_databuf_t* databuf); #endif diff --git a/src/core/message.c b/src/core/message.c index 658744d7..8e72aa0d 100644 --- a/src/core/message.c +++ b/src/core/message.c @@ -37,7 +37,6 @@ struct message { union { msg_inplace_buf_t inp_buf; // The external buffer pointer to a core_databuf_t. - // The one core_databuf_t is get from a buffer pool. core_databuf_t* ext_buf; }; @@ -144,7 +143,7 @@ void* msg_get_buf_ptr(message_t* msg) } if ((msg->type & MSG_DATABUF_KIND_MASK) == MSG_DATABUF_EXTERNAL) { - return core_databuf_peek_ptr(msg->ext_buf); + return core_databuf_get_ptr(msg->ext_buf); } else { return msg->inp_buf.buf_ptr; } diff --git a/src/core/message.h b/src/core/message.h index 17850da8..8e1132d9 100644 --- a/src/core/message.h +++ b/src/core/message.h @@ -28,13 +28,18 @@ typedef enum msg_type { MSG_TYPE_CMD_START = 0, MSG_CMD_NOP = MSG_TYPE_CMD_START, - MSG_CMD_START_NEURON, - MSG_CMD_STOP_NEURON, + MSG_CMD_START_READ, + MSG_CMD_STOP_READ, + MSG_CMD_EXIT_LOOP, MSG_TYPE_CMD_END, MSG_TYPE_CONFIG_START = 0x1000, - MSG_CONFIG_SUB_NEURON_DRIVER = MSG_TYPE_CONFIG_START, - MSG_CONFIG_INFO_STRING, + MSG_CONFIG_INFO_STRING= MSG_TYPE_CONFIG_START, + MSG_CONFIG_REGISTER_ADAPTER, + MSG_CONFIG_UNREGISTER_ADAPTER, + MSG_CONFIG_SUBSCRIBE_ADAPTER, + // for adapter + MSG_CONFIG_READ_ADDR_RANGE, MSG_TYPE_CONFIG_END, MSG_TYPE_EVENT_START = 0x2000, @@ -43,11 +48,11 @@ typedef enum msg_type { MSG_TYPE_EVENT_END, MSG_TYPE_DATA_START = 0x3000, - MSG_DATA_NEURON_DRIVER = MSG_TYPE_DATA_START, + MSG_DATA_NEURON_DATABUF = MSG_TYPE_DATA_START, MSG_TYPE_DATA_END, // User customer message types - MSG_TYPE_VENDOR_START = 0x10000, + MSG_TYPE_VENDOR_START = 0x100000, MSG_TYPE_VENDOR_END, MSG_TYPE_END = MSG_TYPE_VENDOR_END, diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 290d01ea..16df8971 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -34,8 +34,8 @@ typedef struct adapter_bind_info { uint32_t adapter_id; + neu_adapter_t* adapter; nng_pipe adapter_pipe; - int adapter_sock; int bind_count; } adapter_bind_info_t; @@ -52,8 +52,16 @@ typedef struct manager_bind_info { adapter_bind_info_t* bind_adapters; } manager_bind_info_t; +typedef enum manager_state { + MANAGER_STATE_NULL, + MANAGER_STATE_IDLE, + MANAGER_STATE_RUNNING, +} manager_state_e; + struct neu_manager { const char* listen_url; + nng_mtx* mtx; + manager_state_e state; bool stop; nng_thread* thrd; uint32_t new_adapter_id; @@ -118,20 +126,97 @@ static int manager_add_config(neu_manager_t* manager) return rv; } -static int manager_add_adapter(neu_manager_t* manager, neu_adapter_t* adapter) +static uint32_t manager_get_adapter_id(neu_manager_t* manager) +{ + uint32_t adapter_id; + + adapter_id = manager->new_adapter_id++; + return adapter_id; +} + +static void manager_bind_adapter(nng_pipe p, + nng_pipe_ev ev, + void* arg) +{ + int rv = 0; + neu_adapter_t* adapter; + + adapter = (neu_adapter_t*)arg; + log_info("The manager will bind adapter(%s)", + neu_adapter_get_name(adapter)); + + return; +} + +static void manager_unbind_adapter(nng_pipe p, + nng_pipe_ev ev, + void* arg) +{ + int rv = 0; + neu_adapter_t* adapter; + + adapter = (neu_adapter_t*)arg; + log_info("The manager will unbind adapter(%s)", + neu_adapter_get_name(adapter)); + + return; +} + +static int manager_add_adapter(neu_manager_t* manager, + neu_adapter_t* adapter) { int rv = 0; return rv; } -static int manager_remove_adapter(neu_manager_t* manager, neu_adapter_t* adapter) +static int manager_remove_adapter(neu_manager_t* manager, + neu_adapter_t* adapter) { int rv = 0; return rv; } +static int manager_start_adapter(neu_manager_t* manager, + neu_adapter_t* adapter) +{ + int rv = 0; + + rv = nng_pipe_notify(neu_adapter_get_sock(adapter), + NNG_PIPE_EV_ADD_POST, + manager_bind_adapter, + adapter); + neu_adapter_start(adapter, manager); + return rv; +} + +static int manager_stop_adapter(neu_manager_t* manager, + neu_adapter_t* adapter) +{ + int rv = 0; + size_t msg_size; + nng_msg* msg; + + msg_size = msg_inplace_data_get_size(sizeof(uint32_t)); + rv = nng_msg_alloc(&msg, msg_size); + if (rv != 0) { + neu_panic("Can't allocate msg for stop adapter(%s)", + neu_adapter_get_name(adapter)); + } + + void* buf_ptr; + message_t* pay_msg; + pay_msg = (message_t*)nng_msg_body(msg); + msg_inplace_data_init(pay_msg, MSG_CMD_EXIT_LOOP, sizeof(uint32_t)); + buf_ptr = msg_get_buf_ptr(pay_msg); + *(uint32_t*)buf_ptr = 0; // exit_code is 0 + manager_bind_info_t* manager_bind = &manager->bind_info; + nng_sendmsg(manager_bind->mng_sock, msg, 0); + neu_adapter_stop(adapter, manager); + return rv; +} + static void manager_loop(void* arg) { int rv; @@ -149,33 +234,41 @@ static void manager_loop(void* arg) neu_panic("Neuron manager can't listen on %s", manager->listen_url); } + nng_mtx_lock(manager->mtx); + manager->state = MANAGER_STATE_RUNNING; + nng_mtx_unlock(manager->mtx); + #ifdef NEU_HAS_SAMPLE_ADAPTER - sample_adapter_info.id = manager->new_adapter_id; - manager->new_adapter_id++; + sample_adapter_info.id = manager_get_adapter_id(manager); sample_adapter = neu_adapter_create(&sample_adapter_info); - neu_adapter_start(sample_adapter, manager); + manager_start_adapter(manager, sample_adapter); #endif while (1) { nng_msg* msg; + nng_mtx_lock(manager->mtx); + if (manager->stop) { + manager->state = MANAGER_STATE_NULL; + nng_mtx_unlock(manager->mtx); + log_info("Exit loop of the manager"); + break; + } + nng_mtx_unlock(manager->mtx); + rv = nng_recvmsg(manager_bind->mng_sock, &msg, 0); if (rv != 0) { - if (manager->stop) { - break; - } else { - log_warn("Manage pipe no message received"); - continue; - } + log_warn("Manage pipe no message received"); + continue; } - message_t* neu_msg; - neu_msg = nng_msg_body(msg); - switch (msg_get_type(neu_msg)) { + message_t* pay_msg; + pay_msg = nng_msg_body(msg); + switch (msg_get_type(pay_msg)) { case MSG_EVENT_STATUS_STRING: { char* buf_ptr; - buf_ptr = msg_get_buf_ptr(neu_msg); + buf_ptr = msg_get_buf_ptr(pay_msg); log_info("Recieve string: %s", buf_ptr); const char* adapter_str = "manager recv reply"; @@ -184,21 +277,65 @@ static void manager_loop(void* arg) msg_size = msg_inplace_data_get_size(strlen(adapter_str) + 1); rv = nng_msg_alloc(&out_msg, msg_size); if (rv == 0) { - void* msg_ptr; + message_t* msg_ptr; char* buf_ptr; - msg_ptr = nng_msg_body(out_msg); + msg_ptr = (message_t*)nng_msg_body(out_msg); msg_inplace_data_init(msg_ptr, MSG_CONFIG_INFO_STRING, - msg_size); + strlen(adapter_str) + 1); buf_ptr = msg_get_buf_ptr(msg_ptr); memcpy(buf_ptr, adapter_str, strlen(adapter_str)); buf_ptr[strlen(adapter_str)] = 0; nng_sendmsg(manager_bind->mng_sock, out_msg, 0); } + + // for debug + usleep(1000); + nng_msg* read_msg; + msg_size = msg_inplace_data_get_size(sizeof(uint32_t)); + rv = nng_msg_alloc(&read_msg, msg_size); + if (rv == 0) { + message_t* msg_ptr; + void* buf_ptr; + + msg_ptr = (message_t*)nng_msg_body(read_msg); + msg_inplace_data_init(msg_ptr, MSG_CMD_START_READ, + sizeof(uint32_t)); + buf_ptr = msg_get_buf_ptr(msg_ptr); + *(uint32_t*)buf_ptr = 3; + nng_sendmsg(manager_bind->mng_sock, read_msg, 0); + } + break; + } + + case MSG_CMD_EXIT_LOOP: + { + uint32_t exit_code; + + exit_code = *(uint32_t*)msg_get_buf_ptr(pay_msg); + log_info("manager exit loop by exit_code=%d", exit_code); + nng_mtx_lock(manager->mtx); + manager->state = MANAGER_STATE_NULL; + manager->stop = true; + nng_mtx_unlock(manager->mtx); + break; + } + + case MSG_DATA_NEURON_DATABUF: + { + core_databuf_t* databuf; + + databuf = (core_databuf_t*)msg_get_buf_ptr(pay_msg); + + log_info("dispatch databuf to subscribes"); + // TODO: dispatch the databuf to subsribes + + core_databuf_put(databuf); break; } default: - log_warn("Receive a not handled message(type: %d)", msg_get_type(neu_msg)); + log_warn("Receive a not supported message(type: %d)", + msg_get_type(pay_msg)); break; } @@ -206,9 +343,10 @@ static void manager_loop(void* arg) } #ifdef NEU_HAS_SAMPLE_ADAPTER - neu_adapter_stop(sample_adapter, manager); + manager_stop_adapter(manager, sample_adapter); neu_adapter_destroy(sample_adapter); #endif + nng_close(manager_bind->mng_sock); return; } @@ -222,9 +360,17 @@ neu_manager_t* neu_manager_create() return NULL; } + manager->state = MANAGER_STATE_NULL; manager->stop = false; manager->listen_url = manager_url; manager->new_adapter_id = 1; + int rv; + if ((rv = nng_mtx_alloc(&manager->mtx)) != 0) { + log_error("Can't allocate mutex for manager"); + free(manager); + return NULL; + } + init_bind_info(&manager->bind_info); nng_thread_create(&manager->thrd, manager_loop, manager); return manager; @@ -233,35 +379,12 @@ neu_manager_t* neu_manager_create() void neu_manager_destroy(neu_manager_t* manager) { nng_thread_destroy(manager->thrd); + nng_mtx_free(manager->mtx); uninit_bind_info(&manager->bind_info); free(manager); return; } -int neu_manager_bind_adapter(neu_manager_t* manager, neu_adapter_t* adapter) -{ - int rv = 0; - - if (adapter == NULL) { - log_warn("The adapter is NULL that want to bind"); - return (-1); - } - - return rv; -} - -int neu_manager_unbind_adapter(neu_manager_t* manager, neu_adapter_t* adapter) -{ - int rv = 0; - - if (adapter == NULL) { - log_warn("The adapter is NULL that want to unbind"); - return (-1); - } - - return rv; -} - const char* neu_manager_get_url(neu_manager_t* manager) { if (manager == NULL) { diff --git a/src/core/neu_manager.h b/src/core/neu_manager.h index b70747e8..ba416bda 100644 --- a/src/core/neu_manager.h +++ b/src/core/neu_manager.h @@ -28,8 +28,6 @@ typedef struct neu_manager neu_manager_t; neu_manager_t* neu_manager_create(); void neu_manager_destroy(neu_manager_t* manager); -int neu_manager_bind_adapter(neu_manager_t* manager, neu_adapter_t* adapter); -int neu_manager_unbind_adapter(neu_manager_t* manager, neu_adapter_t* adapter); const char* neu_manager_get_url(neu_manager_t* manager); #endif