feat: implement RESP3Subscription

connection.ts

@@ -1,9 +1,15 @@
 import type { Backoff } from "./backoff.ts";
 import type { ConnectionEventMap } from "./events.ts";
 import type { ErrorReplyError } from "./errors.ts";
+import type {
+  DefaultPubSubMessageType,
+  PubSubMessageType,
+  RedisSubscription,
+} from "./subscription.ts";
 import type { TypedEventTarget } from "./internal/typed_event_target.ts";
 import type {
   kUnstableCreateProtocol,
+  kUnstableCreateSubscription,
   kUnstablePipeline,
   kUnstableProtover,
   kUnstableReadReply,
@@ -40,6 +46,12 @@ export interface Connection extends TypedEventTarget<ConnectionEventMap> {
     args?: Array<RedisValue>,
     options?: SendCommandOptions,
   ): Promise<RedisReply>;
+  /**
+   * @private
+   */
+  [kUnstableCreateSubscription]<
+    TMessage extends PubSubMessageType = DefaultPubSubMessageType,
+  >(): RedisSubscription<TMessage>;
   /**
    * @private
    */

default_client.ts

@@ -6,7 +6,7 @@ import type {
   SubscribeCommand,
 } from "./subscription.ts";
 import type { Connection, SendCommandOptions } from "./connection.ts";
-import { DefaultRedisSubscription } from "./default_subscription.ts";
+import { kUnstableCreateSubscription } from "./internal/symbols.ts";
 import type { RedisReply, RedisValue } from "./protocol/shared/types.ts";
 
 export function createDefaultClient(connection: Connection): Client {
@@ -37,9 +37,9 @@ class DefaultClient implements Client {
     command: SubscribeCommand,
     ...channelsOrPatterns: Array<string>
   ): Promise<RedisSubscription<TMessage>> {
-    const subscription = new DefaultRedisSubscription<TMessage>(
-      this.connection,
-    );
+    const subscription = this.connection[kUnstableCreateSubscription]<
+      TMessage
+    >();
     switch (command) {
       case "SUBSCRIBE":
         await subscription.subscribe(...channelsOrPatterns);

default_connection.ts

@@ -5,14 +5,21 @@ import type {
   RedisConnectionOptions,
   SendCommandOptions,
 } from "./connection.ts";
+import { DefaultRedisSubscription } from "./default_subscription.ts";
 import {
   ErrorReplyError,
   InvalidStateError,
   isRetriableError,
 } from "./errors.ts";
 import type { ConnectionEventMap } from "./events.ts";
+import type {
+  DefaultPubSubMessageType,
+  PubSubMessageType,
+  RedisSubscription,
+} from "./subscription.ts";
 import {
   kUnstableCreateProtocol,
+  kUnstableCreateSubscription,
   kUnstablePipeline,
   kUnstableProtover,
   kUnstableReadReply,
@@ -27,7 +34,11 @@ import {
 import { kEmptyRedisArgs } from "./protocol/shared/command.ts";
 import type { Command, Protocol } from "./protocol/shared/protocol.ts";
 import { Protocol as DenoStreamsProtocol } from "./protocol/deno_streams/mod.ts";
-import type { RedisReply, RedisValue } from "./protocol/shared/types.ts";
+import type {
+  Protover,
+  RedisReply,
+  RedisValue,
+} from "./protocol/shared/types.ts";
 import { delay } from "./deps/std/async.ts";
 
 export function createRedisConnection(
@@ -58,6 +69,8 @@ class RedisConnection
   private commandQueue: PendingCommand[] = [];
   #conn!: Deno.Conn;
   #protocol!: Protocol;
+  /** @default {2} */
+  #protover?: Protover;
   #eventTarget = createTypedEventTarget<ConnectionEventMap>();
   #connectingPromise?: PromiseWithResolvers<void>;
 
@@ -191,6 +204,12 @@ class RedisConnection
     );
   }
 
+  [kUnstableCreateSubscription]<
+    TMessage extends PubSubMessageType = DefaultPubSubMessageType,
+  >(): RedisSubscription<TMessage> {
+    return new DefaultRedisSubscription<TMessage>(this);
+  }
+
   [kUnstableReadReply](returnsUint8Arrays?: boolean): Promise<RedisReply> {
     return this.#protocol.readReply(returnsUint8Arrays);
   }
@@ -292,9 +311,11 @@ class RedisConnection
           await this.authenticate(this.options.username, this.options.password);
         }
         if (this.options[kUnstableProtover] != null) {
-          await this.#sendCommandImmediately("HELLO", [
-            this.options[kUnstableProtover],
-          ]);
+          const protover = this.options[kUnstableProtover];
+          await this.#sendCommandImmediately("HELLO", [protover]);
+          if (protover !== 2) {
+            this.#protover = protover;
+          }
         }
         if (this.options.db) {
           await this.selectDb(this.options.db);

internal/symbols.ts

@@ -27,3 +27,10 @@ export const kUnstableProtover = Symbol("deno-redis.protover");
  * @private
  */
 export const kUnstableStartReadLoop = Symbol("deno-redis.startReadLoop");
+
+/**
+ * @private
+ */
+export const kUnstableCreateSubscription = Symbol(
+  "deno-redis.createSubscription",
+);

pool/client.ts

@@ -9,6 +9,7 @@ import type {
 } from "../subscription.ts";
 import { createDefaultClient } from "../default_client.ts";
 import {
+  kUnstableCreateSubscription,
   kUnstablePipeline,
   kUnstableReadReply,
   kUnstableStartReadLoop,
@@ -100,6 +101,7 @@ function createPoolConnection(
       "addEventListener",
       "removeEventListener",
       Symbol.dispose,
+      kUnstableCreateSubscription,
       kUnstableReadReply,
       kUnstableWriteCommand,
       kUnstablePipeline,

protocol/deno_streams/reply.ts

@@ -1,5 +1,6 @@
 import type { BufReader } from "../../deps/std/io.ts";
 import type * as types from "../shared/types.ts";
+import type { ProtocolEvents } from "../shared/types.ts";
 import {
   ArrayReplyCode,
   AttributeReplyCode,
@@ -19,17 +20,32 @@ import {
 } from "../shared/reply.ts";
 import { EOFError, ErrorReplyError, InvalidStateError } from "../../errors.ts";
 import { decoder } from "../../internal/encoding.ts";
+import type { TypedEventTarget } from "../../internal/typed_event_target.ts";
+import { dispatchEvent } from "../../internal/typed_event_target.ts";
 
-export async function readReply(
+export async function readOrEmitReply(
   reader: BufReader,
+  eventTarget: TypedEventTarget<ProtocolEvents>,
   returnUint8Arrays?: boolean,
 ): Promise<types.RedisReply> {
-  const res = await reader.peek(1);
-  if (res == null) {
-    throw new EOFError();
+  const code = await peekReplyCode(reader);
+  if (code === ErrorReplyCode) {
+    await readErrorReplyOrFail(reader);
   }
 
-  const code = res[0];
+  if (code === PushReplyCode) {
+    const reply = await readPushReply(reader, returnUint8Arrays);
+    dispatchEvent(eventTarget, "push", reply ?? []);
+    return readOrEmitReply(reader, eventTarget, returnUint8Arrays);
+  }
+  return readReply(reader, returnUint8Arrays);
+}
+
+export async function readReply(
+  reader: BufReader,
+  returnUint8Arrays?: boolean,
+): Promise<types.RedisReply> {
+  const code = await peekReplyCode(reader);
   if (code === ErrorReplyCode) {
     await readErrorReplyOrFail(reader);
   }
@@ -73,6 +89,15 @@ export async function readReply(
   }
 }
 
+async function peekReplyCode(reader: BufReader): Promise<number> {
+  const res = await reader.peek(1);
+  if (res == null) {
+    throw new EOFError();
+  }
+  const code = res[0];
+  return code;
+}
+
 async function readIntegerReply(
   reader: BufReader,
 ): Promise<number> {

protocol/shared/types.ts

@@ -50,3 +50,7 @@ export type RawOrError = Raw | ErrorReplyError;
 export const okReply = "OK";
 
 export type Protover = 2 | 3;
+
+export type ProtocolEvents = {
+  push: Array<RedisReply>;
+};

protocol/web_streams/reply.ts

@@ -1,4 +1,5 @@
 import type * as types from "../shared/types.ts";
+import type { ProtocolEvents } from "../shared/types.ts";
 import {
   ArrayReplyCode,
   AttributeReplyCode,
@@ -19,12 +20,38 @@ import {
 import { ErrorReplyError, NotImplementedError } from "../../errors.ts";
 import { decoder } from "../../internal/encoding.ts";
 import type { BufferedReadableStream } from "../../internal/buffered_readable_stream.ts";
+import type { TypedEventTarget } from "../../internal/typed_event_target.ts";
+import { dispatchEvent } from "../../internal/typed_event_target.ts";
+
+export async function readOrEmitReply(
+  readable: BufferedReadableStream,
+  eventTarget: TypedEventTarget<ProtocolEvents>,
+  returnUint8Arrays?: boolean,
+): Promise<types.RedisReply> {
+  const line = await readable.readLine();
+  const code = line[0];
+  if (code === PushReplyCode) {
+    const reply = await parseArrayLikeReply(line, readable, returnUint8Arrays);
+    dispatchEvent(eventTarget, "push", reply ?? []);
+    return readOrEmitReply(readable, eventTarget, returnUint8Arrays);
+  } else {
+    return parseLine(line, readable, returnUint8Arrays);
+  }
+}
 
 export async function readReply(
   readable: BufferedReadableStream,
   returnUint8Arrays?: boolean,
-) {
+): Promise<types.RedisReply> {
   const line = await readable.readLine();
+  return parseLine(line, readable, returnUint8Arrays);
+}
+
+async function parseLine(
+  line: Uint8Array,
+  readable: BufferedReadableStream,
+  returnUint8Arrays?: boolean,
+): Promise<types.RedisReply> {
   const code = line[0];
   switch (code) {
     case ErrorReplyCode: {
@@ -56,16 +83,7 @@ export async function readReply(
     }
     case ArrayReplyCode:
     case PushReplyCode: {
-      const size = Number.parseInt(decoder.decode(line.slice(1)));
-      if (size === -1) {
-        // `-1` indicates a null array
-        return null;
-      }
-      const array: Array<types.RedisReply> = [];
-      for (let i = 0; i < size; i++) {
-        array.push(await readReply(readable, returnUint8Arrays));
-      }
-      return array;
+      return parseArrayLikeReply(line, readable, returnUint8Arrays);
     }
     case MapReplyCode: {
       // NOTE: We treat a map type as an array to keep backward compatibility.
@@ -130,3 +148,20 @@ export async function readReply(
       );
   }
 }
+
+async function parseArrayLikeReply(
+  line: Uint8Array,
+  readable: BufferedReadableStream,
+  returnUint8Arrays?: boolean,
+): Promise<Array<types.RedisReply> | null> {
+  const size = Number.parseInt(decoder.decode(line.slice(1)));
+  if (size === -1) {
+    // `-1` indicates a null array
+    return null;
+  }
+  const array: Array<types.RedisReply> = [];
+  for (let i = 0; i < size; i++) {
+    array.push(await readReply(readable, returnUint8Arrays));
+  }
+  return array;
+}