diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 98031ec0..7d694bb4 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -111,26 +111,6 @@ static inline void stop_log_level_timer(neu_adapter_t *adapter) adapter->timer_lev = NULL; } -inline static int send_high_msg(nng_socket socket, nng_msg *msg) -{ - nng_aio *aio = NULL; - int ret = 0; - - nng_aio_alloc(&aio, NULL, NULL); - nng_aio_set_timeout(aio, 3000); - nng_aio_set_msg(aio, msg); - - nng_send_aio(socket, aio); - nng_aio_wait(aio); - if (nng_aio_result(aio) != 0) { - nlog_warn("send msg faile"); - ret = -1; - } - nng_aio_free(aio); - - return ret; -} - int neu_adapter_error() { return create_adapter_error; @@ -471,7 +451,10 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, msg = neu_msg_gen(&header, data); - ret = send_high_msg(adapter->sock, msg); + ret = nng_sendmsg(adapter->sock, msg, 0); + if (ret != 0) { + nng_msg_free(msg); + } return ret; } @@ -489,7 +472,10 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, } nng_msg *msg = neu_msg_gen(header, data); - int ret = send_high_msg(adapter->sock, msg); + int ret = nng_sendmsg(adapter->sock, msg, 0); + if (ret != 0) { + nng_msg_free(msg); + } return ret; } diff --git a/src/core/manager.c b/src/core/manager.c index ca173f41..5567cc25 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -1201,26 +1201,6 @@ inline static void trans_data_free(nng_msg *msg) } } -inline static int send_high_msg(nng_socket socket, nng_msg *msg) -{ - nng_aio *aio = NULL; - int ret = 0; - - nng_aio_alloc(&aio, NULL, NULL); - nng_aio_set_timeout(aio, 3000); - nng_aio_set_msg(aio, msg); - - nng_send_aio(socket, aio); - nng_aio_wait(aio); - if (nng_aio_result(aio) != 0) { - nlog_warn("send msg faile"); - ret = -1; - } - nng_aio_free(aio); - - return ret; -} - inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *node) { @@ -1229,11 +1209,11 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, nng_msg_dup(&out_msg, msg); nng_msg_set_pipe(out_msg, pipe); - - if (send_high_msg(manager->socket, msg) == 0) { + if (nng_sendmsg(manager->socket, out_msg, 0) == 0) { nlog_info("forward msg to %s", node); } else { nlog_warn("forward msg to pipe (%s)%d fail", node, pipe.id); + nng_msg_free(out_msg); } } @@ -1381,9 +1361,12 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, neu_node_manager_get_pipe(manager->node_manager, header->receiver); nng_msg_set_pipe(msg, pipe); - if (send_high_msg(manager->socket, msg) != 0) { - nlog_warn("reply %s to %s fail", neu_reqresp_type_string(header->type), - header->receiver); + int ret = nng_sendmsg(manager->socket, msg, 0); + + if (ret != 0) { + nng_msg_free(msg); + nlog_warn("reply %s to %s, error: %d", + neu_reqresp_type_string(header->type), header->receiver, ret); } }