Implement retriable connection

At the moment, a client will die if the server is killed or the connection is lost. This is a problem especially with clients subscribed to a channel.
Since the Writer/Reader buffers are passed around to I/O functions, it is impossible to reconnect a Redis client.
Having a connection object allows to easily replace the buffers.

Resolves: denodrivers#83

README.md

@@ -35,6 +35,28 @@ const sub = await redis.subscribe("channel");
 
 ## Advanced Usage
 
+### Retriable connection
+
+By default, a client's connection will throw an error if the server dies or the network becomes unavailable.
+A connection can be made "retriable" by setting the value `maxRetryCount` when connecting a new client.
+
+```ts
+const redis = await connect({ ...options, maxRetryCount: 10 });
+
+// The client will try to connect to the server 10 times if the server dies or the network becomes unavailable.
+```
+
+The property is set automatically to `10` when creating a subscriber client.
+After a reconnection succeeds, the client will subscribe again to all the channels and patterns.
+
+```ts
+const redis = await connect(options);
+const subscriberClient = await redis.subscribe("channel");
+
+// The client's connection will now be forced to try to connect to the server 10 times if the server dies or the network
+//   becomes unavailable.
+```
+
 ### Execute raw commands
 
 `redis.executor` is raw level [redis protocol](https://redis.io/topics/protocol) executor.

command.ts

@@ -535,5 +535,6 @@ export type RedisCommands = {
   ): Promise<[BulkString, BulkString[]]>;
 
   readonly isClosed: boolean;
+  readonly isConnected: boolean;
   close(): void;
 };

connection.ts

@@ -0,0 +1,191 @@
+import { BufReader, BufWriter } from "./vendor/https/deno.land/std/io/bufio.ts";
+import {
+  CommandExecutor,
+  CommandFunc,
+  RedisRawReply,
+  muxExecutor,
+  sendCommand,
+} from "./io.ts";
+
+// type Reader = Deno.Reader;
+// type Writer = Deno.Writer;
+type Closer = Deno.Closer;
+
+export type RedisConnectOptions = {
+  hostname?: string;
+  port?: number | string;
+  tls?: boolean;
+  db?: number;
+  password?: string;
+  name?: string;
+  maxRetryCount?: number;
+};
+
+export class RedisConnection {
+  name: string | null = null;
+  closer: Closer | null = null;
+  reader: BufReader | null = null;
+  writer: BufWriter | null = null;
+
+  executor: CommandExecutor | null = null;
+
+  get exec(): CommandFunc<RedisRawReply> {
+    return this.executor!.exec;
+  }
+
+  private _isConnected: boolean = false;
+
+  get isConnected(): boolean {
+    return this._isConnected;
+  }
+
+  private _isClosed: boolean = false;
+
+  get isClosed(): boolean {
+    return this._isClosed;
+  }
+
+  maxRetryCount = 0;
+  private retryCount = 0;
+
+  private connectThunkified: () => Promise<RedisConnection>;
+  private thunkifyConnect(
+    hostname: string,
+    port: string | number,
+    options: RedisConnectOptions,
+  ): () => Promise<RedisConnection> {
+    return async () => {
+      const dialOpts: Deno.ConnectOptions = {
+        hostname,
+        port: parsePortLike(port),
+      };
+      if (!Number.isSafeInteger(dialOpts.port)) {
+        throw new Error("deno-redis: opts.port is invalid");
+      }
+      const conn: Deno.Conn = options?.tls
+        ? await Deno.connectTls(dialOpts)
+        : await Deno.connect(dialOpts);
+
+      if (options.name) this.name = options.name;
+      if (options.maxRetryCount) this.maxRetryCount = options.maxRetryCount;
+
+      this.closer = conn;
+      this.reader = new BufReader(conn);
+      this.writer = new BufWriter(conn);
+      this.executor = muxExecutor(this, this.maxRetryCount > 0);
+
+      this._isClosed = false;
+      this._isConnected = true;
+
+      if (options?.password) this.authenticate(options.password);
+      if (options?.db) this.selectDb(options.db);
+
+      return this as RedisConnection;
+    };
+  }
+
+  constructor(
+    hostname: string,
+    port: number | string,
+    private options: RedisConnectOptions,
+  ) {
+    this.connectThunkified = this.thunkifyConnect(hostname, port, options);
+  }
+
+  private authenticate(
+    password: string | undefined = this.options.password,
+  ): Promise<RedisRawReply> {
+    if (!password) throw new Error("The password is undefined.");
+
+    const readerAsBuffer = this.reader as BufReader;
+    const writerAsBuffer = this.writer as BufWriter;
+
+    return sendCommand(writerAsBuffer, readerAsBuffer, "AUTH", password);
+  }
+
+  private selectDb(
+    databaseIndex: number | undefined = this.options.db,
+  ): Promise<RedisRawReply> {
+    if (!databaseIndex) throw new Error("The database index is undefined.");
+
+    const readerAsBuffer = this.reader as BufReader;
+    const writerAsBuffer = this.writer as BufWriter;
+
+    return sendCommand(writerAsBuffer, readerAsBuffer, "SELECT", databaseIndex);
+  }
+
+  close() {
+    this._isClosed = true;
+    this._isConnected = false;
+    try {
+      this.closer!.close();
+    } catch (error) {
+      if (!(error instanceof Deno.errors.BadResource)) throw error;
+    }
+  }
+
+  /**
+   * Connect to Redis server
+   */
+  async connect(): Promise<RedisConnection> {
+    return this.connectThunkified();
+  }
+
+  async reconnect(): Promise<RedisConnection> {
+    const readerAsBuffer = this.reader as BufReader;
+    const writerAsBuffer = this.writer as BufWriter;
+    if (!readerAsBuffer.peek(1)) throw new Error("Client is closed.");
+
+    try {
+      await sendCommand(writerAsBuffer, readerAsBuffer, "PING");
+      this._isConnected = true;
+
+      return Promise.resolve(this);
+    } catch (error) {
+      this._isConnected = false;
+      return new Promise(
+        (resolve, reject) => {
+          const interval = setInterval(
+            async () => {
+              if (this.retryCount > this.maxRetryCount) {
+                await this.close();
+                reject(new Error("Could not reconnect"));
+              }
+
+              try {
+                await this.close();
+                await this.connect();
+
+                await sendCommand(
+                  this.writer as BufWriter,
+                  this.reader as BufReader,
+                  "PING",
+                );
+
+                this._isConnected = true;
+                this.retryCount = 0;
+                clearInterval(interval);
+                resolve(this);
+              } finally {
+                this.retryCount++;
+              }
+            },
+            1200,
+          );
+        },
+      );
+    }
+  }
+}
+
+function parsePortLike(port: string | number | undefined): number {
+  if (typeof port === "string") {
+    return parseInt(port);
+  } else if (typeof port === "number") {
+    return port;
+  } else if (port === undefined) {
+    return 6379;
+  } else {
+    throw new Error("port is invalid: typeof=" + typeof port);
+  }
+}

