refactor!: rename RelayPool to RelayGroup and modify the API (#19)

Author: hasundue

core/nodes.ts

@@ -1,6 +1,6 @@
 import type { NostrMessage } from "./protocol.d.ts";
 import type { Logger } from "./types.ts";
-import { WebSocketLike, WebSocketReadyState } from "./websockets.ts";
+import { WebSocketLike } from "./websockets.ts";
 import { NonExclusiveWritableStream } from "./streams.ts";
 
 export interface NostrNodeConfig<
@@ -39,7 +39,7 @@ export class NostrNode<
     this.config.modules.forEach((m) => this.addModule(m));
   }
 
-  get status(): WebSocketReadyState {
+  get status(): WebSocket["readyState"] {
     return this.ws.readyState;
   }
 

core/relays.ts

@@ -37,12 +37,6 @@ export class ConnectionClosed extends Error {}
 // Interfaces
 // ----------------------
 
-export interface RelayLike
-  extends NonExclusiveWritableStream<ClientToRelayMessage> {
-  subscribe: Relay["subscribe"];
-  publish: Relay["publish"];
-}
-
 export interface RelayConfig
   extends NostrNodeConfig<RelayFunctionParameterTypeRecord> {
   url: RelayUrl;
@@ -131,10 +125,6 @@ export class Relay extends NostrNode<
         );
         this.callFunction("startSubscription", { controller, ...context });
       },
-      pull: async () => {
-        await this.ws.ready();
-        // TODO: backpressure
-      },
       cancel: (reason) => {
         return this.callFunction("closeSubscription", { reason, ...context });
       },
@@ -170,6 +160,24 @@ export class Relay extends NostrNode<
   }
 }
 
+// ----------------------
+// RelayLikes
+// ----------------------
+
+export interface RelayLike
+  extends NonExclusiveWritableStream<ClientToRelayMessage> {
+  readonly config: RelayLikeConfig;
+  subscribe: Relay["subscribe"];
+  publish: Relay["publish"];
+}
+
+export type RelayLikeConfig = Omit<
+  RelayConfig,
+  "url" | keyof NostrNodeConfig<RelayFunctionParameterTypeRecord>
+>;
+
+export type RelayLikeOptions = Partial<RelayLikeConfig>;
+
 // ----------------------
 // Events
 // ----------------------

core/websockets.ts

