diff --git a/include/neuron/connection/neu_connection.h b/include/neuron/connection/neu_connection.h index bead001f..26ae871c 100644 --- a/include/neuron/connection/neu_connection.h +++ b/include/neuron/connection/neu_connection.h @@ -23,6 +23,8 @@ #include #include +#include "utils/protocol_buf.h" + typedef enum neu_conn_type { NEU_CONN_TCP_SERVER = 1, NEU_CONN_TCP_CLIENT, @@ -80,8 +82,11 @@ typedef struct neu_conn_param { } tcp_client; struct { - char * ip; - uint16_t port; + char * src_ip; + uint16_t src_port; + char * dst_ip; + uint16_t dst_port; + uint16_t timeout; // millisecond } udp; struct { @@ -96,15 +101,13 @@ typedef struct neu_conn_param { typedef struct neu_conn neu_conn_t; -neu_conn_t * neu_conn_new(neu_conn_param_t *param, void *data, - neu_conn_callback connected, - neu_conn_callback disconnected); -neu_conn_t * neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param); -void neu_conn_destory(neu_conn_t *conn); -neu_conn_type_e neu_conn_get_type(neu_conn_t *conn); -int neu_conn_get_fd(neu_conn_t *conn); -int neu_conn_tcp_server_accept(neu_conn_t *conn); -int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd); +neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data, + neu_conn_callback connected, + neu_conn_callback disconnected); +neu_conn_t *neu_conn_reconfig(neu_conn_t *conn, neu_conn_param_t *param); +void neu_conn_destory(neu_conn_t *conn); +int neu_conn_tcp_server_accept(neu_conn_t *conn); +int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd); void neu_conn_flush(neu_conn_t *conn); ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len); @@ -116,4 +119,30 @@ ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf, ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf, ssize_t len); +#define NEU_STREAM_CONSUME(buf, size) \ + neu_protocol_unpack_buf_t protocol_buf = { 0 }; \ + neu_protocol_unpack_buf_init(&protocol_buf, (buf), (size)); \ + while (neu_protocol_unpack_buf_unused_size(&protocol_buf) > 0) + +#define NEU_STREAM_CONSUME_USED_SIZE(used) \ + offset -= used; \ + memmove(recv_buf, recv_buf + used, offset); + +#define NEU_TCP_CLIENT_CONSUME(conn) \ + static __thread uint8_t recv_buf[2048] = { 0 }; \ + static __thread uint16_t offset = 0; \ + ssize_t ret = \ + neu_conn_recv((conn), recv_buf + offset, sizeof(recv_buf) - offset); \ + if (ret > 0) \ + offset += ret; \ + if (ret > 0) \ + NEU_STREAM_CONSUME(recv_buf, offset) + +typedef int (*neu_conn_stream_consume_fn)( + void *context, neu_protocol_unpack_buf_t *protocol_buf); +void neu_conn_stream_consume(neu_conn_t *conn, void *context, + neu_conn_stream_consume_fn fn); +void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context, + neu_conn_stream_consume_fn fn); + #endif diff --git a/include/neuron/utils/protocol_buf.h b/include/neuron/utils/protocol_buf.h new file mode 100644 index 00000000..d9f8628a --- /dev/null +++ b/include/neuron/utils/protocol_buf.h @@ -0,0 +1,133 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. +config_ **/ + +#ifndef _NEU_PROTOCOL_BUF_H_ +#define _NEU_PROTOCOL_BUF_H_ + +#include +#include +#include + +typedef struct { + uint8_t *base; + uint16_t size; + uint16_t offset; +} neu_protocol_buf_t, neu_protocol_pack_buf_t, neu_protocol_unpack_buf_t; + +inline static void neu_protocol_pack_buf_init(neu_protocol_pack_buf_t *buf, + uint8_t *base, uint16_t size) +{ + buf->base = base; + buf->size = size; + buf->offset = size; +} + +inline static void neu_protocol_unpack_buf_init(neu_protocol_unpack_buf_t *buf, + uint8_t *base, uint16_t size) +{ + buf->base = base; + buf->size = size; + buf->offset = 0; +} + +inline static neu_protocol_unpack_buf_t * +neu_protocol_unpack_buf_new(uint16_t size) +{ + neu_protocol_unpack_buf_t *buf = + calloc(1, sizeof(neu_protocol_unpack_buf_t)); + + buf->base = calloc(size, sizeof(uint8_t)); + buf->size = size; + buf->offset = 0; + + return buf; +} + +inline static neu_protocol_pack_buf_t *neu_protocol_pack_buf_new(uint16_t size) +{ + neu_protocol_pack_buf_t *buf = calloc(1, sizeof(neu_protocol_pack_buf_t)); + + buf->base = calloc(size, sizeof(uint8_t)); + buf->size = size; + buf->offset = size; + + return buf; +} + +inline static void neu_protocol_buf_free(neu_protocol_buf_t *buf) +{ + free(buf->base); + free(buf); +} + +inline static void neu_protocol_unpack_buf_reset(neu_protocol_buf_t *buf) +{ + buf->offset = 0; + memset(buf->base, 0, buf->size); +} + +inline static void neu_protocol_pack_buf_reset(neu_protocol_buf_t *buf) +{ + buf->offset = buf->size; + memset(buf->base, 0, buf->size); +} + +inline static uint8_t *neu_protocol_unpack_buf(neu_protocol_unpack_buf_t *buf, + uint16_t size) +{ + if (buf->size - buf->offset < size) { + return NULL; + } + + buf->offset += size; + + return buf->base + buf->offset - size; +} + +inline static uint8_t *neu_protocol_pack_buf(neu_protocol_pack_buf_t *buf, + uint16_t size) +{ + if (buf->offset < size) { + return NULL; + } + + buf->offset -= size; + + return buf->base + buf->offset; +} + +inline static uint16_t +neu_protocol_pack_buf_unused_size(neu_protocol_pack_buf_t *buf) +{ + return buf->offset; +} + +inline static uint16_t +neu_protocol_unpack_buf_unused_size(neu_protocol_unpack_buf_t *buf) +{ + return buf->size - buf->offset; +} + +inline static uint16_t +neu_protocol_unpack_buf_used_size(neu_protocol_unpack_buf_t *buf) +{ + return buf->offset; +} + +#endif \ No newline at end of file diff --git a/src/connection/connection.c b/src/connection/connection.c index 017bd660..ff045034 100644 --- a/src/connection/connection.c +++ b/src/connection/connection.c @@ -24,6 +24,7 @@ #include #include #include +#include #ifndef SOCK_NONBLOCK #include @@ -123,35 +124,6 @@ void neu_conn_destory(neu_conn_t *conn) free(conn); } -neu_conn_type_e neu_conn_get_type(neu_conn_t *conn) -{ - neu_conn_type_e type = { 0 }; - pthread_mutex_lock(&conn->mtx); - type = conn->param.type; - pthread_mutex_unlock(&conn->mtx); - - return type; -} - -int neu_conn_get_fd(neu_conn_t *conn) -{ - int fd = -1; - - pthread_mutex_lock(&conn->mtx); - if (conn->param.type == NEU_CONN_TCP_SERVER) { - if (conn->tcp_server.is_listen) { - fd = conn->fd; - } - } else { - if (conn->is_connected) { - fd = conn->fd; - } - } - pthread_mutex_unlock(&conn->mtx); - - return fd; -} - int neu_conn_tcp_server_accept(neu_conn_t *conn) { struct sockaddr_in client = { 0 }; @@ -259,7 +231,6 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf, ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len) { ssize_t ret = 0; - int fd = 0; pthread_mutex_lock(&conn->mtx); @@ -270,15 +241,14 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len) assert(1 == 0); break; case NEU_CONN_TCP_CLIENT: - fd = conn->fd; + case NEU_CONN_UDP: ret = send(conn->fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT); if (ret != len) { - log_error("tcp client fd: %d, send buf len: %d, ret: %d, " + log_error("conn fd: %d, send buf len: %d, ret: %d, " "errno: %s(%d)", conn->fd, len, ret, strerror(errno), errno); } break; - case NEU_CONN_UDP: case NEU_CONN_TTY_CLIENT: break; } @@ -286,13 +256,13 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len) if (ret == -1 && errno != EAGAIN) { conn_disconnect(conn); if (conn->callback_trigger == true) { - conn->disconnected(conn->data, fd); + conn->disconnected(conn->data, conn->fd); conn->callback_trigger = false; } } if (ret > 0 && conn->callback_trigger == false) { - conn->connected(conn->data, fd); + conn->connected(conn->data, conn->fd); conn->callback_trigger = true; } @@ -308,7 +278,6 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len) ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len) { ssize_t ret = 0; - int fd = 0; pthread_mutex_lock(&conn->mtx); @@ -318,7 +287,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len) assert(1 == 0); break; case NEU_CONN_TCP_CLIENT: - fd = conn->fd; + case NEU_CONN_UDP: if (conn->block) { ret = recv(conn->fd, buf, len, MSG_WAITALL); } else { @@ -330,7 +299,6 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len) conn->fd, len, ret, strerror(errno), errno); } break; - case NEU_CONN_UDP: case NEU_CONN_TTY_CLIENT: break; } @@ -338,13 +306,13 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len) if (errno == EPIPE || ret <= 0) { conn_disconnect(conn); if (conn->callback_trigger == true) { - conn->disconnected(conn->data, fd); + conn->disconnected(conn->data, conn->fd); conn->callback_trigger = false; } } if (ret > 0 && conn->callback_trigger == false) { - conn->connected(conn->data, fd); + conn->connected(conn->data, conn->fd); conn->callback_trigger = true; } @@ -375,13 +343,13 @@ void neu_conn_flush(neu_conn_t *conn) assert(1 == 0); break; case NEU_CONN_TCP_CLIENT: + case NEU_CONN_UDP: while (ret > 0) { pthread_mutex_lock(&conn->mtx); ret = recv(conn->fd, buf, sizeof(buf), 0); pthread_mutex_unlock(&conn->mtx); } break; - case NEU_CONN_UDP: case NEU_CONN_TTY_CLIENT: break; } @@ -399,7 +367,8 @@ static void conn_free_param(neu_conn_t *conn) free(conn->param.params.tcp_client.ip); break; case NEU_CONN_UDP: - free(conn->param.params.udp.ip); + free(conn->param.params.udp.src_ip); + free(conn->param.params.udp.dst_ip); break; case NEU_CONN_TTY_CLIENT: free(conn->param.params.tty_client.device); @@ -436,16 +405,14 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param) conn->param.params.tcp_client.port = param->params.tcp_client.port; conn->param.params.tcp_client.timeout = param->params.tcp_client.timeout; - if (conn->param.params.tcp_client.timeout > 0) { - conn->block = true; - } else { - conn->block = false; - } + conn->block = conn->param.params.tcp_client.timeout > 0; break; case NEU_CONN_UDP: - conn->param.params.udp.ip = strdup(param->params.udp.ip); - conn->param.params.udp.port = param->params.udp.port; - conn->block = false; + conn->param.params.udp.src_ip = strdup(param->params.udp.src_ip); + conn->param.params.udp.src_port = param->params.udp.src_port; + conn->param.params.udp.dst_ip = strdup(param->params.udp.dst_ip); + conn->param.params.udp.dst_port = param->params.udp.dst_port; + conn->block = conn->param.params.udp.timeout > 0; break; case NEU_CONN_TTY_CLIENT: conn->param.params.tty_client.device = @@ -524,17 +491,17 @@ static void conn_tcp_server_stop(neu_conn_t *conn) static void conn_connect(neu_conn_t *conn) { int ret = 0; + int fd = 0; switch (conn->param.type) { case NEU_CONN_TCP_SERVER: break; case NEU_CONN_TCP_CLIENT: { - int fd = 0; if (conn->block) { struct timeval tv = { - .tv_sec = conn->param.params.tcp_server.timeout / 1000, + .tv_sec = conn->param.params.tcp_client.timeout / 1000, .tv_usec = - (conn->param.params.tcp_server.timeout % 1000) * 1000, + (conn->param.params.tcp_client.timeout % 1000) * 1000, }; fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); @@ -566,8 +533,41 @@ static void conn_connect(neu_conn_t *conn) } break; } - case NEU_CONN_UDP: + case NEU_CONN_UDP: { + if (conn->block) { + struct timeval tv = { + .tv_sec = conn->param.params.udp.timeout / 1000, + .tv_usec = (conn->param.params.udp.timeout % 1000) * 1000, + }; + + fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); + } else { + fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); + } + struct sockaddr_in remote = { + .sin_family = AF_INET, + .sin_port = htons(conn->param.params.udp.dst_port), + .sin_addr.s_addr = inet_addr(conn->param.params.udp.dst_ip), + }; + ret = connect(fd, (struct sockaddr *) &remote, + sizeof(struct sockaddr_in)); + if (ret != 0 && errno != EINPROGRESS) { + close(fd); + log_error("connect %s:%d error: %s(%d)", + conn->param.params.udp.dst_ip, + conn->param.params.udp.dst_port, strerror(errno), errno); + conn->is_connected = false; + return; + } else { + log_info("connect %s:%d success", conn->param.params.udp.dst_ip, + conn->param.params.udp.dst_port); + conn->is_connected = true; + conn->fd = fd; + } break; + } case NEU_CONN_TTY_CLIENT: break; } @@ -584,9 +584,8 @@ static void conn_disconnect(neu_conn_t *conn) } break; case NEU_CONN_TCP_CLIENT: - close(conn->fd); - break; case NEU_CONN_UDP: + close(conn->fd); break; case NEU_CONN_TTY_CLIENT: break; @@ -623,5 +622,62 @@ static void conn_tcp_server_del_client(neu_conn_t *conn, int fd) } assert(1 == 0); - return; +} + +void neu_conn_stream_consume(neu_conn_t *conn, void *context, + neu_conn_stream_consume_fn fn) +{ + static __thread uint8_t recv_buf[2048] = { 0 }; + static __thread uint16_t offset = 0; + ssize_t ret = + neu_conn_recv(conn, recv_buf + offset, sizeof(recv_buf) - offset); + if (ret > 0) { + offset += ret; + neu_protocol_unpack_buf_t protocol_buf = { 0 }; + neu_protocol_unpack_buf_init(&protocol_buf, recv_buf, offset); + while (neu_protocol_unpack_buf_unused_size(&protocol_buf) > 0) { + int used = fn(context, &protocol_buf); + + if (used == 0) { + break; + } else if (used == -1) { + memset(recv_buf, 0, sizeof(recv_buf)); + offset = 0; + break; + } else { + assert(used > 0); + offset -= used; + memmove(recv_buf, recv_buf + used, offset); + } + } + } +} + +void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context, + neu_conn_stream_consume_fn fn) +{ + static __thread uint8_t recv_buf[2048] = { 0 }; + static __thread uint16_t offset = 0; + ssize_t ret = neu_conn_tcp_server_recv(conn, fd, recv_buf + offset, + sizeof(recv_buf) - offset); + if (ret > 0) { + offset += ret; + neu_protocol_unpack_buf_t protocol_buf = { 0 }; + neu_protocol_unpack_buf_init(&protocol_buf, recv_buf, offset); + while (neu_protocol_unpack_buf_unused_size(&protocol_buf) > 0) { + int used = fn(context, &protocol_buf); + + if (used == 0) { + break; + } else if (used == -1) { + memset(recv_buf, 0, sizeof(recv_buf)); + offset = 0; + break; + } else { + assert(used > 0); + offset -= used; + memmove(recv_buf, recv_buf + used, offset); + } + } + } } \ No newline at end of file