Require ObservableScopes of state holders to be specified explicitly

Previously we had a ViewModel class which was responsible for little more than creating an ObservableScope. However, since this ObservableScope would be created implicitly upon view model construction, it became a tad bit harder for callers to remember to eventually end the scope (as you wouldn't just have to remember to end ObservableScopes, but also to destroy ViewModels). Requiring the scope to be specified explicitly by the caller also makes it possible for the caller to reuse the scope for other purposes, reducing the number of scopes mentally in flight that need tending to, and for all state holders (not just view models) to be handled uniformly by helper functions such as generateKeyed$.

src/utils/observable.test.ts

@@ -6,9 +6,10 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { test } from "vitest";
+import { Subject } from "rxjs";
 
 import { withTestScheduler } from "./test";
-import { pauseWhen } from "./observable";
+import { generateKeyed$, pauseWhen } from "./observable";
 
 test("pauseWhen", () => {
   withTestScheduler(({ behavior, expectObservable }) => {
@@ -22,3 +23,43 @@ test("pauseWhen", () => {
     ).toBe(outputMarbles);
   });
 });
+
+test("generateKeyed$ has the right output and ends scopes at the right times", () => {
+  const scope1$ = new Subject<string>();
+  const scope2$ = new Subject<string>();
+  const scope3$ = new Subject<string>();
+  const scope4$ = new Subject<string>();
+  const scopeSubjects = [scope1$, scope2$, scope3$, scope4$];
+
+  withTestScheduler(({ hot, expectObservable }) => {
+    // Each scope should start when the input number reaches or surpasses their
+    // number and end when the input number drops back below their number.
+    // At the very end, unsubscribing should end all remaining scopes.
+    const inputMarbles = "       123242";
+    const outputMarbles = "      abcbdb";
+    const subscriptionMarbles = "^-----!";
+    const scope1Marbles = "      y-----n";
+    const scope2Marbles = "      -y----n";
+    const scope3Marbles = "      --ynyn";
+    const scope4Marbles = "      ----yn";
+
+    expectObservable(
+      generateKeyed$(hot<string>(inputMarbles), (input, createOrGet) => {
+        for (let i = 1; i <= +input; i++) {
+          createOrGet(i.toString(), (scope) => {
+            scopeSubjects[i - 1].next("y");
+            scope.onEnd(() => scopeSubjects[i - 1].next("n"));
+            return i.toString();
+          });
+        }
+        return "abcd"[+input - 1];
+      }),
+      subscriptionMarbles,
+    ).toBe(outputMarbles);
+
+    expectObservable(scope1$).toBe(scope1Marbles);
+    expectObservable(scope2$).toBe(scope2Marbles);
+    expectObservable(scope3$).toBe(scope3Marbles);
+    expectObservable(scope4$).toBe(scope4Marbles);
+  });
+});

src/utils/observable.ts

@@ -23,6 +23,7 @@ import {
 } from "rxjs";
 
 import { type Behavior } from "../state/Behavior";
+import { ObservableScope } from "../state/ObservableScope";
 
 const nothing = Symbol("nothing");
 
@@ -117,3 +118,56 @@ export function pauseWhen<T>(pause$: Behavior<boolean>) {
       map(([value]) => value),
     );
 }
+
+/**
+ * Maps a changing input value to an output value consisting of items that have
+ * automatically generated ObservableScopes tied to a key. Items will be
+ * automatically created when their key is requested for the first time, reused
+ * when the same key is requested at a later time, and destroyed (have their
+ * scope ended) when the key is no longer requested.
+ */
+export function generateKeyed$<In, Item, Out>(
+  input$: Observable<In>,
+  project: (
+    input: In,
+    createOrGet: (
+      key: string,
+      factory: (scope: ObservableScope) => Item,
+    ) => Item,
+  ) => Out,
+): Observable<Out> {
+  return input$.pipe(
+    scan<
+      In,
+      {
+        items: Map<string, { item: Item; scope: ObservableScope }>;
+        output: Out;
+      },
+      { items: Map<string, { item: Item; scope: ObservableScope }> }
+    >(
+      (state, data) => {
+        const nextItems = new Map<
+          string,
+          { item: Item; scope: ObservableScope }
+        >();
+        const output = project(data, (key, factory) => {
+          let item = state.items.get(key);
+          if (item === undefined) {
+            const scope = new ObservableScope();
+            item = { item: factory(scope), scope };
+          }
+          nextItems.set(key, item);
+          return item.item;
+        });
+        for (const [key, { scope }] of state.items)
+          if (!nextItems.has(key)) scope.end();
+        return { items: nextItems, output };
+      },
+      { items: new Map() },
+    ),
+    finalizeValue((state) => {
+      for (const { scope } of state.items.values()) scope.end();
+    }),
+    map(({ output }) => output),
+  );
+}

src/utils/test-viewmodel.ts

@@ -27,6 +27,7 @@ import {
   mockMediaDevices,
   mockMuteStates,
   MockRTCSession,
+  testScope,
 } from "./test";
 import { aliceRtcMember, localRtcMember } from "./test-fixtures";
 import { type RaisedHandInfo, type ReactionInfo } from "../reactions";
@@ -134,6 +135,7 @@ export function getBasicCallViewModelEnvironment(
   // const remoteParticipants$ = of([aliceParticipant]);
 
   const vm = new CallViewModel(
+    testScope(),
     rtcSession.asMockedSession(),
     matrixRoom,
     mockMediaDevices({}),

src/utils/test.ts

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
 */
 import { map, type Observable, of, type SchedulerLike } from "rxjs";
 import { type RunHelpers, TestScheduler } from "rxjs/testing";
-import { expect, type MockedObject, vi, vitest } from "vitest";
+import { expect, type MockedObject, onTestFinished, vi, vitest } from "vitest";
 import {
   type RoomMember,
   type Room as MatrixRoom,
@@ -89,6 +89,15 @@ interface TestRunnerGlobal {
   rxjsTestScheduler?: SchedulerLike;
 }
 
+/**
+ * Create a new ObservableScope which ends when the current test ends.
+ */
+export function testScope(): ObservableScope {
+  const scope = new ObservableScope();
+  onTestFinished(() => scope.end());
+  return scope;
+}
+
 /**
  * Run Observables with a scheduler that virtualizes time, for testing purposes.
  */
@@ -267,6 +276,7 @@ export async function withLocalMedia(
   continuation: (vm: LocalUserMediaViewModel) => void | Promise<void>,
 ): Promise<void> {
   const vm = new LocalUserMediaViewModel(
+    testScope(),
     "local",
     mockMatrixRoomMember(localRtcMember, roomMember),
     constant(localParticipant),
@@ -280,11 +290,8 @@ export async function withLocalMedia(
     constant(null),
     constant(null),
   );
-  try {
-    await continuation(vm);
-  } finally {
-    vm.destroy();
-  }
+  // TODO: Simplify to just return the view model
+  await continuation(vm);
 }
 
 export function mockRemoteParticipant(
@@ -308,6 +315,7 @@ export async function withRemoteMedia(
 ): Promise<void> {
   const remoteParticipant = mockRemoteParticipant(participant);
   const vm = new RemoteUserMediaViewModel(
+    testScope(),
     "remote",
     mockMatrixRoomMember(localRtcMember, roomMember),
     of(remoteParticipant),
@@ -321,11 +329,8 @@ export async function withRemoteMedia(
     constant(null),
     constant(null),
   );
-  try {
-    await continuation(vm);
-  } finally {
-    vm.destroy();
-  }
+  // TODO: Simplify to just return the view model
+  await continuation(vm);
 }
 
 export function mockConfig(config: Partial<ResolvedConfigOptions> = {}): void {