feat: add experimental/web_streams_connection module (#405)

`std/io` has been deprecated in [v0.203.0](https://github.com/denoland/deno_std/releases/tag/0.203.0). So, we need to migrate from `Deno.Reader`/`Deno.Writer` to Web Streams API in near future. In preparation for it, this commit experimentally adds support for RESP based on Web Streams API.

benchmark/benchmark.js

@@ -4,6 +4,7 @@ import { dirname, join } from "node:path";
 export function run({
   driver,
   client,
+  outputFilename = driver,
 }) {
   const encoder = new TextEncoder();
   return suite(
@@ -67,7 +68,7 @@ export function run({
       await client.flushdb();
     }),
     save({
-      file: driver,
+      file: outputFilename,
       details: true,
       folder: join(
         dirname(dirname(new URL(import.meta.url).pathname)),

benchmark/deno-redis.ts

@@ -1,12 +1,29 @@
 import { run } from "./benchmark.js";
 import { connect } from "../mod.ts";
+import { connect as connectWebStreams } from "../experimental/web_streams_connection/mod.ts";
 
-const redis = await connect({ hostname: "127.0.0.1" });
-try {
-  await run({
-    client: redis,
-    driver: "deno-redis",
-  });
-} finally {
-  await redis.quit();
+{
+  const redis = await connect({ hostname: "127.0.0.1" });
+  try {
+    await run({
+      client: redis,
+      driver: "deno-redis",
+    });
+  } finally {
+    await redis.quit();
+  }
+}
+
+{
+  const redis = await connectWebStreams({ hostname: "127.0.0.1" });
+  try {
+    await run({
+      client: redis,
+      driver: "deno-redis (experimental/web_streams_connection)",
+      outputFilename:
+        "deno-redis-with-experimental-web-streams-based-connection",
+    });
+  } finally {
+    await redis.quit();
+  }
 }

command.ts

@@ -8,7 +8,7 @@ import type {
   Raw,
   RedisValue,
   SimpleString,
-} from "./protocol/mod.ts";
+} from "./protocol/shared/types.ts";
 import type { RedisPipeline } from "./pipeline.ts";
 import type { RedisSubscription } from "./pubsub.ts";
 import type {

connection.ts

@@ -1,18 +1,19 @@
-import { readReply, sendCommand, sendCommands } from "./protocol/mod.ts";
-import type { RedisReply, RedisValue } from "./protocol/mod.ts";
-import type { Command } from "./protocol/command.ts";
+import { Protocol as DenoStreamsProtocol } from "./protocol/deno_streams/mod.ts";
+import type { RedisReply, RedisValue } from "./protocol/shared/types.ts";
+import type { Command, Protocol } from "./protocol/shared/protocol.ts";
 import type { Backoff } from "./backoff.ts";
 import { exponentialBackoff } from "./backoff.ts";
 import { ErrorReplyError, isRetriableError } from "./errors.ts";
-import { kUnstablePipeline, kUnstableReadReply } from "./internal/symbols.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 {
+  kUnstableCreateProtocol,
+  kUnstablePipeline,
+  kUnstableReadReply,
+} from "./internal/symbols.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;
 
 export interface SendCommandOptions {
   /**
@@ -61,6 +62,11 @@ export interface RedisConnectionOptions {
    * When this option is set, a `PING` command is sent every specified number of seconds.
    */
   healthCheckInterval?: number;
+
+  /**
+   * @private
+   */
+  [kUnstableCreateProtocol]?: (conn: Deno.Conn) => Protocol;
 }
 
 export const kEmptyRedisArgs: Array<RedisValue> = [];
@@ -74,9 +80,6 @@ interface PendingCommand {
 
 export class RedisConnection implements Connection {
   name: string | null = null;
-  private reader!: BufReader;
-  private writer!: BufWriter;
-  private closer!: Closer;
   private maxRetryCount = 10;
 
   private readonly hostname: string;
@@ -86,6 +89,8 @@ export class RedisConnection implements Connection {
   private backoff: Backoff;
 
   private commandQueue: PendingCommand[] = [];
+  #conn!: Deno.Conn;
+  #protocol!: Protocol;
 
   get isClosed(): boolean {
     return this._isClosed;
@@ -160,11 +165,11 @@ export class RedisConnection implements Connection {
   }
 
   [kUnstableReadReply](returnsUint8Arrays?: boolean): Promise<RedisReply> {
-    return readReply(this.reader, returnsUint8Arrays);
+    return this.#protocol.readReply(returnsUint8Arrays);
   }
 
   [kUnstablePipeline](commands: Array<Command>) {
-    return sendCommands(this.writer, this.reader, commands);
+    return this.#protocol.pipeline(commands);
   }
 
   /**
@@ -184,9 +189,9 @@ export class RedisConnection implements Connection {
         ? await Deno.connectTls(dialOpts)
         : await Deno.connect(dialOpts);
 
-      this.closer = conn;
-      this.reader = new BufReader(conn);
-      this.writer = new BufWriter(conn);
+      this.#conn = conn;
+      this.#protocol = this.options?.[kUnstableCreateProtocol]?.(conn) ??
+        new DenoStreamsProtocol(conn);
       this._isClosed = false;
       this._isConnected = true;
 
@@ -222,16 +227,13 @@ export class RedisConnection implements Connection {
     this._isClosed = true;
     this._isConnected = false;
     try {
-      this.closer!.close();
+      this.#conn!.close();
     } catch (error) {
       if (!(error instanceof Deno.errors.BadResource)) throw error;
     }
   }
 
   async reconnect(): Promise<void> {
-    if (!this.reader.peek(1)) {
-      throw new Error("Client is closed.");
-    }
     try {
       await this.sendCommand("PING");
       this._isConnected = true;
@@ -247,9 +249,7 @@ export class RedisConnection implements Connection {
     if (!command) return;
 
     try {
-      const reply = await sendCommand(
-        this.writer,
-        this.reader,
+      const reply = await this.#protocol.sendCommand(
         command.name,
         command.args,
         command.returnUint8Arrays,
@@ -269,9 +269,7 @@ export class RedisConnection implements Connection {
         try {
           await this.connect();
 
-          const reply = await sendCommand(
-            this.writer,
-            this.reader,
+          const reply = await this.#protocol.sendCommand(
             command.name,
             command.args,
             command.returnUint8Arrays,

deno.json

@@ -8,8 +8,11 @@
       "benchmark/ioredis.js"
     ]
   },
+  "test": {
+    "exclude": ["benchmark/", "tmp/", "vendor/"]
+  },
   "tasks": {
-    "test": "deno test --allow-net --allow-read=tests --allow-write=tests/tmp --allow-run=redis-server,redis-cli --coverage=coverage --trace-ops ./tests",
+    "test": "deno test --allow-net --allow-read=tests --allow-write=tests/tmp --allow-run=redis-server,redis-cli --coverage=coverage --trace-ops",
     "test:doc": "deno test --doc --no-run --import-map=import_map.test.json",
     "coverage": "deno coverage ./coverage --lcov --output=coverage/lcov.info",
     "make_mod": "deno run --allow-read --allow-write --allow-run --check tools/make_mod.ts",

errors.ts

@@ -5,6 +5,11 @@ export class ConnectionClosedError extends Error {}
 export class SubscriptionClosedError extends Error {}
 
 export class ErrorReplyError extends Error {}
+export class NotImplementedError extends Error {
+  constructor(message?: string) {
+    super(message ? `Not implemented: ${message}` : "Not implemented");
+  }
+}
 
 export class InvalidStateError extends Error {
   constructor(message?: string) {
@@ -19,5 +24,6 @@ export function isRetriableError(error: Error): boolean {
     error instanceof Deno.errors.ConnectionAborted ||
     error instanceof Deno.errors.ConnectionRefused ||
     error instanceof Deno.errors.ConnectionReset ||
+    error instanceof Deno.errors.UnexpectedEof ||
     error instanceof EOFError);
 }

executor.ts

@@ -1,5 +1,5 @@
 import type { Connection, SendCommandOptions } from "./connection.ts";
-import type { RedisReply, RedisValue } from "./protocol/mod.ts";
+import type { RedisReply, RedisValue } from "./protocol/shared/types.ts";
 
 export interface CommandExecutor {
   readonly connection: Connection;

experimental/cluster/mod.ts

@@ -28,7 +28,7 @@ import type { RedisConnectOptions } from "../../redis.ts";
 import type { CommandExecutor } from "../../executor.ts";
 import type { Connection, SendCommandOptions } from "../../connection.ts";
 import type { Redis } from "../../redis.ts";
-import type { RedisReply, RedisValue } from "../../protocol/mod.ts";
+import type { RedisReply, RedisValue } from "../../protocol/shared/types.ts";
 import { ErrorReplyError } from "../../errors.ts";
 import { delay } from "../../vendor/https/deno.land/std/async/delay.ts";
 import calculateSlot from "../../vendor/https/esm.sh/cluster-key-slot/lib/index.js";

experimental/web_streams_connection/mod.ts

@@ -0,0 +1,15 @@
+import { kUnstableCreateProtocol } from "../../internal/symbols.ts";
+import type { RedisConnectOptions } from "../../redis.ts";
+import { connect as _connect } from "../../redis.ts";
+import { Protocol } from "../../protocol/web_streams/mod.ts";
+
+function createProtocol(conn: Deno.Conn) {
+  return new Protocol(conn);
+}
+
+export function connect(options: RedisConnectOptions) {
+  return _connect({
+    ...options,
+    [kUnstableCreateProtocol]: createProtocol,
+  });
+}

internal/buffered_readable_stream.ts

@@ -0,0 +1,56 @@
+import { concateBytes } from "./concate_bytes.ts";
+
+const LF = "\n".charCodeAt(0);
+/**
+ * Wraps `ReadableStream` to provide buffering. Heavily inspired by `deno_std/io/buf_reader.ts`.
+ *
+ * {@link https://github.com/denoland/deno_std/blob/0.204.0/io/buf_reader.ts}
+ */
+export class BufferedReadableStream {
+  #reader: ReadableStreamDefaultReader<Uint8Array>;
+  #buffer: Uint8Array;
+  constructor(readable: ReadableStream<Uint8Array>) {
+    // TODO: This class could probably be optimized with a BYOB reader.
+    this.#reader = readable.getReader();
+    this.#buffer = new Uint8Array(0);
+  }
+
+  async readLine(): Promise<Uint8Array> {
+    const i = this.#buffer.indexOf(LF);
+    if (i > -1) {
+      return this.#consume(i + 1);
+    }
+    for (;;) {
+      await this.#fill();
+      const i = this.#buffer.indexOf(LF);
+      if (i > -1) return this.#consume(i + 1);
+    }
+  }
+
+  async readFull(buffer: Uint8Array): Promise<void> {
+    if (buffer.length <= this.#buffer.length) {
+      buffer.set(this.#consume(buffer.length));
+      return;
+    }
+    for (;;) {
+      await this.#fill();
+      if (this.#buffer.length >= buffer.length) break;
+    }
+    return this.readFull(buffer);
+  }
+
+  #consume(n: number): Uint8Array {
+    const b = this.#buffer.subarray(0, n);
+    this.#buffer = this.#buffer.subarray(n);
+    return b;
+  }
+
+  async #fill() {
+    const chunk = await this.#reader.read();
+    if (chunk.done) {
+      throw new Deno.errors.BadResource();
+    }
+    const bytes = chunk.value;
+    this.#buffer = concateBytes(this.#buffer, bytes);
+  }
+}