Avoid incorrect call to the previous reply's callback (#1040)

* No reuse of the previous reply callback

When multiple replies are parsed from a socket in one read
a previously found callback might get reused when the current
reply has no known callback.

This can be triggered by the added testcase which unsubscribe to
subscribed (A,B) and a non-subscribed channel (X).
Without this correction a callback for wrong channel is called.
-  In 'unsubscribe B X A', B's callback is called when handling X.
-  Now this is not done, i.e. there is no callback called for X.

* Re-push monitor callback for each reply

MONITORING used the same callback for all replies while parsing
multiple responses. This handling was changed to avoid calling
the wrong callback in some scenarios.
Now also change monitorings repush to work with this change.

Includes an added async monitoring testcase.

1 files changed, 9 insertions, 10 deletions

diff --git a/async.c b/async.c
index 8614b20..6555114 100644
--- a/async.c
+++ b/async.c
@@ -508,7 +508,6 @@ static int redisIsSubscribeReply(redisReply *reply) {
 
 void redisProcessCallbacks(redisAsyncContext *ac) {
     redisContext *c = &(ac->c);
-    redisCallback cb = {NULL, NULL, 0, NULL};
     void *reply = NULL;
     int status;
 
@@ -521,12 +520,6 @@ void redisProcessCallbacks(redisAsyncContext *ac) {
                 __redisAsyncDisconnect(ac);
                 return;
             }
-
-            /* If monitor mode, repush callback */
-            if(c->flags & REDIS_MONITORING) {
-                __redisPushCallback(&ac->replies,&cb);
-            }
-
             /* When the connection is not being disconnected, simply stop
              * trying to get replies and wait for the next loop tick. */
             break;
@@ -547,6 +540,7 @@ void redisProcessCallbacks(redisAsyncContext *ac) {
 
         /* Even if the context is subscribed, pending regular
          * callbacks will get a reply before pub/sub messages arrive. */
+        redisCallback cb = {NULL, NULL, 0, NULL};
         if (__redisShiftCallback(&ac->replies,&cb) != REDIS_OK) {
             /*
              * A spontaneous reply in a not-subscribed context can be the error
@@ -570,9 +564,9 @@ void redisProcessCallbacks(redisAsyncContext *ac) {
                 __redisAsyncDisconnect(ac);
                 return;
             }
-            /* No more regular callbacks and no errors, the context *must* be subscribed or monitoring. */
-            assert((c->flags & REDIS_SUBSCRIBED || c->flags & REDIS_MONITORING));
-            if(c->flags & REDIS_SUBSCRIBED)
+            /* No more regular callbacks and no errors, the context *must* be subscribed. */
+            assert(c->flags & REDIS_SUBSCRIBED);
+            if (c->flags & REDIS_SUBSCRIBED)
                 __redisGetSubscribeCallback(ac,reply,&cb);
         }
 
@@ -594,6 +588,11 @@ void redisProcessCallbacks(redisAsyncContext *ac) {
              * doesn't know what the server will spit out over the wire. */
             c->reader->fn->freeObject(reply);
         }
+
+        /* If in monitor mode, repush the callback */
+        if (c->flags & REDIS_MONITORING) {
+            __redisPushCallback(&ac->replies,&cb);
+        }
     }
 
     /* Disconnect when there was an error reading the reply */