Merge branch 'main' into valere/component_ec_M1
This commit is contained in:
@@ -336,6 +336,11 @@ export interface CallViewModel {
|
||||
* Whether we are sharing our screen.
|
||||
*/
|
||||
sharingScreen$: Behavior<boolean>;
|
||||
/**
|
||||
* The last error from toggling screen sharing, until dismissed.
|
||||
*/
|
||||
screenShareError$: Behavior<Error | null>;
|
||||
dismissScreenShareError: () => void;
|
||||
|
||||
// UI interactions
|
||||
/**
|
||||
@@ -1926,6 +1931,8 @@ export function createCallViewModel$(
|
||||
reconnecting$: localMembership.reconnecting$,
|
||||
livekitRoomItems$,
|
||||
connected$: localMembership.connected$,
|
||||
screenShareError$: localMembership.screenShareError$,
|
||||
dismissScreenShareError: localMembership.dismissScreenShareError,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ import {
|
||||
flushPromises,
|
||||
mockConfig,
|
||||
mockLivekitRoom,
|
||||
mockLocalParticipant,
|
||||
mockMuteStates,
|
||||
withTestScheduler,
|
||||
ownMemberMock,
|
||||
@@ -41,6 +42,7 @@ import {
|
||||
enterRTCSession,
|
||||
PublishState,
|
||||
TrackState,
|
||||
watchScreenShareToggle,
|
||||
} from "./LocalMember";
|
||||
import {
|
||||
FailToGetOpenIdToken,
|
||||
@@ -69,6 +71,31 @@ vi.mock("@livekit/components-core", () => ({
|
||||
.mockReturnValue(of({ isScreenShareEnabled: false })),
|
||||
}));
|
||||
|
||||
describe("watchScreenShareToggle", () => {
|
||||
it("reports nothing when the toggle completes", async () => {
|
||||
const onError = vi.fn();
|
||||
watchScreenShareToggle(Promise.resolve(), true, logger, onError);
|
||||
await flushPromises();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reports failures other than the user cancelling", async () => {
|
||||
const onError = vi.fn();
|
||||
const e = new Error("NotReadableError");
|
||||
watchScreenShareToggle(Promise.reject(e), true, logger, onError);
|
||||
await flushPromises();
|
||||
expect(onError).toHaveBeenCalledWith(e);
|
||||
});
|
||||
|
||||
it("does not report the user cancelling the picker", async () => {
|
||||
const onError = vi.fn();
|
||||
const cancelled = new DOMException("Permission denied", "NotAllowedError");
|
||||
watchScreenShareToggle(Promise.reject(cancelled), true, logger, onError);
|
||||
await flushPromises();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("LocalMembership", () => {
|
||||
describe("enterRTCSession", () => {
|
||||
it("It joins the correct Session", () => {
|
||||
@@ -857,4 +884,98 @@ describe("LocalMembership", () => {
|
||||
scope.end();
|
||||
});
|
||||
});
|
||||
|
||||
describe("toggleScreenSharing", () => {
|
||||
let originalMediaDevices: MediaDevices | undefined;
|
||||
|
||||
beforeAll(() => {
|
||||
mockConfig();
|
||||
// Screen sharing is only offered when getDisplayMedia is available.
|
||||
originalMediaDevices = navigator.mediaDevices;
|
||||
Object.defineProperty(navigator, "mediaDevices", {
|
||||
configurable: true,
|
||||
value: { getDisplayMedia: vi.fn() },
|
||||
});
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
Object.defineProperty(navigator, "mediaDevices", {
|
||||
configurable: true,
|
||||
value: originalMediaDevices,
|
||||
});
|
||||
});
|
||||
|
||||
const createMembershipWithConnection = (
|
||||
connection: Connection | null,
|
||||
): {
|
||||
scope: ObservableScope;
|
||||
localMembership: ReturnType<typeof createLocalMembership$>;
|
||||
} => {
|
||||
const scope = new ObservableScope();
|
||||
const connectionManagerData = new ConnectionManagerData();
|
||||
if (connection) connectionManagerData.add(connection, []);
|
||||
const localMembership = createLocalMembership$({
|
||||
scope,
|
||||
...defaultCreateLocalMemberValues,
|
||||
connectionManager: {
|
||||
connectionManagerData$: constant(new Epoch(connectionManagerData)),
|
||||
},
|
||||
localTransport$: new BehaviorSubject({
|
||||
advertised$: new BehaviorSubject(aTransport),
|
||||
active$: new BehaviorSubject(aTransportWithSFUConfig),
|
||||
}),
|
||||
});
|
||||
return { scope, localMembership };
|
||||
};
|
||||
|
||||
it("surfaces a failure and clears it on dismiss", async () => {
|
||||
const error = new Error("NotReadableError");
|
||||
const setScreenShareEnabled = vi.fn().mockRejectedValue(error);
|
||||
const connection = {
|
||||
state$: constant(ConnectionState.LivekitConnected),
|
||||
transport: aTransport,
|
||||
livekitRoom: mockLivekitRoom({
|
||||
localParticipant: mockLocalParticipant({
|
||||
isScreenShareEnabled: false,
|
||||
setScreenShareEnabled,
|
||||
}),
|
||||
}),
|
||||
} as unknown as Connection;
|
||||
const { scope, localMembership } =
|
||||
createMembershipWithConnection(connection);
|
||||
await flushPromises();
|
||||
|
||||
expect(localMembership.toggleScreenSharing).not.toBeNull();
|
||||
expect(localMembership.screenShareError$.value).toBeNull();
|
||||
|
||||
localMembership.toggleScreenSharing!();
|
||||
await flushPromises();
|
||||
|
||||
expect(setScreenShareEnabled).toHaveBeenCalledWith(
|
||||
true,
|
||||
expect.any(Object),
|
||||
undefined,
|
||||
);
|
||||
expect(localMembership.screenShareError$.value).toBe(error);
|
||||
|
||||
localMembership.dismissScreenShareError();
|
||||
expect(localMembership.screenShareError$.value).toBeNull();
|
||||
|
||||
scope.end();
|
||||
});
|
||||
|
||||
it("does nothing when there is no local participant", async () => {
|
||||
// No connection means participant$ never resolves to a participant.
|
||||
const { scope, localMembership } = createMembershipWithConnection(null);
|
||||
await flushPromises();
|
||||
|
||||
expect(localMembership.toggleScreenSharing).not.toBeNull();
|
||||
localMembership.toggleScreenSharing!();
|
||||
await flushPromises();
|
||||
|
||||
expect(localMembership.screenShareError$.value).toBeNull();
|
||||
|
||||
scope.end();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -206,6 +206,11 @@ export const createLocalMembership$ = ({
|
||||
* Callback to toggle screen sharing. If null, screen sharing is not possible.
|
||||
*/
|
||||
toggleScreenSharing: (() => void) | null;
|
||||
/**
|
||||
* The last error from toggling screen sharing, until dismissed.
|
||||
*/
|
||||
screenShareError$: Behavior<Error | null>;
|
||||
dismissScreenShareError: () => void;
|
||||
// tracks$: Behavior<LocalTrack[]>;
|
||||
participant$: Behavior<LocalParticipant | null>;
|
||||
connection$: Behavior<Connection | null>;
|
||||
@@ -711,6 +716,7 @@ export const createLocalMembership$ = ({
|
||||
),
|
||||
);
|
||||
|
||||
const screenShareError$ = new BehaviorSubject<Error | null>(null);
|
||||
let toggleScreenSharing: (() => void) | null = null;
|
||||
if (
|
||||
"getDisplayMedia" in (navigator.mediaDevices ?? {}) &&
|
||||
@@ -782,13 +788,18 @@ export const createLocalMembership$ = ({
|
||||
// We also allow screen sharing to be toggled even if the connection
|
||||
// is still initializing or publishing tracks, because there's no
|
||||
// technical reason to disallow this. LiveKit will publish if it can.
|
||||
participant$.value
|
||||
?.setScreenShareEnabled(
|
||||
const participant = participant$.value;
|
||||
if (!participant) return;
|
||||
watchScreenShareToggle(
|
||||
participant.setScreenShareEnabled(
|
||||
targetScreenshareState,
|
||||
screenshareSettings,
|
||||
publishOptions,
|
||||
)
|
||||
.catch(logger.error);
|
||||
),
|
||||
targetScreenshareState,
|
||||
logger,
|
||||
(e) => screenShareError$.next(e),
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
@@ -807,11 +818,44 @@ export const createLocalMembership$ = ({
|
||||
),
|
||||
sharingScreen$,
|
||||
toggleScreenSharing,
|
||||
screenShareError$,
|
||||
dismissScreenShareError: () => screenShareError$.next(null),
|
||||
connection$: localConnection$,
|
||||
internalLoggerRef: logger,
|
||||
};
|
||||
};
|
||||
|
||||
/**
|
||||
* Logs the outcome of a screen share toggle and reports failures.
|
||||
*
|
||||
* getDisplayMedia may legitimately take a long time (the user is choosing
|
||||
* what to share) or never settle at all, so nothing is inferred from silence:
|
||||
* the request and its completion are logged with the elapsed time so that a
|
||||
* hang is visible in the logs, and only an explicit rejection is reported.
|
||||
*
|
||||
* The user cancelling the picker rejects with a NotAllowedError; that is
|
||||
* logged but not reported.
|
||||
*/
|
||||
export function watchScreenShareToggle(
|
||||
toggle: Promise<unknown>,
|
||||
enable: boolean,
|
||||
logger: Logger,
|
||||
onError: (e: Error) => void,
|
||||
): void {
|
||||
const what = `Screen share ${enable ? "start" : "stop"}`;
|
||||
const started = Date.now();
|
||||
const elapsed = (): string => `${Date.now() - started} ms`;
|
||||
logger.info(`${what} requested`);
|
||||
toggle.then(
|
||||
() => logger.info(`${what} completed in ${elapsed()}`),
|
||||
(e: unknown) => {
|
||||
logger.error(`${what} failed after ${elapsed()}:`, e);
|
||||
if (e instanceof DOMException && e.name === "NotAllowedError") return;
|
||||
onError(e instanceof Error ? e : new Error(String(e)));
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
export function observeSharingScreen$(p: Participant): Observable<boolean> {
|
||||
return observeParticipantEvents(
|
||||
p,
|
||||
|
||||
@@ -19,14 +19,17 @@ import {
|
||||
import {
|
||||
type LocalParticipant,
|
||||
type RemoteParticipant,
|
||||
type RemoteTrack,
|
||||
type RemoteTrackPublication,
|
||||
type Room as LivekitRoom,
|
||||
RoomEvent,
|
||||
Track,
|
||||
ConnectionState as LivekitConnectionState,
|
||||
} from "livekit-client";
|
||||
import fetchMock from "fetch-mock";
|
||||
import EventEmitter from "events";
|
||||
import { type IOpenIDToken } from "matrix-js-sdk";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import { logger, type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
|
||||
|
||||
import {
|
||||
@@ -509,6 +512,97 @@ describe("remote participants", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("remote track logging", () => {
|
||||
it("logs remote participant and track events on the connection logger", () => {
|
||||
setupTest();
|
||||
const info = vi.fn();
|
||||
const warn = vi.fn();
|
||||
const testLogger = {
|
||||
getChild: (): unknown => testLogger,
|
||||
info,
|
||||
debug: vi.fn(),
|
||||
warn,
|
||||
error: vi.fn(),
|
||||
} as unknown as Logger;
|
||||
new Connection(
|
||||
{
|
||||
client,
|
||||
roomId: ROOM_ID,
|
||||
transport: livekitFocus,
|
||||
scope: testScope,
|
||||
ownMembershipIdentity: ownMemberMock,
|
||||
livekitRoomFactory: () => fakeLivekitRoom,
|
||||
},
|
||||
testLogger,
|
||||
);
|
||||
info.mockClear(); // drop the constructor log line
|
||||
|
||||
const bob = mockRemoteParticipant({
|
||||
identity: "@bob:example.org:DEV111",
|
||||
sid: "PA_bob",
|
||||
});
|
||||
const pub = {
|
||||
kind: "audio",
|
||||
source: "microphone",
|
||||
trackSid: "TR_mic",
|
||||
isMuted: false,
|
||||
isEncrypted: true,
|
||||
} as unknown as RemoteTrackPublication;
|
||||
const messages = (): string[] => info.mock.calls.map((c) => c[0] as string);
|
||||
|
||||
fakeLivekitRoom.emit(RoomEvent.ParticipantConnected, bob);
|
||||
fakeLivekitRoom.emit(
|
||||
RoomEvent.TrackSubscribed,
|
||||
{} as RemoteTrack,
|
||||
pub,
|
||||
bob,
|
||||
);
|
||||
fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, bob);
|
||||
fakeLivekitRoom.emit(
|
||||
RoomEvent.TrackStreamStateChanged,
|
||||
pub,
|
||||
Track.StreamState.Paused,
|
||||
bob,
|
||||
);
|
||||
expect(messages()).toEqual([
|
||||
"Participant connected: @bob:example.org:DEV111 (PA_bob)",
|
||||
"Subscribed: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true muted=false",
|
||||
"Muted: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true",
|
||||
"Stream paused: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true",
|
||||
]);
|
||||
|
||||
// Encryption status changes are logged; cryptor errors are warnings
|
||||
fakeLivekitRoom.emit(
|
||||
RoomEvent.ParticipantEncryptionStatusChanged,
|
||||
false,
|
||||
bob,
|
||||
);
|
||||
expect(messages().at(-1)).toBe(
|
||||
"Encryption status of @bob:example.org:DEV111: encrypted=false",
|
||||
);
|
||||
const cryptorError = new Error("missing key at index 3");
|
||||
fakeLivekitRoom.emit(RoomEvent.EncryptionError, cryptorError, bob);
|
||||
expect(warn).toHaveBeenCalledWith(
|
||||
"Encryption error for @bob:example.org:DEV111:",
|
||||
cryptorError,
|
||||
);
|
||||
|
||||
// Local mute events are already logged by the Publisher
|
||||
fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, {
|
||||
...fakeLocalParticipant,
|
||||
isLocal: true,
|
||||
} as unknown as LocalParticipant);
|
||||
expect(messages().filter((m) => m.startsWith("Muted"))).toHaveLength(1);
|
||||
|
||||
// Listeners are removed when the scope ends
|
||||
testScope.end();
|
||||
fakeLivekitRoom.emit(RoomEvent.ParticipantDisconnected, bob);
|
||||
expect(
|
||||
messages().filter((m) => m.startsWith("Participant disconnected")),
|
||||
).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
|
||||
//
|
||||
// NOT USED ANYMORE ?
|
||||
//
|
||||
|
||||
@@ -13,8 +13,14 @@ import {
|
||||
import {
|
||||
ConnectionError,
|
||||
ConnectionErrorReason,
|
||||
type Participant,
|
||||
type RemoteParticipant,
|
||||
type RemoteTrackPublication,
|
||||
type Room as LivekitRoom,
|
||||
RoomEvent,
|
||||
type SubscriptionError,
|
||||
type Track,
|
||||
type TrackPublication,
|
||||
} from "livekit-client";
|
||||
import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { BehaviorSubject, map } from "rxjs";
|
||||
@@ -164,6 +170,7 @@ export class Connection {
|
||||
// Only tracks remote participants
|
||||
connectedParticipantsObserver(this.livekitRoom),
|
||||
);
|
||||
this.logRemoteTrackEvents();
|
||||
|
||||
scope.onEnd(() => {
|
||||
this.logger.info(`Connection scope ended, stopping connection`);
|
||||
@@ -171,6 +178,114 @@ export class Connection {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Logs the lifecycle of remote participants and their tracks as seen by
|
||||
* livekit-client, so that rageshakes show what a tile should have been
|
||||
* rendering (published, subscribed, muted, paused) for each remote member.
|
||||
*/
|
||||
private logRemoteTrackEvents(): void {
|
||||
const room = this.livekitRoom;
|
||||
const log = this.logger.getChild("[RemoteTracks]");
|
||||
// The encryption flag matters: if the publisher encrypts but this client
|
||||
// believes the track is unencrypted, the cryptor is bypassed and raw
|
||||
// ciphertext reaches the decoder (audible as loud noise bursts).
|
||||
const track = (pub: TrackPublication, p: Participant): string =>
|
||||
`${pub.kind} ${pub.source} ${pub.trackSid} of ${p.identity} encrypted=${pub.isEncrypted}`;
|
||||
|
||||
const onParticipantConnected = (p: RemoteParticipant): void =>
|
||||
log.info(`Participant connected: ${p.identity} (${p.sid})`);
|
||||
const onParticipantDisconnected = (p: RemoteParticipant): void =>
|
||||
log.info(`Participant disconnected: ${p.identity} (${p.sid})`);
|
||||
const onTrackPublished = (
|
||||
pub: RemoteTrackPublication,
|
||||
p: RemoteParticipant,
|
||||
): void => log.info(`Published: ${track(pub, p)} muted=${pub.isMuted}`);
|
||||
const onTrackUnpublished = (
|
||||
pub: RemoteTrackPublication,
|
||||
p: RemoteParticipant,
|
||||
): void => log.info(`Unpublished: ${track(pub, p)}`);
|
||||
const onTrackSubscribed = (
|
||||
_t: Track,
|
||||
pub: RemoteTrackPublication,
|
||||
p: RemoteParticipant,
|
||||
): void => log.info(`Subscribed: ${track(pub, p)} muted=${pub.isMuted}`);
|
||||
const onTrackUnsubscribed = (
|
||||
_t: Track,
|
||||
pub: RemoteTrackPublication,
|
||||
p: RemoteParticipant,
|
||||
): void => log.info(`Unsubscribed: ${track(pub, p)}`);
|
||||
const onTrackSubscriptionFailed = (
|
||||
trackSid: string,
|
||||
p: RemoteParticipant,
|
||||
reason?: SubscriptionError,
|
||||
): void =>
|
||||
log.warn(
|
||||
`Subscription failed: ${trackSid} of ${p.identity} reason=${reason}`,
|
||||
);
|
||||
// Mute events also fire for the local participant, which the Publisher
|
||||
// already logs.
|
||||
const onTrackMuted = (pub: TrackPublication, p: Participant): void => {
|
||||
if (!p.isLocal) log.info(`Muted: ${track(pub, p)}`);
|
||||
};
|
||||
const onTrackUnmuted = (pub: TrackPublication, p: Participant): void => {
|
||||
if (!p.isLocal) log.info(`Unmuted: ${track(pub, p)}`);
|
||||
};
|
||||
const onTrackStreamStateChanged = (
|
||||
pub: RemoteTrackPublication,
|
||||
state: Track.StreamState,
|
||||
p: RemoteParticipant,
|
||||
): void => log.info(`Stream ${state}: ${track(pub, p)}`);
|
||||
const onEncryptionStatusChanged = (
|
||||
encrypted: boolean,
|
||||
p?: Participant,
|
||||
): void =>
|
||||
log.info(
|
||||
`Encryption status of ${p?.identity ?? "unknown participant"}: encrypted=${encrypted}`,
|
||||
);
|
||||
// livekit-client throttles these per cryptor; they indicate frames being
|
||||
// dropped (missing/invalid key).
|
||||
const onEncryptionError = (error: Error, p?: Participant): void =>
|
||||
log.warn(
|
||||
`Encryption error for ${p?.identity ?? "unknown participant"}:`,
|
||||
error,
|
||||
);
|
||||
|
||||
room
|
||||
.on(RoomEvent.ParticipantConnected, onParticipantConnected)
|
||||
.on(RoomEvent.ParticipantDisconnected, onParticipantDisconnected)
|
||||
.on(RoomEvent.TrackPublished, onTrackPublished)
|
||||
.on(RoomEvent.TrackUnpublished, onTrackUnpublished)
|
||||
.on(RoomEvent.TrackSubscribed, onTrackSubscribed)
|
||||
.on(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed)
|
||||
.on(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed)
|
||||
.on(RoomEvent.TrackMuted, onTrackMuted)
|
||||
.on(RoomEvent.TrackUnmuted, onTrackUnmuted)
|
||||
.on(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged)
|
||||
.on(
|
||||
RoomEvent.ParticipantEncryptionStatusChanged,
|
||||
onEncryptionStatusChanged,
|
||||
)
|
||||
.on(RoomEvent.EncryptionError, onEncryptionError);
|
||||
this.scope.onEnd(() => {
|
||||
room
|
||||
.off(RoomEvent.ParticipantConnected, onParticipantConnected)
|
||||
.off(RoomEvent.ParticipantDisconnected, onParticipantDisconnected)
|
||||
.off(RoomEvent.TrackPublished, onTrackPublished)
|
||||
.off(RoomEvent.TrackUnpublished, onTrackUnpublished)
|
||||
.off(RoomEvent.TrackSubscribed, onTrackSubscribed)
|
||||
.off(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed)
|
||||
.off(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed)
|
||||
.off(RoomEvent.TrackMuted, onTrackMuted)
|
||||
.off(RoomEvent.TrackUnmuted, onTrackUnmuted)
|
||||
.off(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged)
|
||||
.off(
|
||||
RoomEvent.ParticipantEncryptionStatusChanged,
|
||||
onEncryptionStatusChanged,
|
||||
)
|
||||
.off(RoomEvent.EncryptionError, onEncryptionError);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts the connection.
|
||||
*
|
||||
|
||||
@@ -14,7 +14,7 @@ import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts";
|
||||
import {
|
||||
createConnectionManager$,
|
||||
type ConnectionManagerData,
|
||||
ConnectionManagerData,
|
||||
} from "./ConnectionManager.ts";
|
||||
import { type ConnectionFactory } from "./ConnectionFactory.ts";
|
||||
import { type Connection } from "./Connection.ts";
|
||||
@@ -203,6 +203,24 @@ describe("connections$ stream", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("ConnectionManagerData", () => {
|
||||
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 connection = { transport: TRANSPORT_1 } as unknown as Connection;
|
||||
const p = (identity: string): RemoteParticipant =>
|
||||
({ identity }) as unknown as RemoteParticipant;
|
||||
data.add(connection, [p("a")]);
|
||||
data.add({ ...connection } as Connection, [p("b")]);
|
||||
expect(warn).toHaveBeenCalledWith(
|
||||
expect.stringContaining(
|
||||
"second connection to https://lk.example.org: existing [a], adding [b]",
|
||||
),
|
||||
);
|
||||
expect(data.getParticipantsForTransport(TRANSPORT_1)).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("connectionManagerData$ stream", () => {
|
||||
// Used in test to control fake connections' remoteParticipants$ streams
|
||||
let fakeRemoteParticipantsStreams: Map<string, Behavior<RemoteParticipant[]>>;
|
||||
|
||||
@@ -30,7 +30,7 @@ export class ConnectionManagerData {
|
||||
{ connection: Connection; participants: RemoteParticipant[] }
|
||||
> = new Map();
|
||||
|
||||
public constructor() {}
|
||||
public constructor(private readonly logger?: Logger) {}
|
||||
|
||||
public add(connection: Connection, participants: RemoteParticipant[]): void {
|
||||
const key = this.getKey(connection.transport);
|
||||
@@ -38,6 +38,11 @@ export class ConnectionManagerData {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -239,7 +244,7 @@ export function createConnectionManager$({
|
||||
lists.reduce((data, { connection, participants }) => {
|
||||
data.add(connection, participants);
|
||||
return data;
|
||||
}, new ConnectionManagerData()),
|
||||
}, new ConnectionManagerData(logger)),
|
||||
epoch,
|
||||
),
|
||||
),
|
||||
|
||||
@@ -5,7 +5,8 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { describe, test, expect, beforeEach, afterEach } from "vitest";
|
||||
import { describe, test, expect, beforeEach, afterEach, vi } from "vitest";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type CallMembership,
|
||||
type LivekitTransport,
|
||||
@@ -77,6 +78,7 @@ 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(),
|
||||
@@ -108,6 +110,9 @@ test("should signal participant not yet connected to livekit", async () => {
|
||||
return true;
|
||||
},
|
||||
);
|
||||
expect(info).toHaveBeenCalledWith(
|
||||
"[RemoteMatrixLivekitMembers] @bob:example.org:DEV000: LiveKit participant missing on no connection",
|
||||
);
|
||||
});
|
||||
|
||||
// Helper to create epoch'ed memberships$ and membershipsWithTransport$ from memberships observable.
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
type LivekitTransportConfig,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { combineLatest, filter, map } from "rxjs";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
|
||||
import { type Behavior } from "../../Behavior";
|
||||
import { type IConnectionManager } from "./ConnectionManager";
|
||||
@@ -110,13 +111,17 @@ export function createRemoteMatrixLivekitMembers$({
|
||||
const participants = transport
|
||||
? managerData.getParticipantsForTransport(transport)
|
||||
: [];
|
||||
const participant =
|
||||
participants.find(
|
||||
(p) => p.identity == membership.rtcBackendIdentity,
|
||||
) ?? null;
|
||||
const matches = participants.filter(
|
||||
(p) => p.identity == membership.rtcBackendIdentity,
|
||||
);
|
||||
const participant = matches[0] ?? null;
|
||||
const connection = transport
|
||||
? managerData.getConnectionForTransport(transport)
|
||||
: null;
|
||||
if (matches.length > 1)
|
||||
logger.warn(
|
||||
`[RemoteMatrixLivekitMembers] ${membership.rtcBackendIdentity}: ${matches.length} LiveKit participants match (sids ${matches.map((p) => p.sid).join(", ")}), using ${participant?.sid}`,
|
||||
);
|
||||
|
||||
yield {
|
||||
// This could just be the backend identity without the other keys.
|
||||
@@ -133,8 +138,16 @@ export function createRemoteMatrixLivekitMembers$({
|
||||
}
|
||||
},
|
||||
// Each update where the key of the generator array do not change will result in updates to the `data$` behavior.
|
||||
(scope, data$, userId, _deviceId, _memberId, _rtcBackendIdentity) => {
|
||||
(scope, data$, userId, _deviceId, _memberId, rtcBackendIdentity) => {
|
||||
const { participant$, ...rest } = scope.splitBehavior(data$);
|
||||
// Log whether the member could be matched to a LiveKit participant,
|
||||
// since a tile shows "waiting for media" for as long as it cannot.
|
||||
participant$.pipe(scope.bind()).subscribe((p) => {
|
||||
const url = data$.value.connection?.transport.livekit_service_url;
|
||||
logger.info(
|
||||
`[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"} on ${url ?? "no connection"}`,
|
||||
);
|
||||
});
|
||||
// will only get called once per backend identity.
|
||||
// updates to data$ and as a result to displayName$ and mxcAvatarUrl$ are more frequent.
|
||||
return {
|
||||
|
||||
Reference in New Issue
Block a user