Reduce ConnectionManager to a ConnectionMap

createConnectionMap$ now returns the behavior of connections and their
participants directly and takes an already resolved local transport
instead of an observable of it. The CallViewModel wraps it with mapScoped
so that a new map is built, in a fresh scope, whenever the local transport
changes, and swallows local transport errors there since the local member
already surfaces them in the UI.

src/room/InCallView.tsx

@@ -694,15 +694,13 @@ export const InCallView: FC<InCallViewProps> = ({
             tab={settingsTab}
             onTabChange={setSettingsTab}
             developerSettingsVm={developerSettingsVm}
-            livekitRooms={allConnections
-              .getConnections()
-              .map((connectionItem) => ({
-                room: connectionItem.livekitRoom,
-                livekitAlias: connectionItem.livekitAlias,
-                // TODO compute is local or tag it in the livekit room items already
-                isLocal: undefined,
-                url: connectionItem.transport.livekit_service_url,
-              }))}
+            livekitRooms={allConnections.map((connectionItem) => ({
+              room: connectionItem.livekitRoom,
+              livekitAlias: connectionItem.livekitAlias,
+              // TODO compute is local or tag it in the livekit room items already
+              isLocal: undefined,
+              url: connectionItem.transport.livekit_service_url,
+            }))}
           />
         </>
       )}

src/state/CallViewModel/CallViewModel.ts

