diff options
Diffstat (limited to 'extra')
-rw-r--r-- | extra/hiredis/libev.h | 120 | ||||
-rw-r--r-- | extra/hiredis/libevent.h | 112 |
2 files changed, 105 insertions, 127 deletions
diff --git a/extra/hiredis/libev.h b/extra/hiredis/libev.h index 3246820..fcb7c24 100644 --- a/extra/hiredis/libev.h +++ b/extra/hiredis/libev.h @@ -1,92 +1,92 @@ #include <sys/types.h> #include <ev.h> #include <hiredis.h> +#include <async.h> -/* Prototype for the error callback. */ -typedef void (redisErrorCallback)(const redisContext*); - -typedef struct libevRedisEvents { - redisContext *context; - redisErrorCallback *err; +typedef struct redisLibevEvents { + redisAsyncContext *context; struct ev_loop *loop; + int reading, writing; ev_io rev, wev; -} libevRedisEvents; +} redisLibevEvents; -void libevRedisReadEvent(struct ev_loop *loop, ev_io *watcher, int revents) { +void redisLibevReadEvent(struct ev_loop *loop, ev_io *watcher, int revents) { ((void)loop); ((void)revents); - libevRedisEvents *e = watcher->data; - - if (redisBufferRead(e->context) == REDIS_ERR) { - redisDisconnect(e->context); - e->err(e->context); - } else { - if (redisProcessCallbacks(e->context) == REDIS_ERR) { - redisDisconnect(e->context); - e->err(e->context); - } - } + redisLibevEvents *e = watcher->data; + redisAsyncHandleRead(e->context); } -void libevRedisWriteEvent(struct ev_loop *loop, ev_io *watcher, int revents) { +void redisLibevWriteEvent(struct ev_loop *loop, ev_io *watcher, int revents) { ((void)loop); ((void)revents); - libevRedisEvents *e = watcher->data; - int done = 0; + redisLibevEvents *e = watcher->data; + redisAsyncHandleWrite(e->context); +} - if (redisBufferWrite(e->context, &done) == REDIS_ERR) { - redisDisconnect(e->context); - e->err(e->context); - } else { - /* Stop firing the write event when done */ - if (done) { - ev_io_stop(e->loop,&e->wev); - ev_io_start(e->loop,&e->rev); - } +void redisLibevAddRead(void *privdata) { + redisLibevEvents *e = privdata; + if (!e->reading) { + e->reading = 1; + ev_io_start(e->loop,&e->rev); } } -void libevRedisCommandCallback(redisContext *c, void *privdata) { - ((void)c); - libevRedisEvents *e = privdata; - ev_io_start(e->loop,&e->wev); +void redisLibevDelRead(void *privdata) { + redisLibevEvents *e = privdata; + if (e->reading) { + e->reading = 0; + ev_io_stop(e->loop,&e->rev); + } } -void libevRedisDisconnectCallback(redisContext *c, void *privdata) { - ((void)c); - libevRedisEvents *e = privdata; - ev_io_stop(e->loop,&e->rev); - ev_io_stop(e->loop,&e->wev); +void redisLibevAddWrite(void *privdata) { + redisLibevEvents *e = privdata; + if (!e->writing) { + e->writing = 1; + ev_io_start(e->loop,&e->wev); + } } -void libevRedisFreeCallback(redisContext *c, void *privdata) { - ((void)c); - libevRedisEvents *e = privdata; +void redisLibevDelWrite(void *privdata) { + redisLibevEvents *e = privdata; + if (e->writing) { + e->writing = 0; + ev_io_stop(e->loop,&e->wev); + } +} + +void redisLibevCleanup(void *privdata) { + redisLibevEvents *e = privdata; + redisLibevDelRead(privdata); + redisLibevDelWrite(privdata); free(e); } -redisContext *libevRedisConnect(struct ev_loop *loop, redisErrorCallback *err, const char *ip, int port) { - libevRedisEvents *e; - redisContext *c = redisConnectNonBlock(ip, port, NULL); - if (c->error != NULL) { - err(c); - redisFree(c); - return NULL; - } +int redisLibevAttach(redisAsyncContext *ac, struct ev_loop *loop) { + redisContext *c = &(ac->c); + redisLibevEvents *e; + + /* Nothing should be attached when something is already attached */ + if (ac->data != NULL) + return REDIS_ERR; /* Create container for context and r/w events */ e = malloc(sizeof(*e)); - e->context = c; - e->err = err; + e->context = ac; e->loop = loop; + e->reading = e->writing = 0; e->rev.data = e; e->wev.data = e; - /* Register callbacks */ - redisSetDisconnectCallback(c,libevRedisDisconnectCallback,e); - redisSetCommandCallback(c,libevRedisCommandCallback,e); - redisSetFreeCallback(c,libevRedisFreeCallback,e); + /* Register functions to start/stop listening for events */ + ac->evAddRead = redisLibevAddRead; + ac->evDelRead = redisLibevDelRead; + ac->evAddWrite = redisLibevAddWrite; + ac->evDelWrite = redisLibevDelWrite; + ac->evCleanup = redisLibevCleanup; + ac->data = e; /* Initialize read/write events */ - ev_io_init(&e->rev,libevRedisReadEvent,c->fd,EV_READ); - ev_io_init(&e->wev,libevRedisWriteEvent,c->fd,EV_WRITE); - return c; + ev_io_init(&e->rev,redisLibevReadEvent,c->fd,EV_READ); + ev_io_init(&e->wev,redisLibevWriteEvent,c->fd,EV_WRITE); + return REDIS_OK; } diff --git a/extra/hiredis/libevent.h b/extra/hiredis/libevent.h index be70b25..f2851f9 100644 --- a/extra/hiredis/libevent.h +++ b/extra/hiredis/libevent.h @@ -1,98 +1,76 @@ #include <sys/types.h> #include <event.h> #include <hiredis.h> +#include <async.h> -/* Prototype for the error callback. */ -typedef void (redisErrorCallback)(const redisContext*); - -typedef struct libeventRedisEvents { - redisContext *context; - redisErrorCallback *err; +typedef struct redisLibeventEvents { + redisAsyncContext *context; struct event rev, wev; -} libeventRedisEvents; +} redisLibeventEvents; -void libeventRedisReadEvent(int fd, short event, void *arg) { +void redisLibeventReadEvent(int fd, short event, void *arg) { ((void)fd); ((void)event); - libeventRedisEvents *e = arg; - - /* Always re-schedule read events */ - event_add(&e->rev,NULL); - - if (redisBufferRead(e->context) == REDIS_ERR) { - redisDisconnect(e->context); - e->err(e->context); - } else { - if (redisProcessCallbacks(e->context) == REDIS_ERR) { - redisDisconnect(e->context); - e->err(e->context); - } - } + redisLibeventEvents *e = arg; + redisAsyncHandleRead(e->context); } -void libeventRedisWriteEvent(int fd, short event, void *arg) { +void redisLibeventWriteEvent(int fd, short event, void *arg) { ((void)fd); ((void)event); - libeventRedisEvents *e = arg; - int done = 0; + redisLibeventEvents *e = arg; + redisAsyncHandleWrite(e->context); +} - if (redisBufferWrite(e->context,&done) == REDIS_ERR) { - redisDisconnect(e->context); - e->err(e->context); - } else { - /* Schedule write event again when writing is not done. */ - if (!done) { - event_add(&e->wev,NULL); - } else { - event_add(&e->rev,NULL); - } - } +void redisLibeventAddRead(void *privdata) { + redisLibeventEvents *e = privdata; + event_add(&e->rev,NULL); } -/* Schedule to be notified on a write event, so the outgoing buffer - * can be flushed to the socket. */ -void libeventRedisCommandCallback(redisContext *c, void *privdata) { - ((void)c); - libeventRedisEvents *e = privdata; +void redisLibeventDelRead(void *privdata) { + redisLibeventEvents *e = privdata; + event_del(&e->rev); +} + +void redisLibeventAddWrite(void *privdata) { + redisLibeventEvents *e = privdata; event_add(&e->wev,NULL); } -/* Remove event handlers when the context gets disconnected. */ -void libeventRedisDisconnectCallback(redisContext *c, void *privdata) { - ((void)c); - libeventRedisEvents *e = privdata; - event_del(&e->rev); +void redisLibeventDelWrite(void *privdata) { + redisLibeventEvents *e = privdata; event_del(&e->wev); } -/* Free the libeventRedisEvents struct when the context is free'd. */ -void libeventRedisFreeCallback(redisContext *c, void *privdata) { - ((void)c); - libeventRedisEvents *e = privdata; +void redisLibeventCleanup(void *privdata) { + redisLibeventEvents *e = privdata; + event_del(&e->rev); + event_del(&e->wev); free(e); } -redisContext *libeventRedisConnect(struct event_base *base, redisErrorCallback *err, const char *ip, int port) { - libeventRedisEvents *e; - redisContext *c = redisConnectNonBlock(ip,port,NULL); - if (c->error != NULL) { - err(c); - redisFree(c); - return NULL; - } +int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) { + redisContext *c = &(ac->c); + redisLibeventEvents *e; + + /* Nothing should be attached when something is already attached */ + if (ac->data != NULL) + return REDIS_ERR; /* Create container for context and r/w events */ e = malloc(sizeof(*e)); - e->context = c; - e->err = err; + e->context = ac; - /* Register callbacks */ - redisSetDisconnectCallback(c,libeventRedisDisconnectCallback,e); - redisSetCommandCallback(c,libeventRedisCommandCallback,e); - redisSetFreeCallback(c,libeventRedisFreeCallback,e); + /* Register functions to start/stop listening for events */ + ac->evAddRead = redisLibeventAddRead; + ac->evDelRead = redisLibeventDelRead; + ac->evAddWrite = redisLibeventAddWrite; + ac->evDelWrite = redisLibeventDelWrite; + ac->evCleanup = redisLibeventCleanup; + ac->data = e; /* Initialize and install read/write events */ - event_set(&e->rev,c->fd,EV_READ,libeventRedisReadEvent,e); - event_set(&e->wev,c->fd,EV_WRITE,libeventRedisWriteEvent,e); + event_set(&e->rev,c->fd,EV_READ,redisLibeventReadEvent,e); + event_set(&e->wev,c->fd,EV_WRITE,redisLibeventWriteEvent,e); event_base_set(base,&e->rev); event_base_set(base,&e->wev); - return c; + return REDIS_OK; } |