From 45725e31a698c108be3df0dd8db8adc1101ea83d Mon Sep 17 00:00:00 2001
From: Chiezo <hasundue@gmail.com>
Date: Tue, 17 Oct 2023 16:26:42 +0900
Subject: [PATCH] refactor!(lib/events): modify the interface of EventPublisher
 (#21)

---
 core/streams.ts      |  1 -
 deno.lock            |  1 +
 lib/events.ts        | 22 ++++++---------------
 lib/events_test.ts   | 47 ++++++++++++++++++++++++++++++++++++++++++++
 lib/x/streamtools.ts |  1 +
 5 files changed, 55 insertions(+), 17 deletions(-)
 create mode 100644 lib/events_test.ts

diff --git a/core/streams.ts b/core/streams.ts
index 43de4f0..0b2348c 100644
--- a/core/streams.ts
+++ b/core/streams.ts
@@ -10,7 +10,6 @@ import { Lock } from "./x/async.ts";
 export class NonExclusiveWritableStream<W = unknown> extends EventTarget
   implements WritableStream<W> {
   readonly locked = false;
-
   readonly #writer: Lock<WritableStreamDefaultWriter<W>>;
 
   constructor(
diff --git a/deno.lock b/deno.lock
index f996195..0b31ba5 100644
--- a/deno.lock
+++ b/deno.lock
@@ -133,6 +133,7 @@
     "https://deno.land/x/emit@0.30.0/mod.ts": "817cb45fcd94d15a43de826161049eee89e3be4bcf06b3a2f60e8cf459e559f1",
     "https://deno.land/x/esbuild@v0.19.4/mod.js": "6277018cfbcad3912fd346409e0b2a9807cf10c9555a15e4aac299b3194fa4fb",
     "https://deno.land/x/streamtools@v0.5.0/collect.ts": "d421e9dcca5ec7b30b731797305efc9b986aa4cf411bf95aec661dc7b1e1ec11",
+    "https://deno.land/x/streamtools@v0.5.0/pipe_through_from.ts": "67b0c908e5dcb748ced483e2b31a1ec08d46fc839298b07bcd854ab4b39a4e2a",
     "https://deno.land/x/streamtools@v0.5.0/pop.ts": "9ee4bbce7409e902a39bfcc0e49eaaab0008492dfba5f6679db28df6b3867299",
     "https://deno.land/x/wasmbuild@0.14.1/cache.ts": "89eea5f3ce6035a1164b3e655c95f21300498920575ade23161421f5b01967f4",
     "https://deno.land/x/wasmbuild@0.14.1/loader.ts": "d98d195a715f823151cbc8baa3f32127337628379a02d9eb2a3c5902dbccfc02"
diff --git a/lib/events.ts b/lib/events.ts
index ba894c2..37b3308 100644
--- a/lib/events.ts
+++ b/lib/events.ts
@@ -2,8 +2,6 @@ import type {
   ClientToRelayMessage,
   EventContentFor,
   EventKind,
-  PrivateKey,
-  RelayLike,
   TagFor,
 } from "../mod.ts";
 import { Stringified } from "../core/types.ts";
@@ -14,23 +12,15 @@ export interface EventInit<K extends EventKind = EventKind> {
   content: EventContentFor<K> | Stringified<EventContentFor<K>>;
 }
 
-import { Signer } from "./signs.ts";
+import type { Signer } from "./signs.ts";
 
 export class EventPublisher<K extends EventKind = EventKind>
-  extends WritableStream<EventInit<K>> {
-  #signer: Signer;
-  #messenger: WritableStreamDefaultWriter<ClientToRelayMessage>;
-
-  constructor(relayLike: RelayLike, nsec: PrivateKey) {
+  extends TransformStream<EventInit<K>, ClientToRelayMessage<"EVENT", K>> {
+  constructor(readonly signer: Signer) {
     super({
-      write: (event) => this.publish(event),
-      close: () => this.#messenger.releaseLock(),
+      transform: (init, controller) => {
+        controller.enqueue(["EVENT", this.signer.sign(init)]);
+      },
     });
-    this.#signer = new Signer(nsec);
-    this.#messenger = relayLike.getWriter();
-  }
-
-  publish<K extends EventKind>(event: EventInit<K>): Promise<void> {
-    return this.#messenger.write(["EVENT", this.#signer.sign(event)]);
   }
 }
diff --git a/lib/events_test.ts b/lib/events_test.ts
new file mode 100644
index 0000000..98e5f98
--- /dev/null
+++ b/lib/events_test.ts
@@ -0,0 +1,47 @@
+import { afterAll, beforeAll, describe, it } from "./std/testing.ts";
+import { assertEquals, assertInstanceOf } from "./std/assert.ts";
+import { pipeThroughFrom } from "./x/streamtools.ts";
+import type {} from "../core/protocol.d.ts";
+import { Relay } from "../core/relays.ts?nips=1";
+import { PrivateKey, Signer } from "./signs.ts";
+import { EventInit, EventPublisher } from "./events.ts";
+import { MockWebSocket } from "../lib/testing.ts";
+
+describe("EventPublisher", () => {
+  let relay: Relay;
+  let signer: Signer;
+  let publisher: EventPublisher<1>;
+
+  beforeAll(() => {
+    globalThis.WebSocket = MockWebSocket;
+    signer = new Signer(PrivateKey.generate());
+  });
+  afterAll(() => {
+    relay?.close();
+  });
+
+  it("should be a TransformStream", () => {
+    publisher = new EventPublisher(signer);
+    assertInstanceOf(publisher, TransformStream);
+  });
+
+  // FIXME: dangling promise
+  it.ignore("should transform EventInit to ClientToRelayMessage", async () => {
+    const { readable, writable } = publisher;
+    const reader = readable.getReader();
+    const writer = writable.getWriter();
+    const init = { kind: 1, content: "hello" } satisfies EventInit<1>;
+    await writer.write(init);
+    const { value } = await reader.read();
+    assertEquals(value, ["EVENT", signer.sign(init)]);
+    await writer.close();
+    await reader.cancel();
+  });
+
+  // FIXME: runtime error
+  it.ignore("should be connectable to a relay", () => {
+    relay = new Relay("wss://example.com");
+    const writable = pipeThroughFrom(relay, publisher);
+    assertInstanceOf(writable, WritableStream);
+  });
+});
diff --git a/lib/x/streamtools.ts b/lib/x/streamtools.ts
index beff2ad..f3b47dc 100644
--- a/lib/x/streamtools.ts
+++ b/lib/x/streamtools.ts
@@ -1,2 +1,3 @@
 export { collect } from "https://deno.land/x/streamtools@v0.5.0/collect.ts";
 export { pop } from "https://deno.land/x/streamtools@v0.5.0/pop.ts";
+export { pipeThroughFrom } from "https://deno.land/x/streamtools@v0.5.0/pipe_through_from.ts";