GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-19 23:02:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #456 from fengzeroz/main

Browse Source 
feat(connection): add udp
This commit is contained in:
fengzeroz
2022-04-28 11:38:32 +08:00
committed by GitHub
parent 57de38e2fb 09cca25975
commit f96d537807
3 changed files with 286 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/neu_connection.h
+40 -11
View File
@@ -23,6 +23,8 @@
#include <stdint.h>
#include <unistd.h>
#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
include/neuron/utils/protocol_buf.h
+133
View File
@@ -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 <memory.h>
#include <stdint.h>
#include <stdlib.h>
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
src/connection/connection.c
+113 -57
View File
@@ -24,6 +24,7 @@
#include <stdbool.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#ifndef SOCK_NONBLOCK
#include <fcntl.h>
@@ -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);
}
}
}
}