feat: support health check (#368)

Author: uki00a

connection.ts

@@ -37,6 +37,10 @@ export interface RedisConnectionOptions {
    */
   maxRetryCount?: number;
   backoff?: Backoff;
+  /**
+   * When this option is set, a `PING` command is sent every specified number of seconds.
+   */
+  healthCheckInterval?: number;
 }
 
 const kEmptyRedisArgs: Array<RedisValue> = [];
@@ -164,6 +168,8 @@ export class RedisConnection implements Connection {
         this.close();
         throw error;
       }
+
+      this.#enableHealthCheckIfNeeded();
     } catch (error) {
       if (error instanceof AuthenticationError) {
         throw (error.cause ?? error);
@@ -253,6 +259,31 @@ export class RedisConnection implements Connection {
   private isManuallyClosedByUser(): boolean {
     return this._isClosed && !this._isConnected;
   }
+
+  #enableHealthCheckIfNeeded() {
+    const { healthCheckInterval } = this.options;
+    if (healthCheckInterval == null) {
+      return;
+    }
+
+    const ping = async () => {
+      if (this.isManuallyClosedByUser()) {
+        return;
+      }
+
+      try {
+        await this.sendCommand("PING");
+        this._isConnected = true;
+      } catch {
+        // TODO: notify the user of an error
+        this._isConnected = false;
+      } finally {
+        setTimeout(ping, healthCheckInterval);
+      }
+    };
+
+    setTimeout(ping, healthCheckInterval);
+  }
 }
 
 class AuthenticationError extends Error {}

tests/commands/connection.ts

@@ -2,6 +2,7 @@ import { connect, createLazyClient } from "../../mod.ts";
 import {
   assert,
   assertEquals,
+  assertExists,
   assertNotEquals,
 } from "../../vendor/https/deno.land/std/testing/asserts.ts";
 import {
@@ -10,6 +11,7 @@ import {
   describe,
   it,
 } from "../../vendor/https/deno.land/std/testing/bdd.ts";
+import { delay } from "../../vendor/https/deno.land/std/async/delay.ts";
 import { newClient } from "../test_util.ts";
 import type { TestServer } from "../test_util.ts";
 import type { Redis } from "../../mod.ts";
@@ -68,6 +70,34 @@ export function connectionTests(
     });
   });
 
+  describe("health check", () => {
+    it("should send a ping every `healthCheckInterval`", async () => {
+      const opts = {
+        ...getOpts(),
+        healthCheckInterval: 10,
+      };
+      const client = await connect(opts);
+      const rawPreviousCommandStats = await client.info("commandstats");
+      await delay(25);
+      const rawCurrentCommandStats = await client.info("commandstats");
+      client.close();
+
+      await delay(10); // NOTE: After closing the connection, no errors should occur
+
+      const previousPingStats =
+        parseCommandStats(rawPreviousCommandStats)["ping"];
+      const currentPingStats =
+        parseCommandStats(rawCurrentCommandStats)["ping"];
+      assertExists(previousPingStats);
+      assertExists(currentPingStats);
+
+      const previousCallCount = previousPingStats["calls"];
+      const currentCallCount = currentPingStats["calls"];
+      const d = currentCallCount - previousCallCount;
+      assert(d >= 2, `${d} should be greater than or equal to 2`);
+    });
+  });
+
   describe("createLazyClient", () => {
     it("returns the lazily connected client", async () => {
       const opts = getOpts();
@@ -114,3 +144,29 @@ export function connectionTests(
     });
   });
 }
+
+function parseCommandStats(
+  stats: string,
+): Record<string, Record<string, number>> {
+  return stats.split("\r\n").reduce((statsByCommand, line) => {
+    if (line.startsWith("#") || line.length === 0) {
+      return statsByCommand;
+    }
+
+    const [section, details] = line.split(":");
+    assertExists(section);
+    assertExists(details);
+    const sectionPrefix = "cmdstat_";
+    assert(section.startsWith(sectionPrefix));
+    const command = section.slice(sectionPrefix.length);
+    statsByCommand[command] = details.split(",").reduce((stats, attr) => {
+      const [key, value] = attr.split("=");
+      assertExists(key);
+      assertExists(value);
+      stats[key] = parseInt(value);
+      return stats;
+    }, {} as Record<string, number>);
+
+    return statsByCommand;
+  }, {} as Record<string, Record<string, number>>);
+}