diff --git a/sdk/src/media-backend/livekit/LivekitBackend.ts b/sdk/src/media-backend/livekit/LivekitBackend.ts index e17f50730..e93a28215 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,21 +236,26 @@ 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) => { + 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}`, + ); logger.info( - `${member.rtcBackendIdentity}: LiveKit participant ${source ? `matched (${source.participant.sid})` : "missing"} on ${locator$.value ? JSON.stringify(locator$.value.transport) : "no transport"}`, + `${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