diff --git a/include/neuron/connection/neu_connection.h b/include/neuron/connection/neu_connection.h index 806baabb..26ae871c 100644 --- a/include/neuron/connection/neu_connection.h +++ b/include/neuron/connection/neu_connection.h @@ -82,8 +82,10 @@ 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; diff --git a/include/neuron/utils/protocol_buf.h b/include/neuron/utils/protocol_buf.h index 8aa14e45..ef267a59 100644 --- a/include/neuron/utils/protocol_buf.h +++ b/include/neuron/utils/protocol_buf.h @@ -122,6 +122,11 @@ inline static uint8_t *neu_protocol_pack_buf(neu_protocol_pack_buf_t *buf, return buf->base + buf->offset; } +inline static uint8_t *neu_protocol_pack_buf_get(neu_protocol_pack_buf_t *buf) +{ + return buf->base + buf->offset; +} + inline static uint16_t neu_protocol_pack_buf_unused_size(neu_protocol_pack_buf_t *buf) { diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 1b7e283f..e8712593 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -1871,10 +1871,10 @@ int neu_adapter_uninit(neu_adapter_t *adapter) { int rv = 0; - adapter->plugin_info.module->intf_funs->uninit(adapter->plugin_info.plugin); if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_uninit((neu_adapter_driver_t *) adapter); } + adapter->plugin_info.module->intf_funs->uninit(adapter->plugin_info.plugin); neu_event_del_io(adapter->events, adapter->nng_io); diff --git a/src/connection/connection.c b/src/connection/connection.c index 4bdfff6a..8d8e2c4e 100644 --- a/src/connection/connection.c +++ b/src/connection/connection.c @@ -373,7 +373,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); @@ -413,9 +414,11 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param) 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 = conn->param.params.udp.timeout > 0; + 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 = @@ -549,22 +552,41 @@ static void conn_connect(neu_conn_t *conn) } else { fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); } + int so_broadcast = 1; + setsockopt(fd, SOL_SOCKET, SO_BROADCAST, &so_broadcast, + sizeof(so_broadcast)); + struct sockaddr_in local = { + .sin_family = AF_INET, + .sin_port = htons(conn->param.params.udp.src_port), + .sin_addr.s_addr = inet_addr(conn->param.params.udp.src_ip), + }; + + ret = bind(fd, (struct sockaddr *) &local, sizeof(struct sockaddr_in)); + if (ret != 0) { + close(fd); + log_error("bind %s:%d error: %s(%d)", conn->param.params.udp.src_ip, + conn->param.params.udp.src_port, strerror(errno), errno); + conn->is_connected = false; + return; + } + struct sockaddr_in remote = { .sin_family = AF_INET, - .sin_port = htons(conn->param.params.udp.port), - .sin_addr.s_addr = inet_addr(conn->param.params.udp.ip), + .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.ip, - conn->param.params.udp.port, strerror(errno), errno); + 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.ip, - conn->param.params.udp.port); + 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; } diff --git a/src/event/event_linux.c b/src/event/event_linux.c index 35ff0f52..ce818c25 100644 --- a/src/event/event_linux.c +++ b/src/event/event_linux.c @@ -33,8 +33,9 @@ #include struct neu_event_timer { - int fd; - void *event_data; + int fd; + void * event_data; + struct itimerspec value; }; struct neu_event_io { @@ -99,7 +100,10 @@ static void *event_loop(void *arg) ssize_t size = read(data->fd, &t, sizeof(t)); assert(size != -1); + epoll_ctl(epoll_fd, EPOLL_CTL_DEL, data->fd, NULL); ret = data->callback.timer(data->usr_data); + timerfd_settime(data->fd, 0, &data->ctx.timer->value, NULL); + epoll_ctl(epoll_fd, EPOLL_CTL_ADD, data->fd, &event); } break; case IO: @@ -195,6 +199,7 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t * events, data->callback.timer = timer.cb; data->ctx.timer = timer_ctx; + timer_ctx->value = value; timer_ctx->fd = timer_fd; timer_ctx->event_data = data;