Merge pull request #484 from fengzeroz/main

protocol buf & timer & udp connection

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;
 

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)
 {

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);
 

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;
         }

src/event/event_linux.c

@@ -33,8 +33,9 @@
 #include <sys/timerfd.h>
 
 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;