diff --git a/playwright/url-transport.spec.ts b/playwright/url-transport.spec.ts
index e71c40d19..4226d72df 100644
--- a/playwright/url-transport.spec.ts
+++ b/playwright/url-transport.spec.ts
@@ -165,6 +165,7 @@ test("Group call with all possible transport shapes at once", async ({
page,
browserName,
}) => {
+ test.slow();
skipOnFirefox(browserName);
// The host starts the call with a url transport.
diff --git a/sdk/src/media-backend/livekit/LivekitBackend.ts b/sdk/src/media-backend/livekit/LivekitBackend.ts
index e17f50730..a00e6af64 100644
--- a/sdk/src/media-backend/livekit/LivekitBackend.ts
+++ b/sdk/src/media-backend/livekit/LivekitBackend.ts
@@ -60,7 +60,10 @@ import {
import { toMediaConnectionState } from "./connectionState";
import { createDataChannel$ } from "./DataChannel";
import { createKeyProvider } from "./KeyProvider";
-import { createLivekitMemberMedia } from "./LivekitMemberMedia";
+import {
+ createLivekitMemberMedia,
+ participantSourcesEqual,
+} from "./LivekitMemberMedia";
import { createLivekitLocalMedia } from "./LocalMedia";
export interface LivekitBackendOptions {
@@ -233,22 +236,28 @@ export function createLivekitBackend(
const connection = connections.getConnectionForTransport(locator);
// A member advertising both a `url` and a `livekit_service_url` may
// have authenticated through either flow, under either identity
- const participant = connections
+ const participants = connections
.getParticipantsForTransport(locator)
- .find((p) => backendIdentities.includes(p.identity));
- return connection && participant
- ? { participant, room: connection.livekitRoom }
+ .filter((p) => backendIdentities.includes(p.identity));
+ return connection && participants.length
+ ? { participants, room: connection.livekitRoom }
: null;
}),
- distinctUntilChanged(
- (a, b) => a?.participant === b?.participant && a?.room === b?.room,
- ),
+ distinctUntilChanged(participantSourcesEqual),
),
);
source$.pipe(memberScope.bind()).subscribe((source) => {
- logger.info(
- `${member.rtcBackendIdentity}: LiveKit participant ${source ? `matched (${source.participant.sid})` : "missing"} on ${locator$.value ? JSON.stringify(locator$.value.transport) : "no transport"}`,
- );
+ const transportString = locator$.value
+ ? JSON.stringify(locator$.value.transport)
+ : "no transport";
+ if (source)
+ logger.info(
+ `${member.rtcBackendIdentity}: ${source.participants.length} LiveKit participant(s) matched (${source.participants.map((p) => p.sid).join(", ")}) on ${transportString}`,
+ );
+ else
+ logger.info(
+ `${member.rtcBackendIdentity}: LiveKit participant missing on ${transportString}`,
+ );
});
return createLivekitMemberMedia(memberScope, source$, encryptionSystem);
};
diff --git a/sdk/src/media-backend/livekit/LivekitMemberMedia.test.ts b/sdk/src/media-backend/livekit/LivekitMemberMedia.test.ts
index 47079e5df..04798d0c6 100644
--- a/sdk/src/media-backend/livekit/LivekitMemberMedia.test.ts
+++ b/sdk/src/media-backend/livekit/LivekitMemberMedia.test.ts
@@ -25,7 +25,7 @@ describe("createLivekitMemberMedia", () => {
const { participant, publications, room, emitter } = fakes();
const media = createLivekitMemberMedia(
testScope(),
- constant({ participant, room }),
+ constant({ participants: [participant], room }),
{ kind: E2eeType.NONE },
);
expect(media.tracks$.value).toEqual([]);
@@ -52,7 +52,7 @@ describe("createLivekitMemberMedia", () => {
const { participant, publications, room, emitter } = fakes();
const media = createLivekitMemberMedia(
testScope(),
- constant({ participant, room }),
+ constant({ participants: [participant], room }),
{ kind: E2eeType.NONE },
);
emitter.emit(
diff --git a/sdk/src/media-backend/livekit/LivekitMemberMedia.ts b/sdk/src/media-backend/livekit/LivekitMemberMedia.ts
index f4e19eb72..eca01946c 100644
--- a/sdk/src/media-backend/livekit/LivekitMemberMedia.ts
+++ b/sdk/src/media-backend/livekit/LivekitMemberMedia.ts
@@ -16,13 +16,20 @@ import {
RoomEvent,
type TrackPublication,
} from "livekit-client";
-import { EMPTY, filter, map, type Observable, of, switchMap } from "rxjs";
+import {
+ EMPTY,
+ filter,
+ map,
+ type Observable,
+ of,
+ switchMap,
+ combineLatest,
+} from "rxjs";
import { type ObservableScope } from "../../reactive/ObservableScope";
import { generateItems } from "../../reactive/observable";
import { E2eeType, type EncryptionSystem } from "../../encryption";
import { type Behavior } from "../../reactive/Behavior";
-import { mapScoped } from "../../utils/mapScoped";
import {
type EncryptionError,
type LocalMemberMedia,
@@ -33,13 +40,27 @@ import {
createLivekitMediaTrack,
createLocalLivekitMediaTrack,
} from "./LivekitMediaTrack";
+import { shallowEquals as arrayShallowEquals } from "../../utils/array";
/** Where a member's media comes from: its participant on a room, null while it has none. */
export type ParticipantSource = {
- participant: P;
+ participants: P[];
room: LivekitRoom;
} | null;
+export function participantSourcesEqual
(
+ a: ParticipantSource
,
+ b: ParticipantSource
,
+): boolean {
+ return (
+ a === b ||
+ (a !== null &&
+ b !== null &&
+ arrayShallowEquals(a.participants, b.participants) &&
+ a.room === b.room)
+ );
+}
+
/**
* A member's media from its participant. The only place, with
* `LivekitMediaTrack`, that reads a LiveKit participant; everything above
@@ -99,21 +120,28 @@ function memberTracks$
(
) => T,
): Behavior {
return scope.behavior(
- mapScoped(scope, source$, (participantScope, { participant, room }) =>
- participantScope.behavior(
- observeParticipantMedia(participant).pipe(
- generateItems(
- `${participant.identity} tracks$`,
- function* () {
- for (const publication of participant.trackPublications.values())
- yield { keys: [publication.trackSid], data: publication };
- },
- (trackScope, publication$) =>
- factory(trackScope, participant, publication$.value, room),
+ source$.pipe(
+ switchMap((source) => {
+ if (source === null) return of(null);
+ const { participants, room } = source;
+ return combineLatest(
+ participants.map((participant) =>
+ observeParticipantMedia(participant).pipe(
+ generateItems(
+ `${participant.identity} tracks$`,
+ function* () {
+ for (const publication of participant.trackPublications.values())
+ yield { keys: [publication.trackSid], data: publication };
+ },
+ (trackScope, publication$) =>
+ factory(trackScope, participant, publication$.value, room),
+ ),
+ ),
),
- ),
- ),
- ).pipe(switchMap((tracks$) => tracks$ ?? of(null))),
+ (...tracks) => tracks.flat(1),
+ );
+ }),
+ ),
);
}
@@ -133,7 +161,7 @@ function encryptionErrors$(
filter(
(message) =>
encryptionSystem.kind === E2eeType.SHARED_KEY ||
- message.includes(source.participant.identity),
+ source.participants.some((p) => message.includes(p.identity)),
),
map((message): EncryptionError | undefined =>
message.includes("MissingKey")
diff --git a/sdk/src/media-backend/livekit/LocalMedia.ts b/sdk/src/media-backend/livekit/LocalMedia.ts
index bb5a85249..242a5534a 100644
--- a/sdk/src/media-backend/livekit/LocalMedia.ts
+++ b/sdk/src/media-backend/livekit/LocalMedia.ts
@@ -10,7 +10,6 @@ import {
type LocalParticipant,
type LocalTrackPublication,
ParticipantEvent,
- type Room as LivekitRoom,
} from "livekit-client";
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
@@ -43,7 +42,11 @@ import { type LocalMediaBackend, MediaConnectionState } from "../api";
import { type TransportLocator } from "./auth";
import { type Connection } from "./Connection";
import { type ConnectionMap } from "./ConnectionMap";
-import { createLocalLivekitMemberMedia } from "./LivekitMemberMedia";
+import {
+ createLocalLivekitMemberMedia,
+ type ParticipantSource,
+ participantSourcesEqual,
+} from "./LivekitMemberMedia";
import { livekitSources } from "./LivekitMediaTrack";
import { type DesiredMedia, Publisher } from "./Publisher";
import { toMediaConnectionState } from "./connectionState";
@@ -217,17 +220,15 @@ function participantAndRoom$(
scope: ObservableScope,
participant$: Behavior,
connection$: Behavior,
-): Behavior<{ participant: LocalParticipant; room: LivekitRoom } | null> {
+): Behavior> {
return scope.behavior(
combineLatest([participant$, connection$]).pipe(
map(([participant, connection]) =>
participant && connection
- ? { participant, room: connection.livekitRoom }
+ ? { participants: [participant], room: connection.livekitRoom }
: null,
),
- distinctUntilChanged(
- (a, b) => a?.participant === b?.participant && a?.room === b?.room,
- ),
+ distinctUntilChanged(participantSourcesEqual),
),
);
}
diff --git a/sdk/src/utils/array.ts b/sdk/src/utils/array.ts
new file mode 100644
index 000000000..1a3fc2e5e
--- /dev/null
+++ b/sdk/src/utils/array.ts
@@ -0,0 +1,16 @@
+/*
+Copyright 2024 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+/**
+ * Determine whether two arrays are equal by shallow comparison.
+ */
+export function shallowEquals(first: A[], second: A[]): boolean {
+ if (first.length !== second.length) return false;
+ for (let i = 0; i < first.length; i++)
+ if (first[i] !== second[i]) return false;
+ return true;
+}