Transactions: move the QUEUED logic inside call(). (Work in progress)

As a side effect this makes things like #5201 fixed automatically.
The problem with handling it in ProcessQueryBuffer() or similar is that
a client that is handled just via the call() interface will misbehave,
and call()-ing MULTI and later some command will see just the other
command executed like if the client was not in a transaction.

For now this has worked well enough but probably it makes more sense if
we can handle that case on call().

WARNING: this comment is not enough and will introduce bugs. It's a WIP.

1 files changed, 16 insertions, 13 deletions

diff --git a/src/server.c b/src/server.c
index 78aee5dbb..fd65bb182 100644
--- a/src/server.c
+++ b/src/server.c
@@ -2410,7 +2410,7 @@ void call(client *c, int flags) {
     int client_old_flags = c->flags;
     struct redisCommand *real_cmd = c->cmd;
 
-    /* Sent the command to clients in MONITOR mode, only if the commands are
+    /* Send the command to clients in MONITOR mode, only if the commands are
      * not generated from reading an AOF. */
     if (listLength(server.monitors) &&
         !server.loading &&
@@ -2419,6 +2419,18 @@ void call(client *c, int flags) {
         replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
     }
 
+    /* If the client is in the context of a transaction, reply with
+     * +QUEUED and just accumulate the command in the client transaction
+     * commands vector. */
+    if (c->flags & CLIENT_MULTI &&
+        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
+        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
+    {
+        queueMultiCommand(c);
+        addReply(c,shared.queued);
+        return;
+    }
+
     /* Initialization: clear the flags that must be set by the command on
      * demand, and initialize the array for additional commands propagation. */
     c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
@@ -2714,18 +2726,9 @@ int processCommand(client *c) {
     }
 
     /* Exec the command */
-    if (c->flags & CLIENT_MULTI &&
-        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
-        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
-    {
-        queueMultiCommand(c);
-        addReply(c,shared.queued);
-    } else {
-        call(c,CMD_CALL_FULL);
-        c->woff = server.master_repl_offset;
-        if (listLength(server.ready_keys))
-            handleClientsBlockedOnKeys();
-    }
+    call(c,CMD_CALL_FULL);
+    c->woff = server.master_repl_offset;
+    if (listLength(server.ready_keys)) handleClientsBlockedOnKeys();
     return C_OK;
 }