fix: auto reconnection hangs (#431)

Closes #431

---------

Co-authored-by: yahiro <yahiro07@users.noreply.github.com>

Author: yahiro07

connection.ts

@@ -19,6 +19,13 @@ export interface SendCommandOptions {
    * @default false
    */
   returnUint8Arrays?: boolean;
+
+  /**
+   * When this option is set, the command is executed directly without queueing.
+   *
+   * @default false
+   */
+  inline?: boolean;
 }
 
 export interface Connection {
@@ -126,8 +133,8 @@ export class RedisConnection implements Connection {
   ): Promise<void> {
     try {
       password && username
-        ? await this.sendCommand("AUTH", [username, password])
-        : await this.sendCommand("AUTH", [password]);
+        ? await this.sendCommand("AUTH", [username, password], { inline: true })
+        : await this.sendCommand("AUTH", [password], { inline: true });
     } catch (error) {
       if (error instanceof ErrorReplyError) {
         throw new AuthenticationError("Authentication failed", {
@@ -143,7 +150,7 @@ export class RedisConnection implements Connection {
     db: number | undefined = this.options.db,
   ): Promise<void> {
     if (!db) throw new Error("The database index is undefined.");
-    await this.sendCommand("SELECT", [db]);
+    await this.sendCommand("SELECT", [db], { inline: true });
   }
 
   private enqueueCommand(
@@ -160,13 +167,16 @@ export class RedisConnection implements Connection {
     args?: Array<RedisValue>,
     options?: SendCommandOptions,
   ): Promise<RedisReply> {
-    const { promise, resolve, reject } = Promise.withResolvers<RedisReply>();
     const execute = () =>
       this.#protocol.sendCommand(
         command,
         args ?? kEmptyRedisArgs,
         options?.returnUint8Arrays,
       );
+    if (options?.inline) {
+      return execute();
+    }
+    const { promise, resolve, reject } = Promise.withResolvers<RedisReply>();
     this.enqueueCommand({ execute, resolve, reject });
 
     return promise;

tests/reconnect_test.ts

@@ -0,0 +1,36 @@
+import { assertEquals } from "../vendor/https/deno.land/std/assert/mod.ts";
+import {
+  beforeAll,
+  describe,
+  it,
+} from "../vendor/https/deno.land/std/testing/bdd.ts";
+import { newClient, nextPort, startRedis, stopRedis } from "./test_util.ts";
+
+describe("reconnect", () => {
+  let port!: number;
+  beforeAll(() => {
+    port = nextPort();
+  });
+
+  it("auto reconnect", async () => {
+    let server = await startRedis({ port });
+    const client = await newClient({ hostname: "127.0.0.1", port });
+    assertEquals(await client.ping(), "PONG");
+    await stopRedis(server);
+    server = await startRedis({ port });
+    assertEquals(await client.ping(), "PONG");
+    client.close();
+    await stopRedis(server);
+  });
+
+  it("auto reconnect, with db spec", async () => {
+    let server = await startRedis({ port });
+    const client = await newClient({ hostname: "127.0.0.1", port, db: 1 });
+    assertEquals(await client.ping(), "PONG");
+    await stopRedis(server);
+    server = await startRedis({ port });
+    assertEquals(await client.ping(), "PONG");
+    client.close();
+    await stopRedis(server);
+  });
+});