beta 6

.tool-versions

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

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/package.json

@@ -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

@@ -2,6 +2,11 @@ import * as BehaviorSubject from "./BehaviorSubject.js";
 
 export { Fiber as t };
 
+/**
+ * Keeps track of all active fibers.
+ */
+const table = new Map<string, Fiber>();
+
 /**
  * A fiber's state.
  */
@@ -31,7 +36,9 @@ export class Fiber {
      * A unique identifier for this fiber.
      */
     public readonly id: string = crypto.randomUUID()
-  ) {}
+  ) {
+    table.set(id, this);
+  }
 
   /**
    * Terminates the fiber. The fiber's state will transition to `Terminated` and
@@ -40,13 +47,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/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,8 +38,19 @@ 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 subscription = observable.subscribe(observer);
 
+      if (metadata) {
+        Fiber.get(metadata.clientAgentId)?.stateChange.subscribe((state) => {
+          switch (state) {
+            case Fiber.State.Terminated:
+              console.log("terminated");
+              subscription.unsubscribe();
+          }
+        });
+      }
+
       return {
         unsubscribe: async () => 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"]
     });