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

Merge pull request #635 from fengzeroz/main

Browse Source 
fix memory leak & feat(connection): wait all msg
This commit is contained in:
fengzeroz
2022-07-13 17:46:42 +08:00
committed by GitHub
parent d664431a64 a93796aa0c
commit d8e9297e07
11 changed files with 108 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/resource/error.resource
+3 -2
View File
@@ -60,5 +60,6 @@ ${NEU_ERR_PLUGIN_DISCONNECTED} 3002
${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ} 3003
${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE} 3004
${NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH} 3007
${NEU_ERR_PLUGIN_TAG_EXPIRED} 3008
${NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE} 3009,
${NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED} 3008
${NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE} 3009
${NEU_ERR_PLUGIN_NOT_RUNNING} 3010
include/neuron/connection/neu_connection.h
+7
View File
@@ -148,4 +148,11 @@ void neu_conn_stream_consume(neu_conn_t *conn, void *context,
void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
neu_conn_stream_consume_fn fn);
typedef neu_buf_result_t (*neu_conn_process_msg)(
void *context, neu_protocol_unpack_buf_t *protocol_buf);
int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte,
neu_conn_process_msg fn);
int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context,
uint16_t n_byte, neu_conn_process_msg fn);
#endif
include/neuron/errcodes.h
+2 -1
View File
@@ -91,8 +91,9 @@ typedef enum {
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ = 3003,
NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE = 3004,
NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH = 3007,
NEU_ERR_PLUGIN_TAG_EXPIRED = 3008,
NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED = 3008,
NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE = 3009,
NEU_ERR_PLUGIN_NOT_RUNNING = 3010,
NEU_ERR_MQTT_FAILURE = 4000,
NEU_ERR_MQTT_NO_CERTFILESET = 4001,
include/neuron/tag.h
+21 -10
View File
@@ -27,22 +27,14 @@ config_ **/
#include "type.h"
#define NEU_TAG_META_SIZE 20
typedef enum {
NEU_ATTRIBUTE_READ = 1,
NEU_ATTRIBUTE_WRITE = 2,
NEU_ATTRIBUTE_SUBSCRIBE = 4,
} neu_attribute_e;
typedef struct {
char * name;
char * address;
char * description;
neu_attribute_e attribute;
neu_type_e type;
} neu_datatag_t;
UT_icd *neu_tag_get_icd();
typedef enum {
NEU_DATATAG_ENDIAN_L16 = 0, // #L
NEU_DATATAG_ENDIAN_B16 = 1, // #B
@@ -72,10 +64,29 @@ typedef union {
} string;
struct {
bool op;
uint8_t bit;
} bit;
} neu_datatag_addr_option_u;
typedef struct {
char * name;
char * address;
char * description;
neu_attribute_e attribute;
neu_type_e type;
neu_datatag_addr_option_u option;
uint8_t meta[NEU_TAG_META_SIZE];
} neu_datatag_t;
inline static bool neu_tag_attribute_test(neu_datatag_t * tag,
neu_attribute_e attribute)
{
return (tag->attribute & attribute) == attribute;
}
UT_icd *neu_tag_get_icd();
int neu_datatag_parse_addr_option(neu_datatag_t * datatag,
neu_datatag_addr_option_u *option);
include/neuron/utils/protocol_buf.h
+5
View File
@@ -24,6 +24,11 @@ config_ **/
#include <stdint.h>
#include <stdlib.h>
typedef struct {
int used;
uint16_t need;
} neu_buf_result_t;
typedef struct {
uint8_t *base;
uint16_t size;
plugins/restful/http.c
+1
View File
@@ -285,6 +285,7 @@ int http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_GROUP_ALREADY_SUBSCRIBED:
case NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH:
case NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE:
case NEU_ERR_PLUGIN_NOT_RUNNING:
status = NNG_HTTP_STATUS_OK;
break;
case NEU_ERR_EINTERNAL:
src/adapter/adapter.c
+1 -1
View File
@@ -85,7 +85,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
rv = nng_pair1_open(&adapter->sock);
assert(rv == 0);
nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 100);
nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 1000);
nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd);
switch (info->module->type) {
src/adapter/driver/driver.c
+8 -4
View File
@@ -234,7 +234,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
int index = utarray_eltidx(tags, tag);
strcpy(resp.tags[index].tag, tag->name);
resp.tags[index].value.type = NEU_TYPE_ERROR;
resp.tags[index].value.value.i32 = NEU_ERR_NODE_NOT_RUNNING;
resp.tags[index].value.value.i32 = NEU_ERR_PLUGIN_NOT_RUNNING;
}
} else {
@@ -257,8 +257,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
neu_reqresp_head_t * req)
{
if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
NEU_ERR_NODE_NOT_RUNNING);
driver->adapter.cb_funs.driver.write_response(
&driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING);
return;
}
@@ -284,6 +284,10 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
if ((tag->attribute & NEU_ATTRIBUTE_WRITE) != NEU_ATTRIBUTE_WRITE) {
driver->adapter.cb_funs.driver.write_response(
&driver->adapter, req, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE);
free(tag->address);
free(tag->name);
free(tag->description);
free(tag);
return;
}
switch (tag->type) {
@@ -678,7 +682,7 @@ static void read_group(int64_t timestamp, int64_t timeout,
if ((timestamp - value.timestamp) > timeout) {
datas[index].value.type = NEU_TYPE_ERROR;
datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_EXPIRED;
datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED;
} else {
datas[index].value.type = tag->type;
datas[index].value.value = value.value;
src/base/group.c
+4
View File
@@ -119,6 +119,8 @@ int neu_group_add_tag(neu_group_t *group, neu_datatag_t *tag)
el->tag->description = strdup(tag->description);
el->tag->type = tag->type;
el->tag->attribute = tag->attribute;
el->tag->option = tag->option;
memcpy(el->tag->meta, tag->meta, sizeof(tag->meta));
HASH_ADD_STR(group->tags, name, el);
update_timestamp(group);
@@ -149,6 +151,8 @@ int neu_group_update_tag(neu_group_t *group, neu_datatag_t *tag)
el->tag->description = strdup(tag->description);
el->tag->type = tag->type;
el->tag->attribute = tag->attribute;
el->tag->option = tag->option;
memcpy(el->tag->meta, tag->meta, sizeof(tag->meta));
HASH_ADD_STR(group->tags, name, el);
update_timestamp(group);
src/base/tag.c
+7 -2
View File
@@ -29,8 +29,10 @@ static void tag_array_copy(void *_dst, const void *_src)
neu_datatag_t *dst = (neu_datatag_t *) _dst;
neu_datatag_t *src = (neu_datatag_t *) _src;
dst->type = src->type;
dst->attribute = src->attribute;
dst->type = src->type;
dst->attribute = src->attribute;
dst->option = src->option;
memcpy(dst->meta, src->meta, sizeof(src->meta));
dst->address = strdup(src->address);
dst->name = strdup(src->name);
dst->description = strdup(src->description);
@@ -164,6 +166,9 @@ int neu_datatag_parse_addr_option(neu_datatag_t * datatag,
if (op != NULL) {
sscanf(op, ".%hhd", &option->bit.bit);
option->bit.op = true;
} else {
option->bit.op = false;
}
break;
src/connection/connection.c
+49
View File
@@ -824,4 +824,53 @@ void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
offset);
}
}
}
int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte,
neu_conn_process_msg fn)
{
uint8_t recv_buf[2048] = { 0 };
uint16_t offset = 0;
ssize_t ret = neu_conn_recv(conn, recv_buf, n_byte);
neu_protocol_unpack_buf_t pbuf = { 0 };
while (ret > 0) {
zlog_recv_protocol(conn->param.log, recv_buf + offset, ret);
offset += ret;
neu_protocol_unpack_buf_init(&pbuf, recv_buf, offset);
neu_buf_result_t result = fn(context, &pbuf);
if (result.need > 0) {
assert(result.need <= sizeof(recv_buf) - offset);
ret = neu_conn_recv(conn, recv_buf + offset, result.need);
} else {
return result.used;
}
}
return ret;
}
int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context,
uint16_t n_byte, neu_conn_process_msg fn)
{
uint8_t recv_buf[2048] = { 0 };
uint16_t offset = 0;
ssize_t ret = neu_conn_tcp_server_recv(conn, fd, recv_buf, n_byte);
neu_protocol_unpack_buf_t pbuf = { 0 };
while (ret > 0) {
zlog_recv_protocol(conn->param.log, recv_buf + offset, ret);
offset += ret;
neu_protocol_unpack_buf_init(&pbuf, recv_buf, offset);
neu_buf_result_t result = fn(context, &pbuf);
if (result.need > 0) {
assert(result.need <= sizeof(recv_buf) - offset);
ret = neu_conn_tcp_server_recv(conn, fd, recv_buf + offset,
result.need);
} else {
return result.used;
}
}
return ret;
}