GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-22 01:05:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Revert "core: using nng_send_aio"

Browse Source 
This reverts commit c7aaf234feebf0a5f69ede76953cb4be1630ea19.
This commit is contained in:
fengzero
2023-09-26 09:04:50 +00:00
parent f27b7b9da6
commit 175f23ad21
2 changed files with 16 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+8 -22
View File
@@ -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;
}
src/core/manager.c
+8 -25
View File
@@ -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);
}
}