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 #include #include +#include -/* 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; + redisLibevEvents *e = watcher->data; + redisAsyncHandleRead(e->context); +} - 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); - } +void redisLibevWriteEvent(struct ev_loop *loop, ev_io *watcher, int revents) { + ((void)loop); ((void)revents); + redisLibevEvents *e = watcher->data; + redisAsyncHandleWrite(e->context); +} + +void redisLibevAddRead(void *privdata) { + redisLibevEvents *e = privdata; + if (!e->reading) { + e->reading = 1; + ev_io_start(e->loop,&e->rev); } } -void libevRedisWriteEvent(struct ev_loop *loop, ev_io *watcher, int revents) { - ((void)loop); ((void)revents); - libevRedisEvents *e = watcher->data; - int done = 0; - - 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 redisLibevDelRead(void *privdata) { + redisLibevEvents *e = privdata; + if (e->reading) { + e->reading = 0; + ev_io_stop(e->loop,&e->rev); } } -void libevRedisCommandCallback(redisContext *c, void *privdata) { - ((void)c); - libevRedisEvents *e = privdata; - ev_io_start(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 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 redisLibevDelWrite(void *privdata) { + redisLibevEvents *e = privdata; + if (e->writing) { + e->writing = 0; + ev_io_stop(e->loop,&e->wev); + } } -void libevRedisFreeCallback(redisContext *c, void *privdata) { - ((void)c); - libevRedisEvents *e = privdata; +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 #include #include +#include -/* 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; + redisLibeventEvents *e = arg; + redisAsyncHandleRead(e->context); +} - /* Always re-schedule read events */ +void redisLibeventWriteEvent(int fd, short event, void *arg) { + ((void)fd); ((void)event); + redisLibeventEvents *e = arg; + redisAsyncHandleWrite(e->context); +} + +void redisLibeventAddRead(void *privdata) { + redisLibeventEvents *e = privdata; 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); - } - } } -void libeventRedisWriteEvent(int fd, short event, void *arg) { - ((void)fd); ((void)event); - libeventRedisEvents *e = arg; - int done = 0; - - 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 redisLibeventDelRead(void *privdata) { + redisLibeventEvents *e = privdata; + event_del(&e->rev); } -/* 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 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; } diff --git a/libev-example.c b/libev-example.c index 49945bf..2d0bc8b 100644 --- a/libev-example.c +++ b/libev-example.c @@ -2,30 +2,37 @@ #include #include #include +#include #include -void getCallback(redisContext *c, redisReply *reply, void *privdata) { - if (reply == NULL) return; /* Error */ - printf("argv[%s]: %s\n", (char*)privdata, reply->reply); +void getCallback(redisAsyncContext *c, redisReply *reply, void *privdata) { + printf("argv[%s]: %s\n", (char*)privdata, reply->str); /* Disconnect after receiving the reply to GET */ - redisDisconnect(c); + redisAsyncDisconnect(c); } -void errorCallback(const redisContext *c) { - printf("Error: %s\n", c->error); +void disconnectCallback(const redisAsyncContext *c, int status) { + if (status != REDIS_OK) { + printf("Error: %s\n", c->error); + } } int main (int argc, char **argv) { signal(SIGPIPE, SIG_IGN); struct ev_loop *loop = ev_default_loop(0); - redisContext *c = libevRedisConnect(loop, errorCallback, "127.0.0.1", 6379); - if (c == NULL) return 1; + redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379); + if (c->error != NULL) { + /* Let *c leak for now... */ + printf("Error: %s\n", c->error); + return 1; + } - redisCommand(c, "SET key %b", argv[argc-1], strlen(argv[argc-1])); - redisCommandWithCallback(c, getCallback, (char*)"end-1", "GET key"); + redisLibevAttach(c,loop); + redisAsyncSetDisconnectCallback(c,disconnectCallback); + redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1])); + redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key"); ev_loop(loop, 0); - redisFree(c); return 0; } diff --git a/libevent-example.c b/libevent-example.c index 9c82850..a5a24b5 100644 --- a/libevent-example.c +++ b/libevent-example.c @@ -2,30 +2,37 @@ #include #include #include +#include #include -void getCallback(redisContext *c, redisReply *reply, void *privdata) { - if (reply == NULL) return; /* Error */ - printf("argv[%s]: %s\n", (const char*)privdata, reply->reply); +void getCallback(redisAsyncContext *c, redisReply *reply, void *privdata) { + printf("argv[%s]: %s\n", (char*)privdata, reply->str); /* Disconnect after receiving the reply to GET */ - redisDisconnect(c); + redisAsyncDisconnect(c); } -void errorCallback(const redisContext *c) { - printf("Error: %s\n", c->error); +void disconnectCallback(const redisAsyncContext *c, int status) { + if (status != REDIS_OK) { + printf("Error: %s\n", c->error); + } } int main (int argc, char **argv) { signal(SIGPIPE, SIG_IGN); struct event_base *base = event_base_new(); - redisContext *c = libeventRedisConnect(base, errorCallback, "127.0.0.1", 6379); - if (c == NULL) return 1; + redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379); + if (c->error != NULL) { + /* Let *c leak for now... */ + printf("Error: %s\n", c->error); + return 1; + } - redisCommand(c, "SET key %b", argv[argc-1], strlen(argv[argc-1])); - redisCommandWithCallback(c, getCallback, (char*)"end-1", "GET key"); + redisLibeventAttach(c,base); + redisAsyncSetDisconnectCallback(c,disconnectCallback); + redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1])); + redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key"); event_base_dispatch(base); - redisFree(c); return 0; }