Beta 6 (#9)

.tool-versions

@@ -1 +1 @@
-bun 1.0.17
+bun 1.1.21

packages/browser/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@daniel-nagy/transporter-browser",
   "type": "module",
-  "version": "1.0.0-beta.5",
+  "version": "1.0.0-beta.6",
   "description": "Typesafe distributed computing in the browser.",
   "author": "Daniel Nagy <1622446+daniel-nagy@users.noreply.github.com>",
   "repository": "github:daniel-nagy/transporter",

packages/browser/web-test-runner.config.js

@@ -2,8 +2,8 @@ import { esbuildPlugin } from "@web/dev-server-esbuild";
 import { playwrightLauncher } from "@web/test-runner-playwright";
 import ts from "typescript";
 
-import tsConfigBase from "./tsconfig-base.json" assert { type: "json" };
-import tsConfigTest from "./tsconfig-test.json" assert { type: "json" };
+import tsConfigBase from "./tsconfig-base.json" with { type: "json" };
+import tsConfigTest from "./tsconfig-test.json" with { type: "json" };
 
 /**
  * @type import("@web/test-runner").TestRunnerConfig

packages/core/README.md

@@ -784,9 +784,9 @@ The Metadata module allows information to be extracted from a proxy.
 ```ts
 type Metadata = {
   /**
-   * The address of the server that provides the value.
+   * The id of the client agent managing this proxy.
    */
-  address: string;
+  clientAgentId: string;
   /**
    * The path to the value in the original object.
    */

packages/core/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@daniel-nagy/transporter",
   "type": "module",
-  "version": "1.0.0-beta.5",
+  "version": "1.0.0-beta.6",
   "description": "Typesafe distributed computing in TypeScript.",
   "author": "Daniel Nagy <1622446+daniel-nagy@users.noreply.github.com>",
   "repository": "github:daniel-nagy/transporter",
@@ -46,7 +46,7 @@
     "@types/sinonjs__fake-timers": "^8.1.5",
     "@typescript-eslint/eslint-plugin": "^6.7.5",
     "@typescript-eslint/parser": "^6.7.5",
-    "bun-types": "^1.0.6",
+    "bun-types": "^1.1.21",
     "eslint": "^8.51.0",
     "eslint-plugin-expect-type": "^0.2.3",
     "eslint-plugin-require-extensions": "^0.1.3",

packages/core/src/ClientAgent.ts

@@ -131,7 +131,7 @@ export class ClientAgent extends Fiber.t {
     const children: Record<string, unknown> = {};
 
     const meta: Metadata.t = {
-      address: this.serverAddress,
+      clientAgentId: this.id,
       objectPath: path
     };
 

packages/core/src/Fiber.ts

@@ -1,7 +1,13 @@
 import * as BehaviorSubject from "./BehaviorSubject.js";
+import * as UUID from "./UUID.js";
 
 export { Fiber as t };
 
+/**
+ * Keeps track of all active fibers.
+ */
+const table = new Map<string, Fiber>();
+
 /**
  * A fiber's state.
  */
@@ -30,8 +36,10 @@ export class Fiber {
     /**
      * A unique identifier for this fiber.
      */
-    public readonly id: string = crypto.randomUUID()
-  ) {}
+    public readonly id: string = UUID.v4()
+  ) {
+    table.set(id, this);
+  }
 
   /**
    * Terminates the fiber. The fiber's state will transition to `Terminated` and
@@ -40,13 +48,21 @@ export class Fiber {
   terminate() {
     this.#state.next(State.Terminated);
     this.#state.complete();
+    table.delete(this.id);
   }
 
   [Symbol.dispose]() {
     this.terminate();
   }
 }
 
+/**
+ * Get a reference to a Fiber from its id.
+ */
+export const get = <T extends Fiber>(id: string): T | null => {
+  return (table.get(id) as T) ?? null;
+};
+
 /**
  * Creates a new `Fiber`.
  */

packages/core/src/Message.ts

@@ -1,4 +1,5 @@
 import * as JsObject from "./JsObject.js";
+import * as UUID from "./UUID.js";
 
 /**
  * Flattens an intersection type into a single type.
@@ -89,7 +90,7 @@ export type CallFunction<Args> = FlattenIntersection<
 export const CallFunction = <Args>({
   address,
   args,
-  id = crypto.randomUUID(),
+  id = UUID.v4(),
   noReply = false,
   path
 }: {
@@ -122,7 +123,7 @@ export type Error<Error> = FlattenIntersection<
 export const Error = <T>({
   address,
   error,
-  id = crypto.randomUUID()
+  id = UUID.v4()
 }: {
   address: string;
   error: T;
@@ -147,7 +148,7 @@ export type GarbageCollect = FlattenIntersection<
  */
 export const GarbageCollect = ({
   address,
-  id = crypto.randomUUID()
+  id = UUID.v4()
 }: {
   address: string;
   id?: string;
@@ -171,7 +172,7 @@ export type SetValue<Value> = FlattenIntersection<
  */
 export const SetValue = <T>({
   address,
-  id = crypto.randomUUID(),
+  id = UUID.v4(),
   value
 }: {
   address: string;

packages/core/src/Metadata.ts

@@ -9,9 +9,9 @@ export const symbol = Symbol.for("metadata");
  */
 export type Metadata = {
   /**
-   * The address of the server that provides the value.
+   * The id of the client agent managing this proxy.
    */
-  address: string;
+  clientAgentId: string;
   /**
    * The path to the value in the original object from the dereferenced value.
    */

packages/core/src/PubSub.ts

@@ -1,3 +1,5 @@
+import * as Fiber from "./Fiber.js";
+import * as Metadata from "./Metadata.js";
 import * as Observable from "./Observable/index.js";
 
 /**
@@ -36,10 +38,22 @@ export function from<T>(observable: Observable.ObservableLike<T>): PubSub<T> {
     subscribe: async (
       observer: AsyncObserver<T> | ((value: T) => Promise<void>)
     ) => {
+      const metadata = Metadata.get(observer);
+      const agent = metadata && Fiber.get(metadata.clientAgentId);
       const subscription = observable.subscribe(observer);
 
+      const innerSubscription = agent?.stateChange.subscribe((state) => {
+        switch (state) {
+          case Fiber.State.Terminated:
+            subscription.unsubscribe();
+        }
+      });
+
       return {
-        unsubscribe: async () => subscription.unsubscribe()
+        unsubscribe: async () => {
+          innerSubscription?.unsubscribe();
+          subscription.unsubscribe();
+        }
       };
     }
   };

packages/core/src/Session.test.ts

@@ -1,6 +1,7 @@
 import { afterEach, describe, expect, test } from "bun:test";
 import { spy, spyOn } from "tinyspy";
 
+import * as BehaviorSubject from "./BehaviorSubject.js";
 import * as Cache from "./Cache.js";
 import * as Fiber from "./Fiber.js";
 import * as Injector from "./Injector.js";
@@ -9,9 +10,12 @@ import * as Message from "./Message.js";
 import * as Metadata from "./Metadata.js";
 import * as Observable from "./Observable/index.js";
 import * as Proxy from "./Proxy.js";
+import * as PubSub from "./PubSub.js";
 import * as Session from "./Session.js";
 import * as Subprotocol from "./Subprotocol.js";
 
+const UUID = expect.stringMatching(/\w{8}-\w{4}-\w{4}-\w{4}-\w{12}/) as string;
+
 afterEach(() => {
   Session.rootSupervisor.tasks.forEach((task) => task.terminate());
 });
@@ -256,12 +260,12 @@ describe("proxied objects", () => {
     const { proxy, dispose } = expose({ bar: async () => {} });
 
     expect(Metadata.get(proxy)).toEqual({
-      address: "",
+      clientAgentId: UUID,
       objectPath: []
     });
 
     expect(Metadata.get(proxy.bar)).toEqual({
-      address: "",
+      clientAgentId: UUID,
       objectPath: ["bar"]
     });
 
@@ -284,9 +288,7 @@ describe("proxied objects", () => {
     const childProxy = await proxy();
 
     expect(Metadata.get(childProxy)).toEqual({
-      // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-      // @ts-ignore
-      address: expect.stringMatching(/\w{8}-\w{4}-\w{4}-\w{4}-\w{12}/),
+      clientAgentId: UUID,
       objectPath: []
     });
 
@@ -295,12 +297,33 @@ describe("proxied objects", () => {
   });
 });
 
+describe("pub/sub", () => {
+  test("A pub/sub is unsubscribed when the session is terminated", async () => {
+    const counter = BehaviorSubject.of(0);
+    const unsubscribe = spy();
+
+    spyOn(counter, "subscribe", () => ({ unsubscribe }));
+
+    const { dispose, proxy, server } = expose({
+      Chat: { counter: PubSub.from(counter) }
+    });
+
+    proxy.Chat.counter.subscribe(async () => {});
+
+    await scheduleTask();
+    server.terminate();
+
+    expect(unsubscribe.callCount).toBe(1);
+    dispose();
+  });
+});
+
 describe("reflection", () => {
   test("the root proxy", () => {
     const { proxy, dispose } = expose(async () => {});
 
     expect(Metadata.get(proxy)).toEqual({
-      address: "",
+      clientAgentId: UUID,
       objectPath: []
     });
 
@@ -311,7 +334,7 @@ describe("reflection", () => {
     const { proxy, dispose } = expose({ a: async () => {} });
 
     expect(Metadata.get(proxy.a)).toEqual({
-      address: "",
+      clientAgentId: UUID,
       objectPath: ["a"]
     });
 

packages/core/src/Session.ts

@@ -11,6 +11,7 @@ import * as ServerAgent from "./ServerAgent.js";
 import * as Subject from "./Subject.js";
 import * as Subprotocol from "./Subprotocol.js";
 import * as Supervisor from "./Supervisor.js";
+import * as UUID from "./UUID.js";
 
 export { Session as t };
 
@@ -113,7 +114,7 @@ export abstract class Session<
   protected createServer(
     this: Session,
     provide: unknown,
-    address: string = crypto.randomUUID()
+    address: string = UUID.v4()
   ): ServerAgent.t {
     const serverAgent = ServerAgent.init({
       address,

packages/core/src/UUID.test.ts

@@ -0,0 +1,9 @@
+import { expect, test } from "bun:test";
+
+import * as UUID from "./UUID.js";
+
+test("generating a v4 UUID", () => {
+  expect(UUID.v4()).toEqual(
+    expect.stringMatching(/\w{8}-\w{4}-\w{4}-\w{4}-\w{12}/)
+  );
+});

packages/core/src/UUID.ts

@@ -0,0 +1,29 @@
+function hex(bits: number) {
+  if (bits > 53) throw new Error("bits must be less than or equal to 53");
+
+  return Math.floor(Math.random() * (2 ** bits - 1))
+    .toString(16)
+    .padStart(Math.ceil(bits / 4), "0");
+}
+
+/**
+ * Generates a v4 UUID. This implementation generates exactly 122 bits of random
+ * data and concatenates the version and variant.
+ *
+ * @see https://datatracker.ietf.org/doc/html/rfc9562#name-uuid-version-4
+ */
+function v4Fallback(): string {
+  const a = hex(48);
+  const b = hex(12);
+  const c = `${hex(31)}${hex(31)}`;
+
+  return `${a.slice(0, 8)}-${a.slice(8)}-4${b}-a${c.slice(0, 3)}-${c.slice(
+    3,
+    15
+  )}`;
+}
+
+export const v4 =
+  typeof crypto !== "undefined" && typeof crypto.randomUUID !== "undefined"
+    ? crypto.randomUUID.bind(crypto)
+    : v4Fallback;