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

rename event

Browse Source 
neu_event_ctx_t -> neu_events_t
neu_event_timer_ctx_t -> neu_event_timer_t
neu_event_io_ctx_t -> neu_event_io_t
This commit is contained in:
fengzero
2022-04-08 05:15:51 +00:00
parent 78b1e9532b
commit 34456d776e
6 changed files with 122 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/event/event.h
+14 -15
View File
@@ -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
}
src/adapter/adapter.c
+24 -4
View File
@@ -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);
}
src/adapter/driver.c
+41 -3
View File
@@ -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 <assert.h>
#include <stdlib.h>
#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;
}
}
src/adapter/driver_internal.h
+3 -1
View File
@@ -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
src/event/event_linux.c
+23 -23
View File
@@ -31,12 +31,12 @@
#include <sys/epoll.h>
#include <sys/timerfd.h>
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);
src/event/event_unix.c
+17 -17
View File
@@ -31,16 +31,16 @@
#include <sys/event.h>
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;