refactor: move command queue from MuxExecutor to RedisConnection (#383)

Author: uki00a

connection.ts

@@ -5,6 +5,10 @@ import { exponentialBackoff } from "./backoff.ts";
 import { ErrorReplyError, isRetriableError } from "./errors.ts";
 import { BufReader } from "./vendor/https/deno.land/std/io/buf_reader.ts";
 import { BufWriter } from "./vendor/https/deno.land/std/io/buf_writer.ts";
+import {
+  Deferred,
+  deferred,
+} from "./vendor/https/deno.land/std/async/deferred.ts";
 import { delay } from "./vendor/https/deno.land/std/async/delay.ts";
 type Closer = Deno.Closer;
 
@@ -50,6 +54,12 @@ export class RedisConnection implements Connection {
   private _isConnected = false;
   private backoff: Backoff;
 
+  private commandQueue: {
+    name: string;
+    args: RedisValue[];
+    promise: Deferred<RedisReply>;
+  }[] = [];
+
   get isClosed(): boolean {
     return this._isClosed;
   }
@@ -104,48 +114,20 @@ export class RedisConnection implements Connection {
     await this.sendCommand("SELECT", [db]);
   }
 
-  async sendCommand(
+  sendCommand(
     command: string,
     args?: Array<RedisValue>,
   ): Promise<RedisReply> {
-    try {
-      const reply = await sendCommand(
-        this.writer,
-        this.reader,
-        command,
-        args ?? kEmptyRedisArgs,
-      );
-      return reply;
-    } catch (error) {
-      if (
-        !isRetriableError(error) ||
-        this.isManuallyClosedByUser()
-      ) {
-        throw error;
-      }
-
-      for (let i = 0; i < this.maxRetryCount; i++) {
-        // Try to reconnect to the server and retry the command
-        this.close();
-        try {
-          await this.connect();
-
-          const reply = await sendCommand(
-            this.writer,
-            this.reader,
-            command,
-            args ?? kEmptyRedisArgs,
-          );
-
-          return reply;
-        } catch { // TODO: use `AggregateError`?
-          const backoff = this.backoff(i);
-          await delay(backoff);
-        }
-      }
-
-      throw error;
+    const promise = deferred<RedisReply>();
+    this.commandQueue.push({
+      name: command,
+      args: args ?? kEmptyRedisArgs,
+      promise,
+    });
+    if (this.commandQueue.length === 1) {
+      this.processCommandQueue();
     }
+    return promise;
   }
 
   /**
@@ -221,6 +203,53 @@ export class RedisConnection implements Connection {
     }
   }
 
+  private async processCommandQueue() {
+    const [command] = this.commandQueue;
+    if (!command) return;
+
+    try {
+      const reply = await sendCommand(
+        this.writer,
+        this.reader,
+        command.name,
+        command.args,
+      );
+      command.promise.resolve(reply);
+    } catch (error) {
+      if (
+        !isRetriableError(error) ||
+        this.isManuallyClosedByUser()
+      ) {
+        return command.promise.reject(error);
+      }
+
+      for (let i = 0; i < this.maxRetryCount; i++) {
+        // Try to reconnect to the server and retry the command
+        this.close();
+        try {
+          await this.connect();
+
+          const reply = await sendCommand(
+            this.writer,
+            this.reader,
+            command.name,
+            command.args,
+          );
+
+          return command.promise.resolve(reply);
+        } catch { // TODO: use `AggregateError`?
+          const backoff = this.backoff(i);
+          await delay(backoff);
+        }
+      }
+
+      command.promise.reject(error);
+    } finally {
+      this.commandQueue.shift();
+      this.processCommandQueue();
+    }
+  }
+
   private isManuallyClosedByUser(): boolean {
     return this._isClosed && !this._isConnected;
   }

executor.ts

@@ -1,8 +1,4 @@
 import type { Connection } from "./connection.ts";
-import {
-  Deferred,
-  deferred,
-} from "./vendor/https/deno.land/std/async/deferred.ts";
 import type { RedisReply, RedisValue } from "./protocol/mod.ts";
 
 export interface CommandExecutor {
@@ -18,40 +14,17 @@ export interface CommandExecutor {
   close(): void;
 }
 
-export class MuxExecutor implements CommandExecutor {
+export class DefaultExecutor implements CommandExecutor {
   constructor(readonly connection: Connection) {}
 
-  private queue: {
-    command: string;
-    args: RedisValue[];
-    d: Deferred<RedisReply>;
-  }[] = [];
-
   exec(
     command: string,
     ...args: RedisValue[]
   ): Promise<RedisReply> {
-    const d = deferred<RedisReply>();
-    this.queue.push({ command, args, d });
-    if (this.queue.length === 1) {
-      this.dequeue();
-    }
-    return d;
+    return this.connection.sendCommand(command, args);
   }
 
   close(): void {
     this.connection.close();
   }
-
-  private dequeue(): void {
-    const [e] = this.queue;
-    if (!e) return;
-    this.connection.sendCommand(e.command, e.args)
-      .then(e.d.resolve)
-      .catch(e.d.reject)
-      .finally(() => {
-        this.queue.shift();
-        this.dequeue();
-      });
-  }
 }

redis.ts

@@ -44,7 +44,7 @@ import type {
 import { RedisConnection } from "./connection.ts";
 import type { Connection } from "./connection.ts";
 import type { RedisConnectionOptions } from "./connection.ts";
-import { CommandExecutor, MuxExecutor } from "./executor.ts";
+import { CommandExecutor, DefaultExecutor } from "./executor.ts";
 import type {
   Binary,
   Bulk,
@@ -2361,7 +2361,7 @@ export interface RedisConnectOptions extends RedisConnectionOptions {
 export async function connect(options: RedisConnectOptions): Promise<Redis> {
   const connection = createRedisConnection(options);
   await connection.connect();
-  const executor = new MuxExecutor(connection);
+  const executor = new DefaultExecutor(connection);
   return create(executor);
 }
 
@@ -2439,7 +2439,7 @@ function createLazyExecutor(connection: Connection): CommandExecutor {
     },
     async exec(command, ...args) {
       if (!executor) {
-        executor = new MuxExecutor(connection);
+        executor = new DefaultExecutor(connection);
         if (!connection.isConnected) {
           await connection.connect();
         }