io.ts

@@ -8,6 +8,7 @@ import {
   Deferred,
 } from "./vendor/https/deno.land/std/async/mod.ts";
 import { ConditionalArray, Bulk, Integer, Status, Raw } from "./command.ts";
+import { RedisConnection } from "./connection.ts";
 
 export type StatusReply = ["status", Status];
 export type IntegerReply = ["integer", Integer];
@@ -175,8 +176,8 @@ function tryParseErrorReply(line: string): never {
 }
 
 export function muxExecutor(
-  r: BufReader,
-  w: BufWriter,
+  connection: RedisConnection,
+  attemptReconnect: boolean = false,
 ): CommandExecutor {
   let queue: {
     command: string;
@@ -187,12 +188,31 @@ export function muxExecutor(
   function dequeue(): void {
     const [e] = queue;
     if (!e) return;
-    sendCommand(w, r, e.command, ...e.args)
+    sendCommand(
+      connection.writer as BufWriter,
+      connection.reader as BufReader,
+      e.command,
+      ...e.args,
+    )
       .then((v) => {
-        // console.log(e.command, e.args, v);
         e.d.resolve(v);
       })
-      .catch((err) => e.d.reject(err))
+      .catch(async (err) => {
+        if (
+          (
+            // Error `BadResource` is thrown when an attempt is made to write to a closed connection,
+            //  Make sure that the connection wasn't explicitly closed by the user before trying to reconnect.
+            (err instanceof Deno.errors.BadResource && !connection.isClosed) ||
+            err instanceof Deno.errors.BrokenPipe ||
+            err instanceof Deno.errors.ConnectionAborted ||
+            err instanceof Deno.errors.ConnectionRefused ||
+            err instanceof Deno.errors.ConnectionReset
+          ) &&
+          attemptReconnect
+        ) {
+          await connection.reconnect();
+        } else e.d.reject(err);
+      })
       .finally(() => {
         queue.shift();
         dequeue();

pipeline.ts

@@ -1,7 +1,4 @@
-import {
-  BufReader,
-  BufWriter,
-} from "./vendor/https/deno.land/std/io/bufio.ts";
+import { BufReader, BufWriter } from "./vendor/https/deno.land/std/io/bufio.ts";
 import {
   createRequest,
   readReply,
@@ -16,6 +13,7 @@ import {
   Deferred,
 } from "./vendor/https/deno.land/std/async/mod.ts";
 import { RedisCommands } from "./command.ts";
+import { RedisConnection } from "./connection.ts";
 
 const encoder = new TextEncoder();
 export type RawReplyOrError = RedisRawReply | ErrorReply;
@@ -25,8 +23,7 @@ export type RedisPipeline = {
 } & RedisCommands;
 
 export function createRedisPipeline(
-  writer: BufWriter,
-  reader: BufReader,
+  connection: RedisConnection,
   opts?: { tx: true },
 ): RedisPipeline {
   let commands: string[] = [];
@@ -49,12 +46,12 @@ export function createRedisPipeline(
 
   async function send(cmds: string[]): Promise<RawReplyOrError[]> {
     const msg = cmds.join("");
-    await writer.write(encoder.encode(msg));
-    await writer.flush();
+    await connection.writer!.write(encoder.encode(msg));
+    await connection.writer!.flush();
     const ret: RawReplyOrError[] = [];
     for (let i = 0; i < cmds.length; i++) {
       try {
-        const rep = await readReply(reader);
+        const rep = await readReply(connection.reader!);
         ret.push(rep);
       } catch (e) {
         if (e instanceof ErrorReplyError) {