Run pending callbacks with NULL reply on error

1 files changed, 39 insertions, 21 deletions

diff --git a/async.c b/async.c
index bc6f330..b80277d 100644
--- a/async.c
+++ b/async.c
@@ -68,6 +68,25 @@ int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallba
     return REDIS_ERR;
 }
 
+/* Helper functions to push/shift callbacks */
+static void __redisPushCallback(redisCallbackList *list, redisCallback *cb) {
+    if (list->head == NULL)
+        list->head = cb;
+    if (list->tail != NULL)
+        list->tail->next = cb;
+    list->tail = cb;
+}
+
+static redisCallback *__redisShiftCallback(redisCallbackList *list) {
+    redisCallback *cb = list->head;
+    if (cb != NULL) {
+        list->head = cb->next;
+        if (cb == list->tail)
+            list->tail = NULL;
+    }
+    return cb;
+}
+
 /* Tries to do a clean disconnect from Redis, meaning it stops new commands
  * from being issued, but tries to flush the output buffer and execute
  * callbacks for all remaining replies.
@@ -83,39 +102,38 @@ void redisAsyncDisconnect(redisAsyncContext *ac) {
 /* Helper function to make the disconnect happen and clean up. */
 static void __redisAsyncDisconnect(redisAsyncContext *ac) {
     redisContext *c = &(ac->c);
+    redisCallback *cb;
     int status;
 
+    /* Make sure error is accessible if there is any */
+    __redisAsyncCopyError(ac);
+    status = (ac->error == NULL) ? REDIS_OK : REDIS_ERR;
+
+    if (status == REDIS_OK) {
+        /* When the connection is cleanly disconnected, there should not
+         * be pending callbacks. */
+        assert((cb = __redisShiftCallback(&ac->replies)) == NULL);
+    } else {
+        /* Callbacks should not be able to issue new commands. */
+        c->flags |= REDIS_DISCONNECTING;
+
+        /* Execute pending callbacks with NULL reply. */
+        while ((cb = __redisShiftCallback(&ac->replies)) != NULL) {
+            if (cb->fn != NULL)
+                cb->fn(ac,NULL,cb->privdata);
+        }
+    }
+
     /* Signal event lib to clean up */
     if (ac->evCleanup) ac->evCleanup(ac->data);
 
     /* Execute callback with proper status */
-    __redisAsyncCopyError(ac);
-    status = (ac->error == NULL) ? REDIS_OK : REDIS_ERR;
     if (ac->onDisconnect) ac->onDisconnect(ac,status);
 
     /* Cleanup self */
     redisFree(c);
 }
 
-/* Helper functions to push/shift callbacks */
-static void __redisPushCallback(redisCallbackList *list, redisCallback *cb) {
-    if (list->head == NULL)
-        list->head = cb;
-    if (list->tail != NULL)
-        list->tail->next = cb;
-    list->tail = cb;
-}
-
-static redisCallback *__redisShiftCallback(redisCallbackList *list) {
-    redisCallback *cb = list->head;
-    if (cb != NULL) {
-        list->head = cb->next;
-        if (cb == list->tail)
-            list->tail = NULL;
-    }
-    return cb;
-}
-
 void redisProcessCallbacks(redisAsyncContext *ac) {
     redisContext *c = &(ac->c);
     redisCallback *cb;