Merge branch 'livekit' into toger5/pseudonomous-identities

src/utils/observable.test.ts

@@ -5,11 +5,12 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import { test } from "vitest";
-import { Subject } from "rxjs";
+import { expect, test } from "vitest";
+import { type Observable, of, Subject, switchMap } from "rxjs";
 
 import { withTestScheduler } from "./test";
-import { generateItems, pauseWhen } from "./observable";
+import { filterBehavior, generateItems, pauseWhen } from "./observable";
+import { type Behavior } from "../state/Behavior";
 
 test("pauseWhen", () => {
   withTestScheduler(({ behavior, expectObservable }) => {
@@ -72,3 +73,31 @@ test("generateItems", () => {
     expectObservable(scope4$).toBe(scope4Marbles);
   });
 });
+
+test("filterBehavior", () => {
+  withTestScheduler(({ behavior, expectObservable }) => {
+    // Filtering the input should segment it into 2 modes of non-null behavior.
+    const inputMarbles = "   abcxabx";
+    const filteredMarbles = "a--xa-x";
+
+    const input$ = behavior(inputMarbles, {
+      a: "a",
+      b: "b",
+      c: "c",
+      x: null,
+    });
+    const filtered$: Observable<Behavior<string> | null> = input$.pipe(
+      filterBehavior((value) => typeof value === "string"),
+    );
+
+    expectObservable(filtered$).toBe(filteredMarbles, {
+      a: expect.any(Object),
+      x: null,
+    });
+    expectObservable(
+      filtered$.pipe(
+        switchMap((value$) => (value$ === null ? of(null) : value$)),
+      ),
+    ).toBe(inputMarbles, { a: "a", b: "b", c: "c", x: null });
+  });
+});

src/utils/observable.ts

@@ -22,6 +22,7 @@ import {
   withLatestFrom,
   BehaviorSubject,
   type OperatorFunction,
+  distinctUntilChanged,
 } from "rxjs";
 
 import { type Behavior } from "../state/Behavior";
@@ -185,6 +186,28 @@ export function generateItemsWithEpoch<
   );
 }
 
+/**
+ * Segments a behavior into periods during which its value matches the filter
+ * (outputting a behavior with a narrowed type) and periods during which it does
+ * not match (outputting null).
+ */
+export function filterBehavior<T, S extends T>(
+  predicate: (value: T) => value is S,
+): OperatorFunction<T, Behavior<S> | null> {
+  return (input$) =>
+    input$.pipe(
+      scan<T, BehaviorSubject<S> | null>((acc$, input) => {
+        if (predicate(input)) {
+          const output$ = acc$ ?? new BehaviorSubject(input);
+          output$.next(input);
+          return output$;
+        }
+        return null;
+      }, null),
+      distinctUntilChanged(),
+    );
+}
+
 function generateItemsInternal<
   Input,
   Keys extends [unknown, ...unknown[]],

src/utils/test-viewmodel.ts

@@ -37,6 +37,7 @@ import {
 import { aliceRtcMember, localRtcMember } from "./test-fixtures";
 import { type RaisedHandInfo, type ReactionInfo } from "../reactions";
 import { constant } from "../state/Behavior";
+import { MatrixRTCMode } from "../settings/settings";
 
 mockConfig({ livekit: { livekit_service_url: "https://example.com" } });
 
@@ -162,6 +163,7 @@ export function getBasicCallViewModelEnvironment(
           setE2EEEnabled: async () => Promise.resolve(),
         }),
       connectionState$: constant(ConnectionState.Connected),
+      matrixRTCMode$: constant(MatrixRTCMode.Legacy),
       ...callViewModelOptions,
     },
     handRaisedSubject$,

src/utils/test.ts

@@ -321,12 +321,12 @@ export function mockLocalParticipant(
 }
 
 export function createLocalMedia(
-  localRtcMember: CallMembership,
+  rtcMember: CallMembership,
   roomMember: Partial<RoomMember>,
   localParticipant: LocalParticipant,
   mediaDevices: MediaDevices,
 ): LocalUserMediaViewModel {
-  const member = mockMatrixRoomMember(localRtcMember, roomMember);
+  const member = mockMatrixRoomMember(rtcMember, roomMember);
   return new LocalUserMediaViewModel(
     testScope(),
     "local",
@@ -361,23 +361,26 @@ export function mockRemoteParticipant(
 }
 
 export function createRemoteMedia(
-  localRtcMember: CallMembership,
+  rtcMember: CallMembership,
   roomMember: Partial<RoomMember>,
-  participant: Partial<RemoteParticipant>,
+  participant: RemoteParticipant | null,
+  livekitRoom: LivekitRoom | undefined = mockLivekitRoom(
+    {},
+    {
+      remoteParticipants$: of(participant ? [participant] : []),
+    },
+  ),
 ): RemoteUserMediaViewModel {
-  const member = mockMatrixRoomMember(localRtcMember, roomMember);
-  const remoteParticipant = mockRemoteParticipant(participant);
+  const member = mockMatrixRoomMember(rtcMember, roomMember);
   return new RemoteUserMediaViewModel(
     testScope(),
     "remote",
     member.userId,
-    of(remoteParticipant),
+    constant(participant),
     {
       kind: E2eeType.PER_PARTICIPANT,
     },
-    constant(
-      mockLivekitRoom({}, { remoteParticipants$: of([remoteParticipant]) }),
-    ),
+    constant(livekitRoom),
     constant("https://rtc-example.org"),
     constant(false),
     constant(member.rawDisplayName ?? "nodisplayname"),