Introduce SUBSCRIBE and PSUBSCRIBE Command (#1360)

* introduce SUBSCRIBE command

* refactor event handler

* finalize interface

* update tests

* cleanup

* replace EventEmitter class for runtime compat

* add: streamOptions in commandOptions

* introduce PSUBSCRIBE command

* introduce docs link refs

* improve type coverage

Author: fahreddinozcan

pkg/commands/command.ts

@@ -31,6 +31,35 @@ export type CommandOptions<TResult, TData> = {
    */
   automaticDeserialization?: boolean;
   latencyLogging?: boolean;
+  /**
+   * Additional headers to be sent with the request
+   */
+  headers?: Record<string, string>;
+
+  /**
+   * Path to append to the URL
+   */
+  path?: string[];
+
+  /**
+   * Options for streaming requests, mainly used for subscribe, monitor commands
+   **/
+  streamOptions?: {
+    /**
+     * Callback to be called when a message is received
+     */
+    onMessage?: (data: string) => void;
+
+    /**
+     * Whether the request is streaming
+     */
+    isStreaming?: boolean;
+
+    /**
+     * Signal to abort the request
+     */
+    signal?: AbortSignal;
+  };
 };
 /**
  * Command offers default (de)serialization and the exec method to all commands.
@@ -42,6 +71,11 @@ export class Command<TResult, TData> {
   public readonly command: (string | number | boolean)[];
   public readonly serialize: Serialize;
   public readonly deserialize: Deserialize<TResult, TData>;
+  protected readonly headers?: Record<string, string>;
+  protected readonly path?: string[];
+  protected readonly onMessage?: (data: string) => void;
+  protected readonly isStreaming: boolean;
+  protected readonly signal?: AbortSignal;
   /**
    * Create a new command instance.
    *
@@ -58,6 +92,11 @@ export class Command<TResult, TData> {
         : (x) => x as unknown as TData;
 
     this.command = command.map((c) => this.serialize(c));
+    this.headers = opts?.headers;
+    this.path = opts?.path;
+    this.onMessage = opts?.streamOptions?.onMessage;
+    this.isStreaming = opts?.streamOptions?.isStreaming ?? false;
+    this.signal = opts?.streamOptions?.signal;
 
     if (opts?.latencyLogging) {
       const originalExec = this.exec.bind(this);
@@ -83,7 +122,12 @@ export class Command<TResult, TData> {
   public async exec(client: Requester): Promise<TData> {
     const { result, error } = await client.request<TResult>({
       body: this.command,
+      path: this.path,
       upstashSyncToken: client.upstashSyncToken,
+      headers: this.headers,
+      onMessage: this.onMessage,
+      isStreaming: this.isStreaming,
+      signal: this.signal,
     });
 
     if (error) {

pkg/commands/psubscribe.test.ts

@@ -0,0 +1,169 @@
+import { expect, test, describe } from "bun:test";
+import { Redis } from "../redis";
+import { newHttpClient } from "../test-utils";
+
+interface TestMessage {
+  user?: string;
+  msg?: string;
+  message?: string;
+  timestamp?: number;
+  type?: string;
+}
+
+describe("Pattern Subscriber", () => {
+  const client = newHttpClient();
+  const redis = new Redis(client);
+
+  test("receives pattern matched messages", async () => {
+    const pattern = "user:*";
+    const receivedMessages: TestMessage[] = [];
+
+    const subscriber = redis.psubscribe<TestMessage>([pattern]);
+    subscriber.on("pmessage", ({ message }) => {
+      receivedMessages.push(message);
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    const testMessage: TestMessage = {
+      user: "testUser",
+      message: "Hello, World!",
+      timestamp: Date.now(),
+    };
+
+    await redis.publish("user:123", testMessage);
+    await redis.publish("user:456", testMessage);
+    await redis.publish("other:789", testMessage); // Should not receive this
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    expect(receivedMessages).toHaveLength(2); // Only messages from user:* channels
+    expect(receivedMessages[0]).toEqual(testMessage);
+    expect(receivedMessages[1]).toEqual(testMessage);
+
+    await subscriber.unsubscribe();
+  }, 10_000);
+
+  test("handles pattern-specific messages with channel info", async () => {
+    const pattern = "chat:*:messages";
+    const messages: { pattern: string; channel: string; message: TestMessage }[] = [];
+
+    const subscriber = redis.psubscribe<TestMessage>([pattern]);
+    subscriber.on("pmessage", (data) => {
+      messages.push(data);
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    await redis.publish("chat:room1:messages", { msg: "Hello Room 1" });
+    await redis.publish("chat:room2:messages", { msg: "Hello Room 2" });
+    await redis.publish("chat:room1:users", { msg: "User joined" }); // Should not receive this
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    expect(messages).toHaveLength(2);
+    expect(messages[0].pattern).toBe("chat:*:messages");
+    expect(messages[0].channel).toBe("chat:room1:messages");
+    expect(messages[0].message).toEqual({ msg: "Hello Room 1" });
+    expect(messages[1].channel).toBe("chat:room2:messages");
+    expect(messages[1].message).toEqual({ msg: "Hello Room 2" });
+
+    await subscriber.unsubscribe();
+  }, 10_000);
+
+  test("handles multiple patterns", async () => {
+    const patterns = ["user:*", "chat:*"];
+    const messages: Record<string, Array<{ channel: string; message: TestMessage }>> = {
+      "user:*": [],
+      "chat:*": [],
+    };
+
+    const subscriber = redis.psubscribe<TestMessage>(patterns);
+    subscriber.on("pmessage", ({ pattern, channel, message }) => {
+      messages[pattern].push({ channel, message });
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    await redis.publish("user:123", { type: "user" });
+    await redis.publish("chat:room1", { type: "chat" });
+    await redis.publish("other:xyz", { type: "other" }); // Should not receive this
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    expect(messages["user:*"]).toHaveLength(1);
+    expect(messages["chat:*"]).toHaveLength(1);
+    expect(messages["user:*"][0].channel).toBe("user:123");
+    expect(messages["chat:*"][0].channel).toBe("chat:room1");
+
+    await subscriber.unsubscribe();
+  }, 10_000);
+
+  test("unsubscribe from specific pattern", async () => {
+    const patterns = ["user:*", "chat:*"];
+    const messages: Record<string, Array<{ channel: string; message: TestMessage }>> = {
+      "user:*": [],
+      "chat:*": [],
+    };
+
+    const subscriber = redis.psubscribe<TestMessage>(patterns);
+    subscriber.on("pmessage", ({ pattern, channel, message }) => {
+      messages[pattern].push({ channel, message });
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    // Initial messages
+    await redis.publish("user:123", { msg: "user1" });
+    await redis.publish("chat:room1", { msg: "chat1" });
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    expect(messages["user:*"]).toHaveLength(1);
+    expect(messages["chat:*"]).toHaveLength(1);
+
+    // Unsubscribe from user:* pattern
+    await subscriber.unsubscribe(["user:*"]);
+    expect(subscriber.getSubscribedChannels()).toEqual(["chat:*"]);
+
+    // Clear messages
+    messages["user:*"] = [];
+    messages["chat:*"] = [];
+
+    // Send more messages
+    await redis.publish("user:123", { msg: "user2" });
+    await redis.publish("chat:room1", { msg: "chat2" });
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    expect(messages["user:*"]).toHaveLength(0); // Should not receive any more user messages
+    expect(messages["chat:*"]).toHaveLength(1); // Should still receive chat messages
+    expect(messages["chat:*"][0].message.msg).toBe("chat2");
+
+    await subscriber.unsubscribe();
+  }, 15_000);
+
+  test("pattern and regular subscriptions work together", async () => {
+    const patternSubscriber = redis.psubscribe<TestMessage>(["user:*"]);
+    const channelSubscriber = redis.subscribe<TestMessage>(["user:123"]);
+
+    const patternMessages: TestMessage[] = [];
+    const channelMessages: TestMessage[] = [];
+
+    patternSubscriber.on("pmessage", ({ message }) => {
+      patternMessages.push(message);
+    });
+
+    channelSubscriber.on("message", ({ message }) => {
+      channelMessages.push(message);
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    const testMessage: TestMessage = { msg: "Hello" };
+    await redis.publish("user:123", testMessage);
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    expect(patternMessages).toHaveLength(1);
+    expect(channelMessages).toHaveLength(1);
+    expect(patternMessages[0]).toEqual(testMessage);
+    expect(channelMessages[0]).toEqual(testMessage);
+
+    await Promise.all([patternSubscriber.unsubscribe(), channelSubscriber.unsubscribe()]);
+  }, 15_000);
+});

pkg/commands/psubscribe.ts

@@ -0,0 +1,25 @@
+import { Command, type CommandOptions } from "./command";
+
+/**
+ * @see https://redis.io/commands/psubscribe
+ */
+export class PSubscribeCommand extends Command<number, number> {
+  constructor(cmd: [...patterns: string[]], opts?: CommandOptions<number, number>) {
+    const sseHeaders = {
+      Accept: "text/event-stream",
+      "Cache-Control": "no-cache",
+      Connection: "keep-alive",
+    };
+
+    super([], {
+      ...opts,
+      headers: sseHeaders,
+      path: ["psubscribe", ...cmd],
+      streamOptions: {
+        isStreaming: true,
+        onMessage: opts?.streamOptions?.onMessage,
+        signal: opts?.streamOptions?.signal,
+      },
+    });
+  }
+}