@@ -39,6 +39,9 @@ import {
   timer,
   takeUntil,
   from,
+  concatWith,
+  catchError,
+  switchAll,
 } from "rxjs";
 import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger";
 import {
@@ -56,6 +59,7 @@ import {
   createToggle$,
   filterBehavior,
   generateItems,
+  mapScoped,
   pauseWhen,
 } from "../../utils/observable";
 import {
@@ -105,7 +109,7 @@ import {
   type OneOnOneMedia,
 } from "../layoutMedia.ts";
 import { ElementCallError, UnknownCallError } from "../../utils/errors.ts";
-import { type Epoch, type ObservableScope } from "../ObservableScope.ts";
+import { Epoch, type ObservableScope } from "../ObservableScope.ts";
 import { createHomeserverConnected$ } from "./localMember/HomeserverConnected.ts";
 import {
   createLocalMembership$,
@@ -123,9 +127,9 @@ import {
   ECConnectionFactory,
 } from "./remoteMembers/ConnectionFactory.ts";
 import {
-  type ConnectionManagerData,
-  createConnectionManager$,
-} from "./remoteMembers/ConnectionManager.ts";
+  ConnectionMap,
+  createConnectionMap$,
+} from "./remoteMembers/ConnectionMap.ts";
 import {
   createRemoteMatrixLivekitMembers$,
   type LocalMatrixLivekitMember,
@@ -371,7 +375,7 @@ export interface CallViewModel {
    * key is generated when someone joins or leaves.
    */
   keyRotationSuppressed$: Behavior<boolean>;
-  allConnections$: Behavior<ConnectionManagerData>;
+  allConnections$: Behavior<Connection[]>;
   /** Participants sorted by livekit room so they can be used in the audio rendering */
   livekitRoomItems$: Behavior<LivekitRoomItem[]>;
   /** use the layout instead, this is just for the sdk export. */
@@ -577,14 +581,24 @@ export function createCallViewModel$(
       options.livekitRoomFactory,
     );
 
-  const connectionManager = createConnectionManager$({
-    scope: scope,
-    connectionFactory: connectionFactory,
-    localTransport$,
-    remoteTransports$: membershipsAndTransports.transports$,
-    logger: logger,
-    ownMembershipIdentity,
-  });
+  const connectionMap$ = scope.behavior<Epoch<ConnectionMap>>(
+    localTransport$.pipe(
+      catchError(() => NEVER), // Swallow errors
+      concatWith(NEVER), // So the Observable doesn't complete prematurely
+      mapScoped("connectionMap$", (localTransport, scope) =>
+        createConnectionMap$({
+          scope,
+          connectionFactory,
+          localTransport,
+          remoteTransports$: membershipsAndTransports.transports$,
+          logger,
+          ownMembershipIdentity,
+        }),
+      ),
+      switchAll(),
+    ),
+    new Epoch(new ConnectionMap()),
+  );
 
   const remoteMatrixLivekitMembers$: Behavior<
     Epoch<RemoteMatrixLivekitMember[]>
@@ -592,7 +606,7 @@ export function createCallViewModel$(
     scope: scope,
     membershipsWithTransport$:
       membershipsAndTransports.membershipsWithTransport$,
-    connectionManager: connectionManager,
+    connectionMap$,
     localUser: { userId, deviceId },
   });
 
@@ -631,7 +645,7 @@ export function createCallViewModel$(
         controlledAudioDevices,
       );
     },
-    connectionManager,
+    connectionMap$,
     client,
     matrixRTCSession,
     localTransport$,
@@ -717,7 +731,7 @@ export function createCallViewModel$(
   });
 
   const allConnections$ = scope.behavior(
-    connectionManager.connectionManagerData$.pipe(map((d) => d.value)),
+    connectionMap$.pipe(map((d) => d.value.getConnections())),
   );
 
   const livekitRoomItems$ = scope.behavior<LivekitRoomItem[]>(

src/state/CallViewModel/localMember/LocalMember.test.ts

@@ -21,7 +21,7 @@ import {
   beforeEach,
   afterEach,
 } from "vitest";
-import { BehaviorSubject, map, of, Subject } from "rxjs";
+import { BehaviorSubject, of, Subject } from "rxjs";
 import { logger } from "matrix-js-sdk/lib/logger";
 import {
   ConnectionState as LivekitConnectionState,
@@ -57,7 +57,7 @@ import {
 import { MatrixRTCTransportMissingError } from "../../../utils/errors";
 import { Epoch, ObservableScope } from "../../ObservableScope";
 import { constant } from "../../Behavior";
-import { ConnectionManagerData } from "../remoteMembers/ConnectionManager";
+import { ConnectionMap } from "../remoteMembers/ConnectionMap";
 import {
   type ConnectionState,
   type Connection,
@@ -261,25 +261,11 @@ describe("LocalMembership", () => {
 
   it("throws error on missing RTC config error", () => {
     withTestScheduler(({ scope, hot, expectObservable }) => {
-      const localTransport$ = scope.behavior<null | LivekitTransport>(
-        hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")),
-        null,
-      );
-
-      // we do not need any connection data since we want to fail before reaching that.
-      const mockConnectionManager = {
-        transports$: scope.behavior(
-          localTransport$.pipe(map((t) => new Epoch([t]))),
-        ),
-        connectionManagerData$: constant(
-          new Epoch(new ConnectionManagerData()),
-        ),
-      };
-
       const localMembership = createLocalMembership$({
         scope,
         ...defaultCreateLocalMemberValues,
-        connectionManager: mockConnectionManager,
+        // we do not need any connection data since we want to fail before reaching that.
+        connectionMap$: constant(new Epoch(new ConnectionMap())),
         localTransport$: hot(
           "1ms #",
           {},
@@ -295,24 +281,18 @@ describe("LocalMembership", () => {
   });
 
   it("logs if callIntent cannot be updated", async () => {
-    const scope = new ObservableScope();
-
-    const mockConnectionManager = {
-      transports$: constant(new Epoch([])),
-      connectionManagerData$: constant(new Epoch(new ConnectionManagerData())),
-    };
     async function reject(): Promise<void> {
       return Promise.reject(new Error("Not connected yet"));
     }
     const localMembership = createLocalMembership$({
-      scope,
+      scope: testScope(),
       ...defaultCreateLocalMemberValues,
       matrixRTCSession: {
         slotId: "m.call#ROOM",
         updateCallIntent: vi.fn().mockImplementation(reject),
         leave: vi.fn(),
       },
-      connectionManager: mockConnectionManager,
+      connectionMap$: constant(new Epoch(new ConnectionMap())),
       localTransport$: constant(mockTransport),
     });
     const expextedLog =
@@ -322,7 +302,6 @@ describe("LocalMembership", () => {
     await flushPromises();
     defaultCreateLocalMemberValues.muteStates.video.setEnabled$.value?.(true);
     expect(internalLogger).toHaveBeenCalledWith(expextedLog);
-    scope.end();
   });
 
   const mockTransport = {
@@ -374,14 +353,12 @@ describe("LocalMembership", () => {
       if (delegationUrl !== null)
         fetchMock.post(delegationUrl, () => ({ status: 401, body: {} }));
 
-      const connectionManagerData = new ConnectionManagerData();
-      connectionManagerData.add(connectionTransportAConnected, []);
+      const connectionMap = new ConnectionMap();
+      connectionMap.add(connectionTransportAConnected, []);
       const localMembership = createLocalMembership$({
         scope,
         ...defaultCreateLocalMemberValues,
-        connectionManager: {
-          connectionManagerData$: constant(new Epoch(connectionManagerData)),
-        },
+        connectionMap$: constant(new Epoch(connectionMap)),
         joinMatrixRTC,
         localTransport$: constant(mockTransport),
         delayId$,
@@ -453,15 +430,13 @@ describe("LocalMembership", () => {
         typeof vi.fn
       >;
 
-    const connectionManagerData = new ConnectionManagerData();
-    connectionManagerData.add(connectionTransportAConnected, []);
-    // connectionManagerData.add(connectionTransportB, []);
+    const connectionMap = new ConnectionMap();
+    connectionMap.add(connectionTransportAConnected, []);
+    // connectionMap.add(connectionTransportB, []);
     const localMembership = createLocalMembership$({
       scope,
       ...defaultCreateLocalMemberValues,
-      connectionManager: {
-        connectionManagerData$: constant(new Epoch(connectionManagerData)),
-      },
+      connectionMap$: constant(new Epoch(connectionMap)),
       localTransport$: constant(mockTransport),
     });
     await flushPromises();
@@ -483,12 +458,10 @@ describe("LocalMembership", () => {
   //
   it("tracks livekit state correctly", async () => {
     const scope = new ObservableScope();
-    const connectionManagerData = new ConnectionManagerData();
+    const connectionMap = new ConnectionMap();
     const localTransport$ = new Subject<LivekitTransport>();
 
-    const connectionManagerData$ = new BehaviorSubject(
-      new Epoch(connectionManagerData),
-    );
+    const connectionMap$ = new BehaviorSubject(new Epoch(connectionMap));
     const publishers: Publisher[] = [];
 
     const publishing$ = new BehaviorSubject<boolean>(false);
@@ -522,9 +495,7 @@ describe("LocalMembership", () => {
     const localMembership = createLocalMembership$({
       scope,
       ...defaultCreateLocalMemberValues,
-      connectionManager: {
-        connectionManagerData$,
-      },
+      connectionMap$: connectionMap$,
       localTransport$,
     });
 
@@ -542,14 +513,14 @@ describe("LocalMembership", () => {
       },
     });
 
-    const connectionManagerData2 = new ConnectionManagerData();
-    connectionManagerData2.add(
+    const connectionMap2 = new ConnectionMap();
+    connectionMap2.add(
       // clone because we will mutate this later.
       { ...connectionTransportAConnecting } as unknown as Connection,
       [],
     );
 
-    connectionManagerData$.next(new Epoch(connectionManagerData2));
+    connectionMap$.next(new Epoch(connectionMap2));
     await flushPromises();
     expect(localMembership.localMemberState$.value).toStrictEqual({
       matrix: RTCMemberStatus.Connected,
@@ -563,7 +534,7 @@ describe("LocalMembership", () => {
     });
 
     (
-      connectionManagerData2.getConnectionForTransport(mockTransport)!
+      connectionMap2.getConnectionForTransport(mockTransport)!
         .state$ as BehaviorSubject<ConnectionState>
     ).next({
       state: "authenticated",
@@ -659,8 +630,8 @@ describe("LocalMembership", () => {
         [boolean, HomeserverDisconnectReason | null]
       >([false, "membership"]);
 
-      const connectionManagerData = new ConnectionManagerData();
-      connectionManagerData.add(connectionTransportAConnected, []);
+      const connectionMap = new ConnectionMap();
+      connectionMap.add(connectionTransportAConnected, []);
 
       createLocalMembership$({
         scope,
@@ -669,9 +640,7 @@ describe("LocalMembership", () => {
           combined$: hsReason$,
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
-        connectionManager: {
-          connectionManagerData$: constant(new Epoch(connectionManagerData)),
-        },
+        connectionMap$: constant(new Epoch(connectionMap)),
         localTransport$: constant(mockTransport),
       });
 
@@ -696,8 +665,8 @@ describe("LocalMembership", () => {
         [boolean, HomeserverDisconnectReason | null]
       >([true, null]);
 
-      const connectionManagerData = new ConnectionManagerData();
-      connectionManagerData.add(connectionTransportAConnected, []);
+      const connectionMap = new ConnectionMap();
+      connectionMap.add(connectionTransportAConnected, []);
 
       createLocalMembership$({
         scope,
@@ -706,9 +675,7 @@ describe("LocalMembership", () => {
           combined$: hsReason$,
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
-        connectionManager: {
-          connectionManagerData$: constant(new Epoch(connectionManagerData)),
-        },
+        connectionMap$: constant(new Epoch(connectionMap)),
         localTransport$: constant(mockTransport),
       });
 
@@ -742,8 +709,8 @@ describe("LocalMembership", () => {
         state$: connectionState$,
       } as unknown as Connection;
 
-      const connectionManagerData = new ConnectionManagerData();
-      connectionManagerData.add(mutableConnection, []);
+      const connectionMap = new ConnectionMap();
+      connectionMap.add(mutableConnection, []);
 
       createLocalMembership$({
         scope,
@@ -755,9 +722,7 @@ describe("LocalMembership", () => {
           ]),
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
-        connectionManager: {
-          connectionManagerData$: constant(new Epoch(connectionManagerData)),
-        },
+        connectionMap$: constant(new Epoch(connectionMap)),
         localTransport$: constant(mockTransport),
       });
 
@@ -792,8 +757,8 @@ describe("LocalMembership", () => {
         [boolean, HomeserverDisconnectReason | null]
       >([true, null]);
 
-      const connectionManagerData = new ConnectionManagerData();
-      connectionManagerData.add(connectionTransportAConnected, []);
+      const connectionMap = new ConnectionMap();
+      connectionMap.add(connectionTransportAConnected, []);
 
       createLocalMembership$({
         scope,
@@ -802,9 +767,7 @@ describe("LocalMembership", () => {
           combined$: hsReason$,
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
-        connectionManager: {
-          connectionManagerData$: constant(new Epoch(connectionManagerData)),
-        },
+        connectionMap$: constant(new Epoch(connectionMap)),
         localTransport$: constant(mockTransport),
       });
 
@@ -863,14 +826,12 @@ describe("LocalMembership", () => {
       localMembership: ReturnType<typeof createLocalMembership$>;
     } => {
       const scope = new ObservableScope();
-      const connectionManagerData = new ConnectionManagerData();
-      if (connection) connectionManagerData.add(connection, []);
+      const connectionMap = new ConnectionMap();
+      if (connection) connectionMap.add(connection, []);
       const localMembership = createLocalMembership$({
         scope,
         ...defaultCreateLocalMemberValues,
-        connectionManager: {
-          connectionManagerData$: constant(new Epoch(connectionManagerData)),
-        },
+        connectionMap$: constant(new Epoch(connectionMap)),
         localTransport$: constant(mockTransport),
       });
       return { scope, localMembership };

src/state/CallViewModel/localMember/LocalMember.ts

@@ -47,8 +47,8 @@ import { deepCompare } from "matrix-js-sdk/lib/utils";
 import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
 
 import { type Behavior } from "../../Behavior.ts";
-import { type IConnectionManager } from "../remoteMembers/ConnectionManager.ts";
-import { type ObservableScope } from "../../ObservableScope.ts";
+import { type ConnectionMap } from "../remoteMembers/ConnectionMap.ts";
+import { type Epoch, type ObservableScope } from "../../ObservableScope.ts";
 import { type Publisher } from "./Publisher.ts";
 import { type MuteStates } from "../../MuteStates.ts";
 import {
@@ -138,7 +138,7 @@ interface Props {
   // that the inputs for those createSomething$() functions should NOT contain any js-sdk objectes
   scope: ObservableScope;
   muteStates: MuteStates;
-  connectionManager: IConnectionManager;
+  connectionMap$: Behavior<Epoch<ConnectionMap>>;
   createPublisherFactory: (connection: Connection) => Publisher;
   joinMatrixRTC: (
     transport: LivekitTransport,
@@ -169,7 +169,7 @@ interface Props {
  *  -
  * @param props The properties required to create the local membership.
  * @param props.scope The observable scope to use.
- * @param props.connectionManager The connection manager to get connections from.
+ * @param props.connectionMap$ The connection map to get connections from.
  * @param props.createPublisherFactory Factory to create a publisher once we have a connection.
  * @param props.joinMatrixRTC Callback to join the matrix RTC session once we have a transport.
  * @param props.homeserverConnected The homeserver connected state.
@@ -191,7 +191,7 @@ interface Props {
  */
 export const createLocalMembership$ = ({
   scope,
-  connectionManager,
+  connectionMap$,
   localTransport$: localTransportWithErrors$,
   homeserverConnected,
   createPublisherFactory,
@@ -324,13 +324,11 @@ export const createLocalMembership$ = ({
 
   // Drop Epoch data here since we will not combine this anymore
   const localConnection$ = scope.behavior(
-    combineLatest([
-      connectionManager.connectionManagerData$,
-      localTransport$,
-    ]).pipe(
-      map(([{ value: connectionData }, transport]) =>
-        connectionData.getConnectionForTransport(transport),
-      ),
+    combineLatest(
+      [connectionMap$, localTransport$],
+      ({ value: connectionMap }, transport) =>
+        connectionMap.getConnectionForTransport(transport),
+    ).pipe(
       tap((connection) => {
         logger.info(
           `Local connection updated: ${JSON.stringify(connection?.transport)}`,

src/state/CallViewModel/remoteMembers/Connection.test.ts

@@ -418,8 +418,8 @@ describe("Start connection states", () => {
     pendingConnect.reject(new Error("Client initiated disconnect"));
     await stopping;
 
-    // start() resolves rather than rejecting (it is not awaited by the
-    // ConnectionManager, so a rejection would be unhandled).
+    // start() resolves rather than rejecting (it is not awaited by the caller,
+    // so a rejection would be unhandled).
     await expect(started).resolves.toBeUndefined();
     expect(capturedStates.at(-1)).toEqual({ state: "stopped" });
     expect(capturedStates.some((st) => st instanceof Error)).toBe(false);

src/state/CallViewModel/remoteMembers/Connection.ts

@@ -356,7 +356,7 @@ export class Connection {
         // stop() was called while we were connecting, which makes the pending
         // connect reject. That is the abort we asked for, not a failure, so
         // don't record an error state on a stopped connection or rethrow it
-        // (start() is not awaited by the ConnectionManager, so a throw here
+        // (start() is not necessarily awaited by the caller, so a throw here
         // becomes an unhandled promise rejection).
         this.logger.debug(`Connect aborted because the connection was stopped`);
         return;

src/state/CallViewModel/remoteMembers/ConnectionManager.ts

@@ -1,269 +0,0 @@
-/*
-Copyright 2025 Element Creations Ltd.
-Copyright 2025 New Vector Ltd.
-
-SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
-import {
-  combineLatest,
-  map,
-  type Observable,
-  of,
-  switchMap,
-  startWith,
-  catchError,
-  NEVER,
-} from "rxjs";
-import { type Logger } from "matrix-js-sdk/lib/logger";
-import { type RemoteParticipant } from "livekit-client";
-import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
-
-import { type Behavior } from "../../Behavior.ts";
-import { type Connection } from "./Connection.ts";
-import { Epoch, type ObservableScope } from "../../ObservableScope.ts";
-import { generateItemsWithEpoch } from "../../../utils/observable.ts";
-import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
-import { type ConnectionFactory } from "./ConnectionFactory.ts";
-
-export class ConnectionManagerData {
-  private readonly store: Map<
-    string,
-    { connection: Connection; participants: RemoteParticipant[] }
-  > = new Map();
-
-  public constructor(private readonly logger?: Logger) {}
-
-  public add(connection: Connection, participants: RemoteParticipant[]): void {
-    const key = this.getKey(connection.transport);
-    const existing = this.store.get(key);
-    if (!existing) {
-      this.store.set(key, { connection, participants });
-    } else {
-      // Transports are deduplicated by URL upstream, so this should never
-      // happen; if it does, members may be matched against the wrong room.
-      this.logger?.warn(
-        `Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`,
-      );
-      existing.participants.push(...participants);
-    }
-  }
-
-  private getKey(transport: LivekitTransport): string {
-    // This is enough as a key because the ConnectionManager is already scoped by room.
-    // We also do not need to consider the slotId at this point since each `MatrixRTCSession` is already scoped by `slotDescription: {id, application}`.
-    return transport.livekit_service_url;
-  }
-
-  public getConnections(): Connection[] {
-    return Array.from(this.store.values()).map(({ connection }) => connection);
-  }
-
-  public getConnectionForTransport(
-    transport: LivekitTransport,
-  ): Connection | null {
-    return this.store.get(this.getKey(transport))?.connection ?? null;
-  }
-
-  public getParticipantsForTransport(
-    transport: LivekitTransport,
-  ): RemoteParticipant[] {
-    const key = this.getKey(transport);
-    const existing = this.store.get(key);
-    if (existing) {
-      return existing.participants;
-    }
-    return [];
-  }
-}
-
-interface Props {
-  scope: ObservableScope;
-  connectionFactory: ConnectionFactory;
-  localTransport$: Observable<LivekitTransport>;
-  remoteTransports$: Behavior<Epoch<LivekitTransport[]>>;
-
-  logger: Logger;
-  ownMembershipIdentity: CallMembershipIdentityParts;
-}
-
-// TODO - write test for scopes (do we really need to bind scope)
-export interface IConnectionManager {
-  connectionManagerData$: Behavior<Epoch<ConnectionManagerData>>;
-}
-
-/**
- * Crete a `ConnectionManager`
- * @param props - Configuration object
- * @param props.scope - The observable scope used by this object
- * @param props.connectionFactory - Used to create new connections
- * @param props.localTransport$ - The transport to publish local media on. (deduplicated with remoteTransports$)
- * @param props.remoteTransports$ - All other transports. The connection manager will create connections for each transport. (deduplicated with localTransport$)
- * @param props.ownMembershipIdentity - The own membership identity to use.
- * @param props.logger - The logger to use.
-
- *
- *   Each of these behaviors can be interpreted as subscribed list of transports.
- *
- *   Using `registerTransports` independent external modules can control what connections
- *   are created by the ConnectionManager.
- *
- *   The connection manager will remove all duplicate transports in each subscibed list.
- *
- *   See `unregisterAllTransports` and `unregisterTransport` for details on how to unsubscribe.
- */
-export function createConnectionManager$({
-  scope,
-  connectionFactory,
-  localTransport$,
-  remoteTransports$,
-  logger: parentLogger,
-  ownMembershipIdentity,
-}: Props): IConnectionManager {
-  const logger = parentLogger.getChild("[ConnectionManager]");
-  // TODO logger: only construct one logger from the client and make it compatible via a EC specific sing
-
-  const localTransportAsArray$ = localTransport$.pipe(
-    // LocalMember already surfaces local transport errors properly in the UI,
-    // here we can just swallow them
-    catchError(() => NEVER),
-    map((transport) => [transport]),
-    startWith([]),
-  );
-
-  /**
-   * All transports currently managed by the ConnectionManager.
-   *
-   * This list does not include duplicate transports.
-   *
-   * It is build based on the list of subscribed transports (`transportsSubscriptions$`).
-   * externally this is modified via `registerTransports()`.
-   */
-  const localAndRemoteTransports$: Behavior<
-    Epoch<{ transport: LivekitTransport; role: "publisher" | "subscriber" }[]>
-  > = scope.behavior(
-    combineLatest([localTransportAsArray$, remoteTransports$]).pipe(
-      // Combine local and remote transports into one transport array, tagged
-      // with the role we take on each of them
-      map(([localTransportAsArray, remoteTransports]) => {
-        const dedupedRemote = removeDuplicateTransports(remoteTransports.value);
-        const remoteWithoutLocal = dedupedRemote.filter(
-          (transport) =>
-            !localTransportAsArray.find((l) =>
-              areLivekitTransportsEqual(l, transport),
-            ),
-        );
-        logger.debug(
-          "remoteWithoutLocal",
-          remoteWithoutLocal,
-          "localTransportAsArray",
-          localTransportAsArray,
-        );
-        return new Epoch(
-          [
-            ...localTransportAsArray.map((transport) => ({
-              transport,
-              role: "publisher" as const,
-            })),
-            ...remoteWithoutLocal.map((transport) => ({
-              transport,
-              role: "subscriber" as const,
-            })),
-          ],
-          remoteTransports.epoch,
-        );
-      }),
-    ),
-  );
-
-  /**
-   * Connections for each transport in use by one or more session members.
-   */
-  const connections$ = scope.behavior(
-    localAndRemoteTransports$.pipe(
-      generateItemsWithEpoch(
-        "ConnectionManager connections$",
-        function* (transports) {
-          for (const { transport, role } of transports) {
-            yield {
-              keys: [role, transport.livekit_service_url],
-              data: undefined,
-            };
-          }
-        },
-        (scope, _data$, role, serviceUrl) => {
-          const connection = connectionFactory.createConnection(
-            scope,
-            role,
-            {
-              type: "livekit",
-              livekit_service_url: serviceUrl,
-            },
-            ownMembershipIdentity,
-            logger,
-          );
-          // Start the connection immediately
-          // Use connection state to track connection progress
-          void connection.start();
-          // TODO subscribe to connection state to retry or log issues?
-          return connection;
-        },
-      ),
-    ),
-  );
-
-  const connectionManagerData$ = scope.behavior(
-    connections$.pipe(
-      switchMap((connections) => {
-        const epoch = connections.epoch;
-
-        // Map the connections to list of {connection, participants}[]
-        const listOfConnectionsWithRemoteParticipants = connections.value.map(
-          (connection) => {
-            return connection.remoteParticipants$.pipe(
-              map((participants) => ({
-                connection,
-                participants,
-              })),
-            );
-          },
-        );
-
-        // probably not required
-
-        if (listOfConnectionsWithRemoteParticipants.length === 0) {
-          return of(new Epoch(new ConnectionManagerData(), epoch));
-        }
-
-        // combineLatest the several streams into a single stream with the ConnectionManagerData
-        return combineLatest(listOfConnectionsWithRemoteParticipants).pipe(
-          map(
-            (lists) =>
-              new Epoch(
-                lists.reduce((data, { connection, participants }) => {
-                  data.add(connection, participants);
-                  return data;
-                }, new ConnectionManagerData(logger)),
-                epoch,
-              ),
-          ),
-        );
-      }),
-    ),
-    new Epoch(new ConnectionManagerData(), -1),
-  );
-
-  return { connectionManagerData$ };
-}
-
-function removeDuplicateTransports<T extends LivekitTransport>(
-  transports: T[],
-): T[] {
-  return transports.reduce((acc, transport) => {
-    if (!acc.some((t) => areLivekitTransportsEqual(t, transport)))
-      acc.push(transport);
-    return acc;
-  }, [] as T[]);
-}

src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts

@@ -6,16 +6,13 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
-import { BehaviorSubject, NEVER } from "rxjs";
+import { BehaviorSubject } from "rxjs";
 import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
 import { type RemoteParticipant } from "livekit-client";
 import { logger } from "matrix-js-sdk/lib/logger";
 
 import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts";
-import {
-  createConnectionManager$,
-  ConnectionManagerData,
-} from "./ConnectionManager.ts";
+import { createConnectionMap$, ConnectionMap } from "./ConnectionMap.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
 import { type Connection } from "./Connection.ts";
 import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
@@ -34,6 +31,15 @@ const TRANSPORT_2: LivekitTransport = {
   livekit_service_url: "https://lk.sample.com",
 };
 
+function findConnection(
+  transport: LivekitTransport,
+  connections: Connection[],
+): Connection | undefined {
+  return connections.find((c) =>
+    areLivekitTransportsEqual(c.transport, transport),
+  );
+}
+
 let fakeConnectionFactory: ConnectionFactory;
 let testScope: ObservableScope;
 
@@ -71,10 +77,10 @@ afterEach(() => {
 describe("connections$ stream", () => {
   test("Should create and start new connections for each transports", () => {
     withTestScheduler(({ behavior, expectObservable }) => {
-      const { connectionManagerData$ } = createConnectionManager$({
+      const connectionMap$ = createConnectionMap$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),
@@ -83,7 +89,7 @@ describe("connections$ stream", () => {
       });
 
       expectObservable(
-        connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())),
+        connectionMap$.pipe(mapEpoch((d) => d.getConnections())),
       ).toBe("a", {
         a: expect.toSatisfy((e: Epoch<Connection[]>) => {
           const connections = e.value;
@@ -93,15 +99,11 @@ describe("connections$ stream", () => {
             vi.mocked(fakeConnectionFactory).createConnection,
           ).toHaveBeenCalledTimes(2);
 
-          const conn1 = connections.find((c) =>
-            areLivekitTransportsEqual(c.transport, TRANSPORT_1),
-          );
+          const conn1 = findConnection(TRANSPORT_1, connections);
           expect(conn1).toBeDefined();
           expect(conn1!.start).toHaveBeenCalled();
 
-          const conn2 = connections.find((c) =>
-            areLivekitTransportsEqual(c.transport, TRANSPORT_2),
-          );
+          const conn2 = findConnection(TRANSPORT_2, connections);
           expect(conn2).toBeDefined();
           expect(conn2!.start).toHaveBeenCalled();
           return true;
@@ -112,10 +114,10 @@ describe("connections$ stream", () => {
 
   test("Should start connection only once", () => {
     withTestScheduler(({ behavior, expectObservable }) => {
-      const { connectionManagerData$ } = createConnectionManager$({
+      const connectionMap$ = createConnectionMap$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("abcdef", {
           a: new Epoch([TRANSPORT_1], 0),
           b: new Epoch([TRANSPORT_1], 1),
@@ -129,7 +131,7 @@ describe("connections$ stream", () => {
       });
 
       expectObservable(
-        connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())),
+        connectionMap$.pipe(mapEpoch((d) => d.getConnections())),
       ).toBe("xxxxxa", {
         x: expect.anything(),
         a: expect.toSatisfy((e: Epoch<Connection[]>) => {
@@ -140,14 +142,10 @@ describe("connections$ stream", () => {
             vi.mocked(fakeConnectionFactory).createConnection,
           ).toHaveBeenCalledTimes(2);
 
-          const conn2 = connections.find((c) =>
-            areLivekitTransportsEqual(c.transport, TRANSPORT_2),
-          );
+          const conn2 = findConnection(TRANSPORT_2, connections);
           expect(conn2).toBeDefined();
 
-          const conn1 = connections.find((c) =>
-            areLivekitTransportsEqual(c.transport, TRANSPORT_1),
-          );
+          const conn1 = findConnection(TRANSPORT_1, connections);
           expect(conn1).toBeDefined();
           expect(conn1!.start).toHaveBeenCalledOnce();
 
@@ -159,10 +157,10 @@ describe("connections$ stream", () => {
 
   test("Should cleanup connections when not needed anymore", () => {
     withTestScheduler(({ behavior, expectObservable }) => {
-      const { connectionManagerData$ } = createConnectionManager$({
+      const connectionMap$ = createConnectionMap$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("abc", {
           a: new Epoch([TRANSPORT_1], 0),
           b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
@@ -173,7 +171,7 @@ describe("connections$ stream", () => {
       });
 
       expectObservable(
-        connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())),
+        connectionMap$.pipe(mapEpoch((d) => d.getConnections())),
       ).toBe("xab", {
         x: expect.anything(),
         a: expect.toSatisfy((e: Epoch<Connection[]>) => {
@@ -186,8 +184,9 @@ describe("connections$ stream", () => {
 
           expect(connections.length).toBe(1);
           // The second connection should have been stopped has it is no longer needed.
-          const connection2 = allCreatedConnections.find((c) =>
-            areLivekitTransportsEqual(c.transport, TRANSPORT_2),
+          const connection2 = findConnection(
+            TRANSPORT_2,
+            allCreatedConnections,
           );
           expect(connection2).toBeDefined();
           expect(connection2!.stop).toHaveBeenCalled();
@@ -203,10 +202,10 @@ describe("connections$ stream", () => {
   });
 });
 
-describe("ConnectionManagerData", () => {
+describe("ConnectionMap", () => {
   test("warns when a second connection to the same URL is merged", () => {
     const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
-    const data = new ConnectionManagerData(logger);
+    const data = new ConnectionMap(logger);
     const connection = { transport: TRANSPORT_1 } as unknown as Connection;
     const p = (identity: string): RemoteParticipant =>
       ({ identity }) as unknown as RemoteParticipant;
@@ -221,7 +220,7 @@ describe("ConnectionManagerData", () => {
   });
 });
 
-describe("connectionManagerData$ stream", () => {
+describe("connectionMap$", () => {
   // Used in test to control fake connections' remoteParticipants$ streams
   let fakeRemoteParticipantsStreams: Map<string, Behavior<RemoteParticipant[]>>;
 
@@ -298,10 +297,10 @@ describe("connectionManagerData$ stream", () => {
       );
       // ==============================
 
-      const { connectionManagerData$ } = createConnectionManager$({
+      const connectionMap$ = createConnectionMap$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),
@@ -309,16 +308,16 @@ describe("connectionManagerData$ stream", () => {
         ownMembershipIdentity: ownMemberMock,
       });
 
-      expectObservable(connectionManagerData$).toBe("abcd", {
+      expectObservable(connectionMap$).toBe("abcd", {
         a: expect.toSatisfy((e) => {
-          const data: ConnectionManagerData = e.value;
+          const data: ConnectionMap = e.value;
           expect(data.getConnections().length).toBe(2);
           expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(0);
           expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0);
           return true;
         }),
         b: expect.toSatisfy((e) => {
-          const data: ConnectionManagerData = e.value;
+          const data: ConnectionMap = e.value;
           expect(data.getConnections().length).toBe(2);
           expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1);
           expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0);
@@ -328,7 +327,7 @@ describe("connectionManagerData$ stream", () => {
           return true;
         }),
         c: expect.toSatisfy((e) => {
-          const data: ConnectionManagerData = e.value;
+          const data: ConnectionMap = e.value;
           expect(data.getConnections().length).toBe(2);
           expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1);
           expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1);
@@ -341,7 +340,7 @@ describe("connectionManagerData$ stream", () => {
           return true;
         }),
         d: expect.toSatisfy((e) => {
-          const data: ConnectionManagerData = e.value;
+          const data: ConnectionMap = e.value;
           expect(data.getConnections().length).toBe(2);
           expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(2);
           expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1);

src/state/CallViewModel/remoteMembers/ConnectionMap.ts

@@ -0,0 +1,193 @@
+/*
+Copyright 2025-2026 Element Creations Ltd.
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
+import { combineLatest, map, of, switchMap, tap } from "rxjs";
+import { type Logger } from "matrix-js-sdk/lib/logger";
+import { type RemoteParticipant } from "livekit-client";
+import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
+
+import { type Behavior } from "../../Behavior.ts";
+import { type Connection } from "./Connection.ts";
+import {
+  Epoch,
+  mapEpoch,
+  type ObservableScope,
+} from "../../ObservableScope.ts";
+import { generateItemsWithEpoch } from "../../../utils/observable.ts";
+import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
+import { type ConnectionFactory } from "./ConnectionFactory.ts";
+
+function removeDuplicateTransports<T extends LivekitTransport>(
+  transports: T[],
+): T[] {
+  return transports.reduce((acc, transport) => {
+    if (!acc.some((t) => areLivekitTransportsEqual(t, transport)))
+      acc.push(transport);
+    return acc;
+  }, [] as T[]);
+}
+
+export class ConnectionMap {
+  private readonly map = new Map<
+    string,
+    { connection: Connection; participants: RemoteParticipant[] }
+  >();
+
+  public constructor(private readonly logger?: Logger) {}
+
+  public add(connection: Connection, participants: RemoteParticipant[]): void {
+    const key = this.getKey(connection.transport);
+    const existing = this.map.get(key);
+    if (!existing) {
+      this.map.set(key, { connection, participants });
+    } else {
+      // Transports are deduplicated by URL upstream, so this should never
+      // happen; if it does, members may be matched against the wrong room.
+      this.logger?.warn(
+        `Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`,
+      );
+      existing.participants.push(...participants);
+    }
+  }
+
+  private getKey(transport: LivekitTransport): string {
+    // This is enough as a key because the ConnectionMap is already scoped by room.
+    // We also do not need to consider the slotId at this point since each `MatrixRTCSession` is already scoped by `slotDescription: {id, application}`.
+    return transport.livekit_service_url;
+  }
+
+  public getConnections(): Connection[] {
+    return Array.from(this.map.values()).map(({ connection }) => connection);
+  }
+
+  public getConnectionForTransport(
+    transport: LivekitTransport,
+  ): Connection | null {
+    return this.map.get(this.getKey(transport))?.connection ?? null;
+  }
+
+  public getParticipantsForTransport(
+    transport: LivekitTransport,
+  ): RemoteParticipant[] {
+    const existing = this.map.get(this.getKey(transport));
+    if (existing) {
+      return existing.participants;
+    }
+    return [];
+  }
+}
+
+interface Props {
+  scope: ObservableScope;
+  connectionFactory: ConnectionFactory;
+  localTransport: LivekitTransport;
+  remoteTransports$: Behavior<Epoch<LivekitTransport[]>>;
+
+  logger: Logger;
+  ownMembershipIdentity: CallMembershipIdentityParts;
+}
+
+/**
+ * Create a dynamic map of connections and their associated backend participants.
+ * @param props - Configuration object
+ * @param props.scope - The observable scope used by this object
+ * @param props.connectionFactory - Used to create new connections
+ * @param props.localTransport - The transport to publish local media on. (deduplicated with remoteTransports$)
+ * @param props.remoteTransports$ - All other transports. The connection manager will create connections for each transport. (deduplicated with localTransport)
+ * @param props.ownMembershipIdentity - The own membership identity to use.
+ * @param props.logger - The logger to use.
+ */
+export function createConnectionMap$({
+  scope,
+  connectionFactory,
+  localTransport,
+  remoteTransports$,
+  logger: parentLogger,
+  ownMembershipIdentity,
+}: Props): Behavior<Epoch<ConnectionMap>> {
+  const logger = parentLogger.getChild("[ConnectionMap]");
+  // TODO logger: only construct one logger from the client and make it compatible via a EC specific sing
+
+  /**
+   * All transports currently represented in the connection map. This list does
+   * not include duplicate transports.
+   */
+  const localAndRemoteTransports$: Behavior<Epoch<LivekitTransport[]>> =
+    scope.behavior(
+      remoteTransports$.pipe(
+        mapEpoch((remoteTransports) =>
+          removeDuplicateTransports([localTransport, ...remoteTransports]),
+        ),
+        tap((transports) =>
+          logger.debug("localAndRemoteTransports$ = ", transports),
+        ),
+      ),
+    );
+
+  /**
+   * Connections for each transport in use by one or more session members.
+   */
+  const connections$ = scope.behavior(
+    localAndRemoteTransports$.pipe(
+      generateItemsWithEpoch(
+        "ConnectionMap connections$",
+        function* (transports) {
+          for (const transport of transports) {
+            const role = areLivekitTransportsEqual(transport, localTransport)
+              ? ("publisher" as const)
+              : ("subscriber" as const);
+            yield {
+              keys: [role, transport.livekit_service_url],
+              data: undefined,
+            };
+          }
+        },
+        (scope, _data$, role, serviceUrl) => {
+          const connection = connectionFactory.createConnection(
+            scope,
+            role,
+            { type: "livekit", livekit_service_url: serviceUrl },
+            ownMembershipIdentity,
+            logger,
+          );
+          // Start the connection immediately
+          // Use connection state to track connection progress
+          void connection.start();
+          // TODO subscribe to connection state to retry or log issues?
+          return connection;
+        },
+      ),
+    ),
+  );
+
+  return scope.behavior(
+    connections$.pipe(
+      switchMap(({ value: connections, epoch }) => {
+        if (connections.length === 0)
+          return of(new Epoch(new ConnectionMap(), epoch));
+
+        return combineLatest(
+          // Map the connections to list of Observable<{connection, participants}>
+          connections.map((connection) =>
+            connection.remoteParticipants$.pipe(
+              map((participants) => ({ connection, participants })),
+            ),
+          ),
+          // Collect into a single ConnectionMap
+          (...list) => {
+            const map = new ConnectionMap(logger);
+            for (const item of list)
+              map.add(item.connection, item.participants);
+            return new Epoch(map, epoch);
+          },
+        );
+      }),
+    ),
+  );
+}

src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts

@@ -13,7 +13,6 @@ import {
 } from "matrix-js-sdk/lib/matrixrtc";
 import { BehaviorSubject, combineLatest, map, type Observable } from "rxjs";
 
-import { type IConnectionManager } from "./ConnectionManager.ts";
 import {
   type RemoteMatrixLivekitMember,
   createRemoteMatrixLivekitMembers$,
@@ -24,7 +23,7 @@ import {
   ObservableScope,
   trackEpoch,
 } from "../../ObservableScope.ts";
-import { ConnectionManagerData } from "./ConnectionManager.ts";
+import { ConnectionMap } from "./ConnectionMap.ts";
 import {
   flushPromises,
   mockRtcMembership,
@@ -78,23 +77,18 @@ function epochMeWith$<T, U>(
 test("should signal participant not yet connected to livekit", async () => {
   const info = vi.spyOn(logger, "info");
   const mockedMemberships$ = new BehaviorSubject([bobMembership]);
-  const mockConnectionManagerData$ = new BehaviorSubject(
-    new ConnectionManagerData(),
-  );
+  const mockConnectionMap$ = new BehaviorSubject(new ConnectionMap());
   const { memberships$, membershipsWithTransport$ } =
     createEpochedMemberships$(mockedMemberships$);
 
-  const connectionManagerData$ = epochMeWith$(
-    memberships$,
-    mockConnectionManagerData$,
+  const connectionMap$ = testScope.behavior(
+    epochMeWith$(memberships$, mockConnectionMap$),
   );
 
   const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({
     scope: testScope,
     membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
-    connectionManager: {
-      connectionManagerData$: connectionManagerData$,
-    } as unknown as IConnectionManager,
+    connectionMap$,
     localUser: localRtcMember,
   });
 
@@ -152,22 +146,19 @@ test("should signal participant on a connection that is publishing", async () =>
   const connection = {
     transport: bobMembership.getTransport(),
   } as unknown as Connection;
-  const dataWithPublisher = new ConnectionManagerData();
+  const dataWithPublisher = new ConnectionMap();
   dataWithPublisher.add(connection, [
     mockRemoteParticipant({ identity: bobParticipantId }),
   ]);
 
-  const connectionManagerData$ = epochMeWith$(
-    memberships$,
-    constant(dataWithPublisher),
+  const connectionMap$ = testScope.behavior(
+    epochMeWith$(memberships$, constant(dataWithPublisher)),
   );
 
   const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({
     scope: testScope,
     membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
-    connectionManager: {
-      connectionManagerData$: connectionManagerData$,
-    } as unknown as IConnectionManager,
+    connectionMap$,
     localUser: localRtcMember,
   });
 
@@ -195,20 +186,17 @@ test("should signal participant on a connection that is not publishing", async (
   const connection = {
     transport: bobMembership.getTransport(),
   } as unknown as Connection;
-  const dataWithPublisher = new ConnectionManagerData();
+  const dataWithPublisher = new ConnectionMap();
   dataWithPublisher.add(connection, []);
 
-  const connectionManagerData$ = epochMeWith$(
-    memberships$,
-    constant(dataWithPublisher),
+  const connectionMap$ = testScope.behavior(
+    epochMeWith$(memberships$, constant(dataWithPublisher)),
   );
 
   const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({
     scope: testScope,
     membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
-    connectionManager: {
-      connectionManagerData$: connectionManagerData$,
-    } as unknown as IConnectionManager,
+    connectionMap$,
     localUser: localRtcMember,
   });
   await flushPromises();
@@ -228,7 +216,7 @@ describe("Publication edge case", () => {
     const { memberships$, membershipsWithTransport$ } =
       createEpochedMemberships$(constant([bobMembership, carlMembership]));
 
-    const connectionWithPublisher = new ConnectionManagerData();
+    const connectionWithPublisher = new ConnectionMap();
     const bobParticipantId = fallbackMemberId(
       bobMembership.userId,
       bobMembership.deviceId,
@@ -247,17 +235,14 @@ describe("Publication edge case", () => {
       mockRemoteParticipant({ identity: bobParticipantId }),
     ]);
 
-    const connectionManagerData$ = epochMeWith$(
-      memberships$,
-      constant(connectionWithPublisher),
+    const connectionMap$ = testScope.behavior(
+      epochMeWith$(memberships$, constant(connectionWithPublisher)),
     );
 
     const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({
       scope: testScope,
       membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
-      connectionManager: {
-        connectionManagerData$: connectionManagerData$,
-      } as unknown as IConnectionManager,
+      connectionMap$,
       localUser: localRtcMember,
     });
     await flushPromises();
@@ -287,7 +272,7 @@ test("bob is publishing in the wrong connection", async () => {
   const { memberships$, membershipsWithTransport$ } =
     createEpochedMemberships$(mockedMemberships$);
 
-  const connectionWithPublisher = new ConnectionManagerData();
+  const connectionWithPublisher = new ConnectionMap();
 
   const bobParticipantId = fallbackMemberId(
     bobMembership.userId,
@@ -306,17 +291,14 @@ test("bob is publishing in the wrong connection", async () => {
   const connectionsWithPublisher$ = new BehaviorSubject(
     connectionWithPublisher,
   );
-  const connectionManagerData$ = epochMeWith$(
-    memberships$,
-    connectionsWithPublisher$,
+  const connectionMap$ = testScope.behavior(
+    epochMeWith$(memberships$, connectionsWithPublisher$),
   );
 
   const remoteMatrixLivekitMembers$ = createRemoteMatrixLivekitMembers$({
     scope: testScope,
     membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
-    connectionManager: {
-      connectionManagerData$: connectionManagerData$,
-    } as unknown as IConnectionManager,
+    connectionMap$,
     localUser: localRtcMember,
   });
 

src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts

@@ -14,7 +14,7 @@ import { combineLatest, filter, map } from "rxjs";
 import { logger } from "matrix-js-sdk/lib/logger";
 
 import { type Behavior } from "../../Behavior";
-import { type IConnectionManager } from "./ConnectionManager";
+import { type ConnectionMap } from "./ConnectionMap";
 import { Epoch, type ObservableScope } from "../../ObservableScope";
 import { type Connection } from "./Connection";
 import { generateItemsWithEpoch } from "../../../utils/observable";
@@ -62,7 +62,7 @@ interface Props {
   membershipsWithTransport$: Behavior<
     Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
   >;
-  connectionManager: IConnectionManager;
+  connectionMap$: Behavior<Epoch<ConnectionMap>>;
   localUser: { deviceId: string; userId: string };
 }
 
@@ -72,24 +72,21 @@ interface Props {
  * It has a small public interface:
  *  - in (via constructor):
  *    - an observable of CallMembership[] to track the call members (The matrix side)
- *    - a `ConnectionManager` for the lk rooms (The livekit side)
+ *    - a `ConnectionMap` for the lk rooms (The livekit side)
  *  - out (via public Observable):
  *    - `remoteMatrixLivekitMember` an observable of MatrixLivekitMember[] to track the remote members and associated livekit data.
  */
 export function createRemoteMatrixLivekitMembers$({
   scope,
   membershipsWithTransport$,
-  connectionManager,
+  connectionMap$,
   localUser,
 }: Props): Behavior<Epoch<RemoteMatrixLivekitMember[]>> {
   /**
    * Behavior of all the remote call members and their associated livekit data (if available).
    */
   return scope.behavior(
-    combineLatest([
-      membershipsWithTransport$,
-      connectionManager.connectionManagerData$,
-    ]).pipe(
+    combineLatest([membershipsWithTransport$, connectionMap$]).pipe(
       filter((values) =>
         values.every((value) => value.epoch === values[0].epoch),
       ),

src/state/CallViewModel/remoteMembers/integration.test.ts

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { test, vi, expect, beforeEach, afterEach } from "vitest";
-import { BehaviorSubject, NEVER } from "rxjs";
+import { BehaviorSubject } from "rxjs";
 import { type Room as LivekitRoom } from "livekit-client";
 import EventEmitter from "events";
 import fetchMock from "fetch-mock";
@@ -21,6 +21,7 @@ import {
 import { ECConnectionFactory } from "./ConnectionFactory.ts";
 import { type ClientOpenIDParts } from "../../../livekit/auth";
 import {
+  exampleTransport,
   mockMediaDevices,
   mockRtcMembership,
   ownMemberMock,
@@ -32,11 +33,11 @@ import {
   createRemoteMatrixLivekitMembers$,
   type RemoteMatrixLivekitMember,
 } from "./MatrixLivekitMembers.ts";
-import { createConnectionManager$ } from "./ConnectionManager.ts";
+import { createConnectionMap$ } from "./ConnectionMap.ts";
 import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
 import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
 
-// Test the integration of ConnectionManager and MatrixLivekitMerger
+// Test the integration of ConnectionMap and MatrixLivekitMembers
 
 let testScope: ObservableScope;
 let ecConnectionFactory: ECConnectionFactory;
@@ -122,10 +123,10 @@ test("bob, carl, then bob joining no tracks yet", () => {
       memberships$,
     );
 
-    const connectionManager = createConnectionManager$({
+    const connectionMap$ = createConnectionMap$({
       scope: testScope,
       connectionFactory: ecConnectionFactory,
-      localTransport$: NEVER,
+      localTransport: exampleTransport,
       remoteTransports$: membershipsAndTransports.transports$,
       logger: logger,
       ownMembershipIdentity: ownMemberMock,
@@ -135,7 +136,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
       scope: testScope,
       membershipsWithTransport$:
         membershipsAndTransports.membershipsWithTransport$,
-      connectionManager,
+      connectionMap$,
       localUser: localRtcMember,
     });
 

src/utils/observable.ts

@@ -218,6 +218,28 @@ export function filterBehavior<T, S extends T>(
     );
 }
 
+/**
+ * Applies a given projection to each value emitted by the source Observable.
+ * The `project` function additionally receives an {@link ObservableScope}.
+ */
+export function mapScoped<A, B>(
+  name: string,
+  project: (a: A, scope: ObservableScope) => B,
+): OperatorFunction<A, B> {
+  return (a$) =>
+    a$.pipe(
+      generateItemsInternal(
+        name,
+        function* (a) {
+          yield { keys: [a], data: undefined };
+        },
+        (scope, _data$, a) => project(a, scope),
+        (items) => items,
+      ),
+      map(([item]) => item),
+    );
+}
+
 function generateItemsInternal<
   Input,
   Keys extends [unknown, ...unknown[]],