diff --git a/include/neuron/event/event.h b/include/neuron/event/event.h index 932846c1..d2eedb63 100644 --- a/include/neuron/event/event.h +++ b/include/neuron/event/event.h @@ -26,43 +26,42 @@ extern "C" { #endif -typedef struct neu_event_ctx neu_event_ctx_t; +typedef struct neu_events neu_events_t; -neu_event_ctx_t *neu_event_new(void); -int neu_event_close(neu_event_ctx_t *events); +neu_events_t *neu_event_new(void); +int neu_event_close(neu_events_t *events); -typedef struct neu_event_timer_ctx neu_event_timer_ctx_t; +typedef struct neu_event_timer neu_event_timer_t; typedef int (*neu_event_timer_callback)(void *usr_data); -typedef struct neu_event_timer { +typedef struct neu_event_timer_param { int64_t second; int64_t millisecond; void * usr_data; neu_event_timer_callback cb; -} neu_event_timer_t; +} neu_event_timer_param_t; -neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, - neu_event_timer_t timer); -int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer); +neu_event_timer_t *neu_event_add_timer(neu_events_t * events, + neu_event_timer_param_t timer); +int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer); enum neu_event_io_type { NEU_EVENT_IO_READ = 0x1, NEU_EVENT_IO_CLOSED = 0x2, NEU_EVENT_IO_HUP = 0x3, }; -typedef struct neu_event_io_ctx neu_event_io_ctx_t; +typedef struct neu_event_io neu_event_io_t; typedef int (*neu_event_io_callback)(enum neu_event_io_type type, int fd, void *usr_data); -typedef struct neu_event_io { +typedef struct neu_event_io_param { int fd; void * usr_data; neu_event_io_callback cb; -} neu_event_io_t; +} neu_event_io_param_t; -neu_event_io_ctx_t *neu_event_add_io(neu_event_ctx_t *events, - neu_event_io_t io); -int neu_event_del_io(neu_event_ctx_t *events, neu_event_io_ctx_t *io); +neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io); +int neu_event_del_io(neu_events_t *events, neu_event_io_t *io); #ifdef __cplusplus } diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index df5aa701..38b3f47b 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -977,7 +977,12 @@ static void adapter_loop(void *arg) req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_subscribe_node_t); req.buf = (void *) &sub_node_req; - intf_funs->request(adapter->plugin, &req); + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + pay_msg_type, &req); + } else { + intf_funs->request(adapter->plugin, &req); + } break; } @@ -999,7 +1004,12 @@ static void adapter_loop(void *arg) req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_unsubscribe_node_t); req.buf = (void *) &unsub_node_req; - intf_funs->request(adapter->plugin, &req); + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + pay_msg_type, &req); + } else { + intf_funs->request(adapter->plugin, &req); + } break; } @@ -1020,7 +1030,12 @@ static void adapter_loop(void *arg) req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_read_t); req.buf = (void *) &read_req; - intf_funs->request(adapter->plugin, &req); + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + pay_msg_type, &req); + } else { + intf_funs->request(adapter->plugin, &req); + } break; } @@ -1085,7 +1100,12 @@ static void adapter_loop(void *arg) req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_write_t); req.buf = (void *) &write_req; - intf_funs->request(adapter->plugin, &req); + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + pay_msg_type, &req); + } else { + intf_funs->request(adapter->plugin, &req); + } if (is_need_free) { neu_dvalue_free(write_req.data_val); } diff --git a/src/adapter/driver.c b/src/adapter/driver.c index aac2b7b7..c518b126 100644 --- a/src/adapter/driver.c +++ b/src/adapter/driver.c @@ -16,21 +16,28 @@ * along with this program; if not, write to the Free Software Foundation, * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ +#include #include #include "event/event.h" +#include "utils/utarray.h" #include "adapter.h" #include "adapter_internal.h" #include "cache.h" +#include "driver_internal.h" struct neu_adapter_driver { neu_adapter_t adapter; neu_driver_cache_t *cache; - neu_event_ctx_t *msg_events; - neu_event_ctx_t *driver_events; + neu_events_t *msg_events; + neu_events_t *driver_events; +}; +// read/write/subscribe/unsubscribe + +struct subscribe_group { }; neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter) @@ -78,7 +85,38 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver) return 0; } -void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver) +void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, + msg_type_e type, neu_request_t *req) { (void) driver; + switch (type) { + case MSG_CMD_READ_DATA: { + neu_reqresp_read_t *read_req = (neu_reqresp_read_t *) req->buf; + + (void) read_req->grp_config; + + break; + } + case MSG_CMD_WRITE_DATA: + break; + case MSG_CMD_SUBSCRIBE_NODE: { + neu_reqresp_subscribe_node_t *sub_req = + (neu_reqresp_subscribe_node_t *) req->buf; + + (void) sub_req->grp_config; + + break; + } + case MSG_CMD_UNSUBSCRIBE_NODE: { + neu_reqresp_unsubscribe_node_t *unsub_req = + (neu_reqresp_unsubscribe_node_t *) req->buf; + + (void) unsub_req->grp_config; + + break; + } + default: + assert(1 == 0); + break; + } } diff --git a/src/adapter/driver_internal.h b/src/adapter/driver_internal.h index 233b8ac2..206c3a16 100644 --- a/src/adapter/driver_internal.h +++ b/src/adapter/driver_internal.h @@ -21,6 +21,7 @@ #define _NEU_ADAPTER_DRIVER_INTERNAL_H_ #include "adapter.h" +#include "core/message.h" neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter); @@ -30,6 +31,7 @@ int neu_adapter_driver_stop(neu_adapter_driver_t *driver); int neu_adapter_driver_init(neu_adapter_driver_t *driver); int neu_adapter_driver_uninit(neu_adapter_driver_t *driver); -void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver); +void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, + msg_type_e type, neu_request_t *req); #endif diff --git a/src/event/event_linux.c b/src/event/event_linux.c index 3126d977..9c034be8 100644 --- a/src/event/event_linux.c +++ b/src/event/event_linux.c @@ -31,12 +31,12 @@ #include #include -struct neu_event_timer_ctx { +struct neu_event_timer { int fd; void *event_data; }; -struct neu_event_io_ctx { +struct neu_event_io { int fd; void *event_data; }; @@ -51,8 +51,8 @@ struct event_data { neu_event_timer_callback timer; } callback; union { - struct neu_event_io_ctx * io; - struct neu_event_timer_ctx *timer; + struct neu_event_io * io; + struct neu_event_timer *timer; } ctx; void *usr_data; @@ -61,7 +61,7 @@ struct event_data { TAILQ_ENTRY(event_data) node; }; -struct neu_event_ctx { +struct neu_events { int epoll_fd; bool running; pthread_t thread; @@ -72,9 +72,9 @@ struct neu_event_ctx { static void *event_loop(void *arg) { - neu_event_ctx_t *events = (neu_event_ctx_t *) arg; - bool running = events->running; - int epoll_fd = events->epoll_fd; + neu_events_t *events = (neu_events_t *) arg; + bool running = events->running; + int epoll_fd = events->epoll_fd; while (running) { struct epoll_event event = { 0 }; @@ -129,9 +129,9 @@ static void *event_loop(void *arg) return NULL; }; -neu_event_ctx_t *neu_event_new(void) +neu_events_t *neu_event_new(void) { - neu_event_ctx_t *events = calloc(1, sizeof(struct neu_event_ctx)); + neu_events_t *events = calloc(1, sizeof(struct neu_events)); events->epoll_fd = epoll_create(1); events->running = true; @@ -144,7 +144,7 @@ neu_event_ctx_t *neu_event_new(void) return events; }; -int neu_event_close(neu_event_ctx_t *events) +int neu_event_close(neu_events_t *events) { struct event_data *data = NULL; @@ -178,8 +178,8 @@ int neu_event_close(neu_event_ctx_t *events) return 0; } -neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, - neu_event_timer_t timer) +neu_event_timer_t *neu_event_add_timer(neu_events_t * events, + neu_event_timer_param_t timer) { int ret = 0; int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0); @@ -189,9 +189,9 @@ neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, .it_interval.tv_sec = timer.second, .it_interval.tv_nsec = timer.millisecond * 1000 * 1000, }; - struct event_data * data = calloc(1, sizeof(struct event_data)); - struct epoll_event event = { .events = EPOLLIN, .data.ptr = data }; - neu_event_timer_ctx_t *timer_ctx = calloc(1, sizeof(neu_event_timer_ctx_t)); + struct event_data *data = calloc(1, sizeof(struct event_data)); + struct epoll_event event = { .events = EPOLLIN, .data.ptr = data }; + neu_event_timer_t *timer_ctx = calloc(1, sizeof(neu_event_timer_t)); timerfd_settime(timer_fd, 0, &value, NULL); @@ -217,7 +217,7 @@ neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, return timer_ctx; } -int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer) +int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer) { struct event_data *data = NULL; log_info("del timer: %d from epoll: %d", timer->fd, events->epoll_fd); @@ -242,12 +242,12 @@ int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer) return 0; } -neu_event_io_ctx_t *neu_event_add_io(neu_event_ctx_t *events, neu_event_io_t io) +neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io) { - int ret = 0; - neu_event_io_ctx_t *io_ctx = calloc(1, sizeof(neu_event_io_ctx_t)); - struct event_data * data = calloc(1, sizeof(struct event_data)); - struct epoll_event event = { .events = + int ret = 0; + neu_event_io_t * io_ctx = calloc(1, sizeof(neu_event_io_t)); + struct event_data *data = calloc(1, sizeof(struct event_data)); + struct epoll_event event = { .events = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP, .data.ptr = data }; @@ -272,7 +272,7 @@ neu_event_io_ctx_t *neu_event_add_io(neu_event_ctx_t *events, neu_event_io_t io) return io_ctx; } -int neu_event_del_io(neu_event_ctx_t *events, neu_event_io_ctx_t *io) +int neu_event_del_io(neu_events_t *events, neu_event_io_t *io) { struct event_data *data = NULL; log_info("del io: %d from epoll: %d", io->fd, events->epoll_fd); diff --git a/src/event/event_unix.c b/src/event/event_unix.c index 5b5ce7aa..cc8b8048 100644 --- a/src/event/event_unix.c +++ b/src/event/event_unix.c @@ -31,16 +31,16 @@ #include -struct neu_event_timer_ctx { +struct neu_event_timer { int id; void * usr_data; neu_event_timer_callback timer; }; -struct neu_event_io_ctx { +struct neu_event_io { }; -struct neu_event_ctx { +struct neu_events { int kq; bool running; pthread_t thread; @@ -50,8 +50,8 @@ struct neu_event_ctx { static void *event_loop(void *arg) { - neu_event_ctx_t *events = (neu_event_ctx_t *) arg; - bool running = events->running; + neu_events_t *events = (neu_events_t *) arg; + bool running = events->running; while (running) { struct kevent event = { 0 }; @@ -69,7 +69,7 @@ static void *event_loop(void *arg) } if (event.filter == EVFILT_TIMER) { - neu_event_timer_ctx_t *ctx = (neu_event_timer_ctx_t *) event.udata; + neu_event_timer_t *ctx = (neu_event_timer_t *) event.udata; ret = ctx->timer(ctx->usr_data); log_debug("timer trigger: %d, ret: %d", ctx->id, ret); @@ -83,9 +83,9 @@ static void *event_loop(void *arg) return NULL; } -neu_event_ctx_t *neu_event_new(void) +neu_events_t *neu_event_new(void) { - neu_event_ctx_t *events = calloc(1, sizeof(neu_event_ctx_t)); + neu_events_t *events = calloc(1, sizeof(neu_events_t)); pthread_mutex_init(&events->mtx, NULL); @@ -98,7 +98,7 @@ neu_event_ctx_t *neu_event_new(void) return events; }; -int neu_event_close(neu_event_ctx_t *events) +int neu_event_close(neu_events_t *events) { pthread_mutex_lock(&events->mtx); events->running = false; @@ -111,12 +111,12 @@ int neu_event_close(neu_event_ctx_t *events) return 0; } -neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, - neu_event_timer_t timer) +neu_event_timer_t *neu_event_add_timer(neu_events_t * events, + neu_event_timer_param_t timer) { - struct kevent ke = { 0 }; - neu_event_timer_ctx_t *ctx = calloc(1, sizeof(neu_event_timer_ctx_t)); - int ret = 0; + struct kevent ke = { 0 }; + neu_event_timer_t *ctx = calloc(1, sizeof(neu_event_timer_t)); + int ret = 0; ctx->id = events->timer_id++; ctx->usr_data = timer.usr_data; @@ -133,7 +133,7 @@ neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, return ctx; } -int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer) +int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer) { struct kevent ke = { 0 }; @@ -145,14 +145,14 @@ int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer) return 0; } -neu_event_io_ctx_t *neu_event_add_io(neu_event_ctx_t *events, neu_event_io_t io) +neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io) { (void) events; (void) io; return NULL; } -int neu_event_del_io(neu_event_ctx_t *events, neu_event_io_ctx_t *io) +int neu_event_del_io(neu_events_t *events, neu_event_io_t *io) { (void) events; (void) io;