#include #include #include /* Prototype for the error callback. */ typedef void (redisErrorCallback)(const redisContext*); typedef struct libeventRedisEvents { redisContext *context; redisErrorCallback *err; struct event rev, wev; } libeventRedisEvents; void libeventRedisReadEvent(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); } } } 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); } } } /* 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; 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); 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; 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; } /* Create container for context and r/w events */ e = malloc(sizeof(*e)); e->context = c; e->err = err; /* Register callbacks */ redisSetDisconnectCallback(c,libeventRedisDisconnectCallback,e); redisSetCommandCallback(c,libeventRedisCommandCallback,e); redisSetFreeCallback(c,libeventRedisFreeCallback,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_base_set(base,&e->rev); event_base_set(base,&e->wev); return c; }