2 files changed, 91 insertions, 30 deletions

diff --git a/adapters/libevent.h b/adapters/libevent.h
index 7d2bef1..a495277 100644
--- a/adapters/libevent.h
+++ b/adapters/libevent.h
@@ -34,48 +34,113 @@
 #include "../hiredis.h"
 #include "../async.h"
 
+#define REDIS_LIBEVENT_DELETED 0x01
+#define REDIS_LIBEVENT_ENTERED 0x02
+
 typedef struct redisLibeventEvents {
     redisAsyncContext *context;
-    struct event *rev, *wev;
+    struct event *ev;
+    struct event_base *base;
+    struct timeval tv;
+    short flags;
+    short state;
 } redisLibeventEvents;
 
-static void redisLibeventReadEvent(int fd, short event, void *arg) {
-    ((void)fd); ((void)event);
-    redisLibeventEvents *e = (redisLibeventEvents*)arg;
-    redisAsyncHandleRead(e->context);
+static void redisLibeventDestroy(redisLibeventEvents *e) {
+    free(e);
 }
 
-static void redisLibeventWriteEvent(int fd, short event, void *arg) {
-    ((void)fd); ((void)event);
+static void redisLibeventHandler(int fd, short event, void *arg) {
+    ((void)fd);
     redisLibeventEvents *e = (redisLibeventEvents*)arg;
-    redisAsyncHandleWrite(e->context);
+    e->state |= REDIS_LIBEVENT_ENTERED;
+
+    #define CHECK_DELETED() if (e->state & REDIS_LIBEVENT_DELETED) {\
+        redisLibeventDestroy(e);\
+        return; \
+    }
+
+    if ((event & EV_TIMEOUT) && (e->state & REDIS_LIBEVENT_DELETED) == 0) {
+        redisAsyncHandleTimeout(e->context);
+        CHECK_DELETED();
+    }
+
+    if ((event & EV_READ) && e->context && (e->state & REDIS_LIBEVENT_DELETED) == 0) {
+        redisAsyncHandleRead(e->context);
+        CHECK_DELETED();
+    }
+
+    if ((event & EV_WRITE) && e->context && (e->state & REDIS_LIBEVENT_DELETED) == 0) {
+        redisAsyncHandleWrite(e->context);
+        CHECK_DELETED();
+    }
+
+    e->state &= ~REDIS_LIBEVENT_ENTERED;
+    #undef CHECK_DELETED
+}
+
+static void redisLibeventUpdate(void *privdata, short flag, int isRemove) {
+    redisLibeventEvents *e = (redisLibeventEvents *)privdata;
+    const struct timeval *tv = e->tv.tv_sec || e->tv.tv_usec ? &e->tv : NULL;
+
+    if (isRemove) {
+        if ((e->flags & flag) == 0) {
+            return;
+        } else {
+            e->flags &= ~flag;
+        }
+    } else {
+        if (e->flags & flag) {
+            return;
+        } else {
+            e->flags |= flag;
+        }
+    }
+
+    event_del(e->ev);
+    event_assign(e->ev, e->base, e->context->c.fd, e->flags | EV_PERSIST,
+                 redisLibeventHandler, privdata);
+    event_add(e->ev, tv);
 }
 
 static void redisLibeventAddRead(void *privdata) {
-    redisLibeventEvents *e = (redisLibeventEvents*)privdata;
-    event_add(e->rev,NULL);
+    redisLibeventUpdate(privdata, EV_READ, 0);
 }
 
 static void redisLibeventDelRead(void *privdata) {
-    redisLibeventEvents *e = (redisLibeventEvents*)privdata;
-    event_del(e->rev);
+    redisLibeventUpdate(privdata, EV_READ, 1);
 }
 
 static void redisLibeventAddWrite(void *privdata) {
-    redisLibeventEvents *e = (redisLibeventEvents*)privdata;
-    event_add(e->wev,NULL);
+    redisLibeventUpdate(privdata, EV_WRITE, 0);
 }
 
 static void redisLibeventDelWrite(void *privdata) {
-    redisLibeventEvents *e = (redisLibeventEvents*)privdata;
-    event_del(e->wev);
+    redisLibeventUpdate(privdata, EV_WRITE, 1);
 }
 
 static void redisLibeventCleanup(void *privdata) {
     redisLibeventEvents *e = (redisLibeventEvents*)privdata;
-    event_free(e->rev);
-    event_free(e->wev);
-    free(e);
+    if (!e) {
+        return;
+    }
+    event_del(e->ev);
+    event_free(e->ev);
+    e->ev = NULL;
+
+    if (e->state & REDIS_LIBEVENT_ENTERED) {
+        e->state |= REDIS_LIBEVENT_DELETED;
+    } else {
+        redisLibeventDestroy(e);
+    }
+}
+
+static void redisLibeventSetTimeout(void *privdata, struct timeval tv) {
+    redisLibeventEvents *e = (redisLibeventEvents *)privdata;
+    short flags = e->flags;
+    e->flags = 0;
+    e->tv = tv;
+    redisLibeventUpdate(e, flags, 0);
 }
 
 static int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) {
@@ -87,7 +152,7 @@ static int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) {
         return REDIS_ERR;
 
     /* Create container for context and r/w events */
-    e = (redisLibeventEvents*)malloc(sizeof(*e));
+    e = (redisLibeventEvents*)calloc(1, sizeof(*e));
     e->context = ac;
 
     /* Register functions to start/stop listening for events */
@@ -96,13 +161,12 @@ static int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) {
     ac->ev.addWrite = redisLibeventAddWrite;
     ac->ev.delWrite = redisLibeventDelWrite;
     ac->ev.cleanup = redisLibeventCleanup;
+    ac->ev.scheduleTimer = redisLibeventSetTimeout;
     ac->ev.data = e;
 
     /* Initialize and install read/write events */
-    e->rev = event_new(base, c->fd, EV_READ, redisLibeventReadEvent, e);
-    e->wev = event_new(base, c->fd, EV_WRITE, redisLibeventWriteEvent, e);
-    event_add(e->rev, NULL);
-    event_add(e->wev, NULL);
+    e->ev = event_new(base, c->fd, EV_READ | EV_WRITE, redisLibeventHandler, e);
+    e->base = base;
     return REDIS_OK;
 }
 #endif
diff --git a/adapters/libuv.h b/adapters/libuv.h
index ff08c25..39ef7cf 100644
--- a/adapters/libuv.h
+++ b/adapters/libuv.h
@@ -15,15 +15,12 @@ typedef struct redisLibuvEvents {
 
 static void redisLibuvPoll(uv_poll_t* handle, int status, int events) {
   redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
+  int ev = (status ? p->events : events);
 
-  if (status != 0) {
-    return;
-  }
-
-  if (p->context != NULL && (events & UV_READABLE)) {
+  if (p->context != NULL && (ev & UV_READABLE)) {
     redisAsyncHandleRead(p->context);
   }
-  if (p->context != NULL && (events & UV_WRITABLE)) {
+  if (p->context != NULL && (ev & UV_WRITABLE)) {
     redisAsyncHandleWrite(p->context);
   }
 }