@@ -1,10 +1,8 @@
-import { Notify } from "./x/async.ts";
-
 export type WebSocketEventType = keyof WebSocketEventMap;
 
 export interface WebSocketLike {
   readonly url: string;
-  readonly readyState: WebSocketReadyState;
+  readonly readyState: WebSocket["readyState"];
   send(
     data: string | ArrayBufferLike | Blob | ArrayBufferView,
   ): void | Promise<void>;
@@ -25,14 +23,11 @@ type EventListenerOptionsMap = Map<
  */
 export class LazyWebSocket implements WebSocketLike {
   #ws?: WebSocket;
-  #createWebSocket: () => WebSocket;
-
+  readonly #createWebSocket: () => WebSocket;
   readonly #eventListenerMap = new Map<
     WebSocketEventType,
     EventListenerOptionsMap
   >();
-  readonly #notifier = new Notify();
-
   readonly url: string;
 
   constructor(
@@ -41,11 +36,6 @@ export class LazyWebSocket implements WebSocketLike {
   ) {
     this.#createWebSocket = () => {
       const ws = new WebSocket(url, protocols);
-      for (const type of ["close", "open"] as const) {
-        ws.addEventListener(type, () => {
-          this.#notifier.notifyAll();
-        });
-      }
       this.#eventListenerMap.forEach((map, type) => {
         map.forEach((options, listener) => {
           ws.addEventListener(type, listener, options);
@@ -60,21 +50,26 @@ export class LazyWebSocket implements WebSocketLike {
     return this.#ws ??= this.#createWebSocket();
   }
 
+  #once(type: WebSocketEventType): Promise<void> {
+    return new Promise<void>((resolve) => {
+      this.#created().addEventListener(type, () => resolve(), { once: true });
+    });
+  }
+
   async #ready(): Promise<WebSocket> {
     this.#ws = this.#created();
     switch (this.#ws.readyState) {
       case WebSocket.CONNECTING:
-        await this.#notifier.notified();
-        /* falls through */
-      case WebSocket.OPEN:
         break;
+      case WebSocket.OPEN:
+        return this.#ws;
       case WebSocket.CLOSING:
-        await this.#notifier.notified();
+        await this.#once("close");
         /* falls through */
       case WebSocket.CLOSED:
         this.#ws = this.#createWebSocket();
-        await this.#notifier.notified();
     }
+    await this.#once("open");
     return this.#ws;
   }
 
@@ -93,21 +88,21 @@ export class LazyWebSocket implements WebSocketLike {
     }
     switch (this.#ws.readyState) {
       case WebSocket.CONNECTING:
-        await this.#notifier.notified();
+        await this.#once("open");
         /* falls through */
       case WebSocket.OPEN:
         this.#ws.close(code, reason);
         /* falls through */
       case WebSocket.CLOSING:
-        await this.#notifier.notified();
+        await this.#once("close");
         /* falls through */
       case WebSocket.CLOSED:
         break;
     }
     this.#ws = undefined;
   }
 
-  get readyState(): WebSocketReadyState {
+  get readyState(): WebSocket["readyState"] {
     return this.#ws ? this.#ws.readyState : WebSocket.CLOSED;
   }
 
@@ -138,13 +133,3 @@ export class LazyWebSocket implements WebSocketLike {
     return this.#ws?.dispatchEvent(event) ?? false;
   };
 }
-
-/**
- * The ready state of a WebSocket.
- */
-export enum WebSocketReadyState {
-  CONNECTING = 0,
-  OPEN = 1,
-  CLOSING = 2,
-  CLOSED = 3,
-}

core/x/async.ts

@@ -1,2 +1 @@
-export { Notify } from "https://deno.land/x/async@v2.0.2/notify.ts";
 export { Lock } from "https://deno.land/x/async@v2.0.2/lock.ts";

deno.lock

@@ -2,13 +2,12 @@ import type {
   ClientToRelayMessage,
   EventKind,
   NostrEvent,
-  RelayUrl,
   SubscriptionFilter,
 } from "../core/protocol.d.ts";
 import {
-  Relay,
-  RelayInit,
   RelayLike,
+  RelayLikeConfig,
+  RelayLikeOptions,
   SubscriptionOptions,
 } from "../core/relays.ts";
 import { NonExclusiveWritableStream } from "../core/streams.ts";
@@ -17,26 +16,28 @@ import { Distinctor, merge } from "../lib/streams.ts";
 /**
  * A pool of relays that can be used as a single relay.
  */
-export class RelayPool extends NonExclusiveWritableStream<ClientToRelayMessage>
+export class RelayGroup extends NonExclusiveWritableStream<ClientToRelayMessage>
   implements RelayLike {
-  readonly relays: Relay[];
-
-  #relays_read: Relay[];
-
-  constructor(...init: (RelayUrl | RelayInit)[]) {
-    const relays = init.map((i) => new Relay(i));
+  readonly config: Readonly<RelayLikeConfig>;
+  #relays_read: RelayLike[];
+  #relays_write: RelayLike[];
 
+  constructor(readonly relays: RelayLike[], options?: RelayLikeOptions) {
     const writers = relays.filter((r) => r.config.write)
       .map((r) => r.getWriter());
-
     super({
       async write(msg) {
         await Promise.all(writers.map((r) => r.write(msg)));
       },
-    }, { highWaterMark: Math.max(...relays.map((r) => r.config.nbuffer)) });
-
-    this.relays = relays;
+    });
+    this.config = {
+      name: relays.map((r) => r.config.name).join(", "),
+      read: true,
+      write: true,
+      ...options,
+    };
     this.#relays_read = this.relays.filter((r) => r.config.read);
+    this.#relays_write = this.relays.filter((r) => r.config.write);
   }
 
   // ----------------------
@@ -54,7 +55,7 @@ export class RelayPool extends NonExclusiveWritableStream<ClientToRelayMessage>
   async publish<K extends EventKind>(
     msg: NostrEvent<K>,
   ) {
-    await Promise.all(this.relays.map((r) => r.publish(msg)));
+    await Promise.all(this.#relays_write.map((r) => r.publish(msg)));
   }
 
   // ----------------------

lib/pools.ts lib/relays.ts

@@ -0,0 +1,75 @@
+import { afterAll, beforeAll, describe, it } from "../lib/std/testing.ts";
+import { assertEquals, assertInstanceOf } from "../lib/std/assert.ts";
+import { NostrEvent } from "../core/protocol.d.ts";
+import { Relay } from "../core/relays.ts?nips=1";
+import { RelayGroup } from "../lib/relays.ts";
+import { MockWebSocket } from "../lib/testing.ts";
+
+describe("RelayGroup", () => {
+  let relays: Relay[];
+  let group: RelayGroup;
+  let sub: ReadableStream<NostrEvent>;
+
+  // ----------------------
+  // Setup
+  // ----------------------
+  beforeAll(() => {
+    globalThis.WebSocket = MockWebSocket;
+    relays = [
+      new Relay("ws://localhost:80", {
+        name: "relay-1",
+        read: true,
+        write: true,
+      }),
+      new Relay("ws://localhost:81", {
+        name: "relay-2",
+        read: true,
+        write: false,
+      }),
+      new Relay("ws://localhost:82", {
+        name: "relay-3",
+        read: false,
+        write: true,
+      }),
+    ];
+  });
+  afterAll(() => {
+    group.close();
+  });
+
+  // ----------------------
+  // Constructor
+  // ----------------------
+  it("should create a group of relays", () => {
+    group = new RelayGroup(relays);
+    assertInstanceOf(group, RelayGroup);
+  });
+  it("should not have a url", () => {
+    // @ts-expect-error RelayGroup does not have a url
+    assertEquals(group.url, undefined);
+  });
+  it("should have a default name", () => {
+    assertEquals(group.config.name, "relay-1, relay-2, relay-3");
+  });
+  it("should have a custom name if provided", () => {
+    const group = new RelayGroup(relays, { name: "custom" });
+    assertEquals(group.config.name, "custom");
+  });
+  it("should have default read and write config", () => {
+    assertEquals(group.config.read, true);
+    assertEquals(group.config.write, true);
+  });
+  it("should have custom read and write config if provided", () => {
+    const group = new RelayGroup(relays, { read: false, write: false });
+    assertEquals(group.config.read, false);
+    assertEquals(group.config.write, false);
+  });
+
+  // ----------------------
+  // Subscription
+  // ----------------------
+  it("should create a subscription", () => {
+    sub = group.subscribe({ kinds: [1] }, { id: "test-group" });
+    assertInstanceOf(sub, ReadableStream);
+  });
+});

lib/relays_test.ts

@@ -1,12 +1,9 @@
-export { mergeReadableStreams as merge } from "./std/streams.ts";
-
 /**
  * TransformStream which filters out duplicate values from a stream.
  */
 export class Distinctor<R = unknown, T = unknown>
   extends TransformStream<R, R> {
   #seen: Set<T>;
-
   constructor(protected readonly fn: (value: R) => T) {
     super({
       transform: (value, controller) => {
@@ -35,3 +32,29 @@ export class Transformer<R = unknown, W = unknown>
     });
   }
 }
+
+export function merge<T>(
+  ...streams: ReadableStream<T>[]
+) {
+  const readers = streams.map((r) => r.getReader());
+  return new ReadableStream<T>({
+    async pull(controller) {
+      await Promise.any(readers.map(async (r) => {
+        const { value, done } = await r.read();
+        if (done) {
+          readers.splice(readers.indexOf(r), 1);
+          r.releaseLock();
+        }
+        if (value) {
+          controller.enqueue(value);
+        }
+      })).catch((e) => {
+        if (e instanceof AggregateError) {
+          controller.close();
+        } else {
+          throw e;
+        }
+      });
+    },
+  });
+}