mirror of
https://github.com/vector-im/element-call.git
synced 2026-10-16 23:45:50 +00:00
Robin
3927c2b7dc
A transport may soon carry a LiveKit SFU `url` instead of, or as well as, a `livekit_service_url`, so a service URL alone no longer identifies it. Pass the whole transport to the tiles' debug stats and print it as JSON in log lines instead of picking out the service URL.
1021 lines
34 KiB
TypeScript
1021 lines
34 KiB
TypeScript
/*
|
|
Copyright 2025 Element Creations Ltd.
|
|
|
|
SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
|
Please see LICENSE in the repository root for full details.
|
|
*/
|
|
|
|
import {
|
|
ConnectionState as LivekitConnectionState,
|
|
type Participant,
|
|
ParticipantEvent,
|
|
type LocalParticipant,
|
|
type ScreenShareCaptureOptions,
|
|
type TrackPublishOptions,
|
|
RoomEvent,
|
|
MediaDeviceFailure,
|
|
} from "livekit-client";
|
|
import { observeParticipantEvents } from "@livekit/components-core";
|
|
import { type MatrixClient } from "matrix-js-sdk";
|
|
import {
|
|
Status as RTCSessionStatus,
|
|
type LivekitTransport,
|
|
type MatrixRTCSession,
|
|
type RTCCallIntent,
|
|
type RTCNotificationType,
|
|
} from "matrix-js-sdk/lib/matrixrtc";
|
|
import {
|
|
BehaviorSubject,
|
|
catchError,
|
|
combineLatest,
|
|
distinctUntilChanged,
|
|
from,
|
|
fromEvent,
|
|
map,
|
|
type Observable,
|
|
of,
|
|
pairwise,
|
|
startWith,
|
|
switchMap,
|
|
tap,
|
|
NEVER,
|
|
concat,
|
|
race,
|
|
Subject,
|
|
} from "rxjs";
|
|
import { type Logger } from "matrix-js-sdk/lib/logger";
|
|
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 Publisher } from "./Publisher.ts";
|
|
import { type MuteStates } from "../../MuteStates.ts";
|
|
import {
|
|
ElementCallError,
|
|
FailToStartLivekitConnection,
|
|
MembershipManagerError,
|
|
UnknownCallError,
|
|
} from "../../../utils/errors.ts";
|
|
import { type HostBridge } from "../../../HostBridge.ts";
|
|
|
|
import { PosthogAnalytics } from "../../../analytics/PosthogAnalytics.ts";
|
|
import {
|
|
advancedScreenShare,
|
|
screenShareResolution,
|
|
screenShareFramerate,
|
|
screenShareBitrate,
|
|
screenShareCodec,
|
|
parseResolution,
|
|
} from "../../../settings/settings.ts";
|
|
import {
|
|
MatrixRTCMode,
|
|
type ResolvedDelayedLeaveTimings,
|
|
} from "../../../config/ConfigOptions.ts";
|
|
import { Config } from "../../../config/Config.ts";
|
|
import {
|
|
type ConnectionState,
|
|
type Connection,
|
|
} from "../remoteMembers/Connection.ts";
|
|
import { type HomeserverConnected } from "./HomeserverConnected.ts";
|
|
import { type LocalTransport } from "./LocalTransport.ts";
|
|
import { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts";
|
|
|
|
export enum TransportState {
|
|
/** Not even a transport is available to the LocalMembership */
|
|
Waiting = "transport_waiting",
|
|
}
|
|
|
|
export enum PublishState {
|
|
WaitingForUser = "publish_waiting_for_user",
|
|
// XXX: This state is removed for now since we do not have full control over
|
|
// track publication anymore with the publisher abstraction, might come back in the future?
|
|
// /** Implies lk connection is connected */
|
|
// Starting = "publish_start_publishing",
|
|
/** Implies lk connection is connected */
|
|
Publishing = "publish_publishing",
|
|
}
|
|
|
|
// TODO not sure how to map that correctly with the
|
|
// new publisher that does not manage tracks itself anymore
|
|
export enum TrackState {
|
|
/** The track is waiting for user input to create tracks (waiting to call `startTracks()`) */
|
|
WaitingForUser = "tracks_waiting_for_user",
|
|
// XXX: This state is removed for now since we do not have full control over
|
|
// track creation anymore with the publisher abstraction, might come back in the future?
|
|
// /** Implies lk connection is connected */
|
|
// Creating = "tracks_creating",
|
|
/** Implies lk connection is connected */
|
|
Ready = "tracks_ready",
|
|
}
|
|
|
|
export type LocalMemberMediaState =
|
|
| {
|
|
tracks: TrackState;
|
|
connection: ConnectionState;
|
|
}
|
|
| PublishState
|
|
| ElementCallError;
|
|
export type LocalMemberState =
|
|
| ElementCallError
|
|
| TransportState.Waiting
|
|
| {
|
|
media: LocalMemberMediaState;
|
|
matrix: ElementCallError | RTCSessionStatus;
|
|
};
|
|
|
|
/*
|
|
* - get transport to use
|
|
* - get openId + jwt token
|
|
* - wait for createTrack() call
|
|
* - create tracks
|
|
* - wait for join() call
|
|
* - Publisher.publishTracks()
|
|
* - send join state/sticky event
|
|
*/
|
|
|
|
interface Props {
|
|
// TODO add a comment into some code style readme or file header callviewmodel
|
|
// that the inputs for those createSomething$() functions should NOT contain any js-sdk objectes
|
|
scope: ObservableScope;
|
|
muteStates: MuteStates;
|
|
connectionManager: IConnectionManager;
|
|
createPublisherFactory: (connection: Connection) => Publisher;
|
|
joinMatrixRTC: (
|
|
transport: LivekitTransport,
|
|
delayedLeaveTimings: ResolvedDelayedLeaveTimings,
|
|
) => void;
|
|
homeserverConnected: HomeserverConnected;
|
|
roomId: string;
|
|
ownMembershipIdentity: CallMembershipIdentityParts;
|
|
localTransport$: Observable<LocalTransport>;
|
|
client: Pick<MatrixClient, "getDeviceId" | "getOpenIdToken">;
|
|
matrixRTCSession: Pick<MatrixRTCSession, "updateCallIntent" | "leave">;
|
|
/** Whether to hide the screen-sharing button. */
|
|
hideScreensharing: boolean;
|
|
/** The application hosting Element Call, to be kept informed of join/leave. */
|
|
hostBridge: HostBridge;
|
|
baseUrl: string;
|
|
delayId$: Behavior<string | null>;
|
|
matrixRTCMode: MatrixRTCMode;
|
|
logger: Logger;
|
|
}
|
|
|
|
/**
|
|
* This class is responsible for managing the own membership in a room.
|
|
* We want
|
|
* - a publisher
|
|
* -
|
|
* @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.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.
|
|
* @param props.localTransport The transport to advertise in our membership.
|
|
* @param props.logger The logger to use.
|
|
* @param props.muteStates The mute states for video and audio.
|
|
* @param props.matrixRTCSession The matrix RTC session to join.
|
|
* @param props.baseUrl Base URL of the homeserver.
|
|
* @param props.delayId$ ID of the delayed leave event to delegate to the SFU.
|
|
* @param props.roomId The room ID used as the call identifier in analytics events.
|
|
* @param props.hideScreensharing Whether to hide the screen-sharing button.
|
|
* @param props.hostBridge The application hosting Element Call.
|
|
* @returns
|
|
* - publisher: The handle to create tracks and publish them to the room.
|
|
* - connected$: the current connection state. Including matrix server and livekit server connection. (only considering the livekit server we are using for our own media publication)
|
|
* - transport$: the transport object the ownMembership$ ended up using.
|
|
* - connectionState: the current connection state. Including matrix server and livekit server connection.
|
|
* - sharingScreen$: Whether we are sharing our screen. `undefined` if we cannot share the screen.
|
|
*/
|
|
export const createLocalMembership$ = ({
|
|
scope,
|
|
connectionManager,
|
|
localTransport$: localTransportWithErrors$,
|
|
homeserverConnected,
|
|
createPublisherFactory,
|
|
joinMatrixRTC,
|
|
logger: parentLogger,
|
|
muteStates,
|
|
client,
|
|
matrixRTCSession,
|
|
baseUrl,
|
|
roomId,
|
|
hideScreensharing,
|
|
hostBridge,
|
|
ownMembershipIdentity,
|
|
delayId$,
|
|
matrixRTCMode,
|
|
}: Props): {
|
|
/**
|
|
* This request to start audio and video tracks.
|
|
* Can be called early to pre-emptively get media permissions and start devices.
|
|
*/
|
|
startTracks: () => void;
|
|
/**
|
|
* This sets a inner state (shouldPublish) to true and instructs the js-sdk and livekit to keep the user
|
|
* connected to matrix and livekit.
|
|
*/
|
|
requestJoinAndPublish: () => void;
|
|
requestDisconnect: () => void;
|
|
localMemberState$: Behavior<LocalMemberState>;
|
|
sharingScreen$: Behavior<boolean>;
|
|
/**
|
|
* 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>;
|
|
/**
|
|
* Tracks the homserver and livekit connected state and based on that computes reconnecting.
|
|
*/
|
|
reconnecting$: Behavior<boolean>;
|
|
/** Shorthand for homeserverConnected.rtcSession === Status.Disconnected
|
|
* Direct translation to the js-sdk membership manager connection `Status`.
|
|
*/
|
|
disconnected$: Behavior<boolean>;
|
|
/**
|
|
* Fully connected
|
|
*/
|
|
connected$: Behavior<boolean>;
|
|
internalLoggerRef: Logger;
|
|
} => {
|
|
const logger = parentLogger.getChild("[LocalMembership]");
|
|
logger.debug(`Creating local membership..`);
|
|
|
|
// Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error.
|
|
const fatalTransportError$ = new Subject<ElementCallError>();
|
|
const localTransport$ = localTransportWithErrors$.pipe(
|
|
catchError((e: unknown) => {
|
|
let error: ElementCallError;
|
|
if (e instanceof ElementCallError) {
|
|
error = e;
|
|
} else {
|
|
error = new UnknownCallError(
|
|
e instanceof Error
|
|
? e
|
|
: new Error("Unknown error from localTransport"),
|
|
);
|
|
}
|
|
fatalTransportError$.next(error);
|
|
return NEVER; // Make this Observable swallow the error
|
|
}),
|
|
);
|
|
|
|
async function checkDelegationSupport(
|
|
endpointUrl: string,
|
|
serviceName: string,
|
|
): Promise<boolean> {
|
|
logger.info(`Checking whether ${serviceName} supports delegation…`);
|
|
try {
|
|
// Bluntly hit the endpoint without auth to check for a 404. Unfortunately
|
|
// we can't wrap this in a retry loop, as many servers don't just disable
|
|
// delegation support, but in fact are from a time before the endpoint
|
|
// existed at all, therefore we can hit CORS errors which would just gum
|
|
// up the retry loop. (May be revisited after Matrix 2.0.)
|
|
const res = await fetch(endpointUrl, { method: "POST" });
|
|
if (res.status === 404) {
|
|
logger.warn(`${serviceName} does not support delegation`);
|
|
return false;
|
|
} else {
|
|
logger.info(`${serviceName} supports delegation`);
|
|
return true;
|
|
}
|
|
} catch (e) {
|
|
logger.warn(
|
|
`Failed to determine whether ${serviceName} supports delegation, assuming no support`,
|
|
e,
|
|
);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
const homeserverSupportsDelegation = checkDelegationSupport(
|
|
baseUrl +
|
|
"/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave",
|
|
"homeserver",
|
|
);
|
|
|
|
// The transport that we will advertise in our membership, paired with info as
|
|
// to whether delayed event delegation is supported
|
|
const joinParams$ = scope.behavior(
|
|
localTransport$.pipe(
|
|
switchMap(async ({ transport }) => {
|
|
const transportSupportsDelegation = checkDelegationSupport(
|
|
transport.livekit_service_url + "/delegate_delayed_leave",
|
|
`transport ${transport.livekit_service_url}`,
|
|
);
|
|
return {
|
|
transport,
|
|
delegationSupported:
|
|
(await homeserverSupportsDelegation) ||
|
|
(await transportSupportsDelegation),
|
|
};
|
|
}),
|
|
),
|
|
null,
|
|
);
|
|
|
|
// 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),
|
|
),
|
|
tap((connection) => {
|
|
logger.info(
|
|
`Local connection updated: ${JSON.stringify(connection?.transport)}`,
|
|
);
|
|
}),
|
|
),
|
|
null,
|
|
);
|
|
|
|
// Tracks error that happen when creating the local tracks.
|
|
const mediaErrors$ = localConnection$.pipe(
|
|
switchMap((connection) => {
|
|
if (!connection) {
|
|
return of(null);
|
|
} else {
|
|
return fromEvent(
|
|
connection.livekitRoom,
|
|
RoomEvent.MediaDevicesError,
|
|
(error: Error) => {
|
|
return MediaDeviceFailure.getFailure(error) ?? null;
|
|
},
|
|
);
|
|
}
|
|
}),
|
|
);
|
|
|
|
mediaErrors$.pipe(scope.bind()).subscribe((error) => {
|
|
if (error) {
|
|
// This is a MediaDevice error, can be PermissionDenied, NotFound, DeviceInUse, Other.
|
|
// Will also occurs if you cancel screen sharing browser prompt.
|
|
// This is not necessarily fatal, since the user might be able to join without media.
|
|
// XXX We might want to give some user feedback here to let them know their media is not working.
|
|
logger.error(`Failed to create local tracks:`, error);
|
|
}
|
|
});
|
|
// MATRIX RELATED
|
|
|
|
// This should be used in a combineLatest with publisher$ to connect.
|
|
// to make it possible to call startTracks before the preferredTransport$ has resolved.
|
|
const trackStartRequested = Promise.withResolvers<void>();
|
|
|
|
// This should be used in a combineLatest with publisher$ to connect.
|
|
// to make it possible to call startTracks before the preferredTransport$ has resolved.
|
|
const joinAndPublishRequested$ = new BehaviorSubject(false);
|
|
|
|
/**
|
|
* The publisher is stored in here an abstracts creating and publishing tracks.
|
|
*/
|
|
const publisher$ = new BehaviorSubject<Publisher | null>(null);
|
|
|
|
const startTracks = (): void => {
|
|
trackStartRequested.resolve();
|
|
// This used to return the tracks, but now they are only accessible via the publisher.
|
|
};
|
|
|
|
const requestJoinAndPublish = (): void => {
|
|
trackStartRequested.resolve();
|
|
joinAndPublishRequested$.next(true);
|
|
};
|
|
|
|
const requestDisconnect = (): void => {
|
|
joinAndPublishRequested$.next(false);
|
|
};
|
|
|
|
// Take care of the publisher$
|
|
// create a new one as soon as a local Connection is available
|
|
//
|
|
// Recreate a new one once the local connection changes
|
|
// - stop publishing
|
|
// - destruct all current streams
|
|
// - overwrite current publisher
|
|
scope.reconcile(localConnection$, async (connection) => {
|
|
logger.info(
|
|
"reconcile based on new localConnection:",
|
|
JSON.stringify(connection?.transport),
|
|
);
|
|
if (connection !== null) {
|
|
const publisher = createPublisherFactory(connection);
|
|
publisher$.next(publisher);
|
|
|
|
// Clean-up callback
|
|
return Promise.resolve(async (): Promise<void> => {
|
|
await publisher.destroy();
|
|
});
|
|
}
|
|
});
|
|
|
|
// Use reconcile here to not run concurrent createAndSetupTracks calls
|
|
// `tracks$` will update once they are ready.
|
|
scope.reconcile(
|
|
scope.behavior(
|
|
combineLatest([
|
|
publisher$ /*, tracks$*/,
|
|
from(trackStartRequested.promise),
|
|
]),
|
|
null,
|
|
),
|
|
async (valueIfReady) => {
|
|
if (!valueIfReady) return;
|
|
const [publisher] = valueIfReady;
|
|
if (publisher) {
|
|
await publisher.createAndSetupTracks().catch((e) => logger.error(e));
|
|
}
|
|
},
|
|
);
|
|
|
|
// Based on `connectRequested$` we start publishing tracks. (once they are there!)
|
|
scope.reconcile(
|
|
scope.behavior(combineLatest([publisher$, joinAndPublishRequested$])),
|
|
async ([publisher, shouldJoinAndPublish]) => {
|
|
// Get the current publishing state to avoid redundant calls.
|
|
const isPublishing = publisher?.shouldPublish === true;
|
|
if (shouldJoinAndPublish && !isPublishing) {
|
|
try {
|
|
await publisher?.startPublishing();
|
|
} catch (error) {
|
|
const message =
|
|
error instanceof Error ? error.message : String(error);
|
|
setPublishError(new FailToStartLivekitConnection(message));
|
|
}
|
|
} else if (isPublishing) {
|
|
try {
|
|
await publisher?.stopPublishing();
|
|
} catch (error) {
|
|
setPublishError(new UnknownCallError(error as Error));
|
|
}
|
|
}
|
|
},
|
|
);
|
|
|
|
// STATE COMPUTATION
|
|
|
|
// These are non fatal since we can join a room and concume media even though publishing failed.
|
|
const publishError$ = new BehaviorSubject<ElementCallError | null>(null);
|
|
const setPublishError = (e: ElementCallError): void => {
|
|
if (publishError$.value !== null) {
|
|
logger.error("Multiple Media Errors:", e);
|
|
} else {
|
|
publishError$.next(e);
|
|
}
|
|
};
|
|
|
|
const localConnectionState$ = localConnection$.pipe(
|
|
switchMap((connection) =>
|
|
connection ? connection.state$ : of({ state: "initializing" as const }),
|
|
),
|
|
);
|
|
|
|
const mediaState$: Behavior<LocalMemberMediaState> = scope.behavior(
|
|
combineLatest([
|
|
localConnectionState$,
|
|
joinAndPublishRequested$,
|
|
from(trackStartRequested.promise).pipe(
|
|
map(() => true),
|
|
startWith(false),
|
|
),
|
|
]).pipe(
|
|
map(([localConnectionState, shouldPublish, shouldStartTracks]) => {
|
|
const trackState: TrackState = shouldStartTracks
|
|
? TrackState.Ready
|
|
: TrackState.WaitingForUser;
|
|
|
|
if (
|
|
localConnectionState.state !== "authenticated" ||
|
|
localConnectionState.livekitState !==
|
|
LivekitConnectionState.Connected ||
|
|
trackState !== TrackState.Ready
|
|
)
|
|
return {
|
|
connection: localConnectionState,
|
|
tracks: trackState,
|
|
};
|
|
if (!shouldPublish) return PublishState.WaitingForUser;
|
|
// if (!publishing) return PublishState.Starting;
|
|
return PublishState.Publishing;
|
|
}),
|
|
distinctUntilChanged(deepCompare),
|
|
),
|
|
);
|
|
const fatalMatrixError$ = new BehaviorSubject<ElementCallError | null>(null);
|
|
const setMatrixError = (e: ElementCallError): void => {
|
|
if (fatalMatrixError$.value !== null) {
|
|
logger.error("Multiple Matrix Errors:", e);
|
|
} else {
|
|
fatalMatrixError$.next(e);
|
|
}
|
|
};
|
|
|
|
const localMemberState$ = scope.behavior<LocalMemberState>(
|
|
concat(
|
|
// Waiting until…
|
|
of(TransportState.Waiting),
|
|
race(
|
|
// either there is a fatal transport error
|
|
fatalTransportError$,
|
|
// or the transport is available.
|
|
localTransport$.pipe(
|
|
switchMap(() =>
|
|
// Once available, track session/media state.
|
|
combineLatest([
|
|
mediaState$,
|
|
homeserverConnected.rtsSession$,
|
|
fatalMatrixError$,
|
|
publishError$,
|
|
]).pipe(
|
|
map(
|
|
([
|
|
mediaState,
|
|
rtcSessionStatus,
|
|
fatalMatrixError,
|
|
publishError,
|
|
]) => ({
|
|
matrix: fatalMatrixError ?? rtcSessionStatus,
|
|
media: publishError ?? mediaState,
|
|
}),
|
|
),
|
|
),
|
|
),
|
|
),
|
|
),
|
|
),
|
|
);
|
|
|
|
/**
|
|
* The disconnect reason for the combined Matrix + LiveKit connection, or null
|
|
* when fully connected. Homeserver reasons take priority over livekit.
|
|
* Both connectivity state and reason come from the same combineLatest emission,
|
|
* avoiding any race between the two.
|
|
*/
|
|
const connectionDisconnectReason$ = scope.behavior(
|
|
combineLatest([
|
|
homeserverConnected.combined$,
|
|
localConnectionState$.pipe(
|
|
map(
|
|
(state) =>
|
|
state.state === "authenticated" &&
|
|
state.livekitState === LivekitConnectionState.Connected,
|
|
),
|
|
),
|
|
]).pipe(
|
|
map(([[hsConnected, hsReason], livekitConnected]) => {
|
|
if (!hsConnected) return hsReason!;
|
|
if (!livekitConnected) return "livekit" as const;
|
|
return null;
|
|
}),
|
|
tap((v) => logger.debug("livekit+matrix: Connected state changed", v)),
|
|
),
|
|
);
|
|
|
|
/**
|
|
* Whether we are "fully" connected to the call. Accounts for both the
|
|
* connection to the MatrixRTC session and the LiveKit publish connection.
|
|
*/
|
|
const matrixAndLivekitConnected$ = scope.behavior(
|
|
connectionDisconnectReason$.pipe(map((reason) => reason === null)),
|
|
);
|
|
|
|
/**
|
|
* Whether we should tell the user that we're reconnecting to the call.
|
|
*/
|
|
const reconnecting$ = scope.behavior(
|
|
matrixAndLivekitConnected$.pipe(
|
|
pairwise(),
|
|
map(([prev, current]) => prev === true && current === false),
|
|
),
|
|
false,
|
|
);
|
|
|
|
let reconnectStart: {
|
|
time: number;
|
|
reason: NonNullable<(typeof connectionDisconnectReason$)["value"]>;
|
|
} | null = null;
|
|
connectionDisconnectReason$
|
|
.pipe(distinctUntilChanged(), pairwise(), scope.bind())
|
|
.subscribe(([prev, reason]) => {
|
|
if (reason !== null) {
|
|
// Only begin tracking when transitioning FROM connected (null → non-null).
|
|
// This prevents the initial startup phase — where we may be non-null before
|
|
// the first real connection — from being counted as a reconnect.
|
|
if (prev === null) {
|
|
reconnectStart ??= { time: Date.now(), reason };
|
|
}
|
|
} else if (reconnectStart !== null) {
|
|
PosthogAnalytics.instance.eventCallReconnecting.track(
|
|
roomId,
|
|
reconnectStart.reason,
|
|
(Date.now() - reconnectStart.time) / 1000,
|
|
);
|
|
PosthogAnalytics.instance.eventCallEnded.cacheReconnecting(
|
|
reconnectStart.reason,
|
|
);
|
|
reconnectStart = null;
|
|
}
|
|
});
|
|
|
|
// inform the host about the connect and disconnect intent from the user.
|
|
scope
|
|
.behavior(joinAndPublishRequested$.pipe(pairwise(), scope.bind()), [
|
|
undefined,
|
|
joinAndPublishRequested$.value,
|
|
])
|
|
.subscribe(([prev, current]) => {
|
|
// JOIN prev=false (was left) => current-true (now joiend)
|
|
if (!prev && current) {
|
|
hostBridge.notifyJoined().catch((e) => {
|
|
logger.error("Failed to notify the host that we joined", e);
|
|
});
|
|
}
|
|
// LEAVE prev=false (was joined) => current-true (now left)
|
|
if (prev && !current) {
|
|
hostBridge.notifyHungUp().catch((e) => {
|
|
logger.error("Failed to notify the host that we hung up", e);
|
|
});
|
|
}
|
|
});
|
|
|
|
muteStates.video.enabled$.pipe(scope.bind()).subscribe((videoEnabled) => {
|
|
void matrixRTCSession
|
|
.updateCallIntent(videoEnabled ? "video" : "audio")
|
|
.catch((e) => {
|
|
if (e instanceof Error && e.message === "Not connected yet") {
|
|
logger.debug(
|
|
"'not connected yet' while updating the call intent (this is expected on startup)",
|
|
);
|
|
} else {
|
|
throw e;
|
|
}
|
|
});
|
|
});
|
|
|
|
// Join and leave the session as needed
|
|
scope.reconcile(
|
|
scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])),
|
|
async ([joinParams, shouldConnect]) => {
|
|
if (!joinParams) return;
|
|
// if shouldConnect=false we will do the disconnect as the cleanup from the previous reconcile iteration.
|
|
if (!shouldConnect) return;
|
|
const sessionConfig = Config.get().matrix_rtc_session;
|
|
|
|
try {
|
|
joinMatrixRTC(
|
|
joinParams.transport,
|
|
joinParams.delegationSupported
|
|
? sessionConfig.delegated_delayed_leave
|
|
: sessionConfig.delayed_leave,
|
|
);
|
|
} catch (error) {
|
|
logger.error("Error entering RTC session", error);
|
|
if (error instanceof Error)
|
|
setMatrixError(new MembershipManagerError(error));
|
|
}
|
|
|
|
return Promise.resolve(async (): Promise<void> => {
|
|
try {
|
|
// TODO Update matrixRTCSession to allow udpating the transport without leaving the session!
|
|
await matrixRTCSession.leave(1000);
|
|
} catch (e) {
|
|
logger.error("Error leaving RTC session", e);
|
|
}
|
|
});
|
|
},
|
|
);
|
|
|
|
const participant$ = scope.behavior(
|
|
localConnection$.pipe(
|
|
map((c) => c?.livekitRoom?.localParticipant ?? null),
|
|
tap((p) => {
|
|
logger.debug("participant$ updated:", p?.identity);
|
|
}),
|
|
),
|
|
);
|
|
|
|
// Delegate delayed leaves to the SFU
|
|
scope.reconcile(
|
|
scope.behavior(combineLatest([joinParams$, delayId$])),
|
|
async ([joinParams, delayId]) => {
|
|
if (joinParams?.delegationSupported && delayId !== null) {
|
|
try {
|
|
// This will technically cause the service to issue a new JWT token,
|
|
// but it's safe to discard. We're only interested in triggering
|
|
// delegation.
|
|
await getSFUConfigWithOpenID(
|
|
client,
|
|
ownMembershipIdentity,
|
|
joinParams.transport.livekit_service_url,
|
|
roomId,
|
|
{ matrixRTCMode, delayEndpointBaseUrl: baseUrl, delayId },
|
|
logger,
|
|
);
|
|
} catch (e) {
|
|
// TODO: Surface this to the user as a service interruption?
|
|
logger.error(
|
|
`Failed to delegate leave to ${joinParams.transport.livekit_service_url}`,
|
|
e,
|
|
);
|
|
}
|
|
}
|
|
},
|
|
);
|
|
|
|
// Pause upstream of all local media tracks when we're disconnected from
|
|
// MatrixRTC, because it can be an unpleasant surprise for the app to say
|
|
// 'reconnecting' and yet still be transmitting your media to others.
|
|
// We use matrixConnected$ rather than reconnecting$ because we want to
|
|
// pause tracks during the initial joining sequence too until we're sure
|
|
// that our own media is displayed on screen.
|
|
// TODO refactor this based no livekitState$
|
|
combineLatest([participant$, homeserverConnected.combined$])
|
|
.pipe(scope.bind())
|
|
.subscribe(([participant, [connected]]) => {
|
|
if (!participant) return;
|
|
const publications = participant.trackPublications.values();
|
|
if (connected) {
|
|
for (const p of publications) {
|
|
if (p.track?.isUpstreamPaused === true) {
|
|
const kind = p.track.kind;
|
|
logger.info(
|
|
`Resuming ${kind} track (MatrixRTC connection present)`,
|
|
);
|
|
p.track
|
|
.resumeUpstream()
|
|
.catch((e) =>
|
|
logger.error(
|
|
`Failed to resume ${kind} track after MatrixRTC reconnection`,
|
|
e,
|
|
),
|
|
);
|
|
}
|
|
}
|
|
} else {
|
|
for (const p of publications) {
|
|
if (p.track?.isUpstreamPaused === false) {
|
|
const kind = p.track.kind;
|
|
logger.info(
|
|
`Pausing ${kind} track (uncertain MatrixRTC connection)`,
|
|
);
|
|
p.track
|
|
.pauseUpstream()
|
|
.catch((e) =>
|
|
logger.error(
|
|
`Failed to pause ${kind} track after entering uncertain MatrixRTC connection`,
|
|
e,
|
|
),
|
|
);
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
/**
|
|
* Whether the user is currently sharing their screen.
|
|
*/
|
|
const sharingScreen$ = scope.behavior(
|
|
participant$.pipe(
|
|
switchMap((p) => (p !== null ? observeSharingScreen$(p) : of(false))),
|
|
),
|
|
);
|
|
|
|
const screenShareError$ = new BehaviorSubject<Error | null>(null);
|
|
let toggleScreenSharing: (() => void) | null = null;
|
|
if (
|
|
"getDisplayMedia" in (navigator.mediaDevices ?? {}) &&
|
|
!hideScreensharing
|
|
) {
|
|
toggleScreenSharing = (): void => {
|
|
const screenshareSettings: ScreenShareCaptureOptions = {
|
|
// Screen share audio shouldn't have any filtering.
|
|
// "echoCancellation" is purposely excluded, as setting it to
|
|
// false causes the screen share audio track to include
|
|
// an echo of the incoming participant's voice
|
|
audio: {
|
|
autoGainControl: false,
|
|
noiseSuppression: false,
|
|
voiceIsolation: false,
|
|
},
|
|
selfBrowserSurface: "include",
|
|
surfaceSwitching: "include",
|
|
systemAudio: "include",
|
|
};
|
|
|
|
let publishOptions: TrackPublishOptions | undefined;
|
|
|
|
if (advancedScreenShare.getValue()) {
|
|
// User has advanced screen share settings enabled
|
|
const { width, height } = parseResolution(
|
|
screenShareResolution.getValue(),
|
|
);
|
|
const fps = screenShareFramerate.getValue();
|
|
const bps = screenShareBitrate.getValue();
|
|
const codec = screenShareCodec.getValue();
|
|
|
|
screenshareSettings.resolution = {
|
|
width,
|
|
height,
|
|
frameRate: fps,
|
|
};
|
|
|
|
publishOptions = {
|
|
screenShareEncoding: {
|
|
maxBitrate: bps,
|
|
maxFramerate: fps,
|
|
},
|
|
videoCodec: codec,
|
|
};
|
|
} else {
|
|
// Fall back to config.json settings if available
|
|
const screenConf = Config.get().media_quality?.screen_share;
|
|
if (screenConf?.max_resolution) {
|
|
screenshareSettings.resolution = {
|
|
width: Math.round((screenConf.max_resolution * 16) / 9),
|
|
height: screenConf.max_resolution,
|
|
frameRate: screenConf.max_framerate ?? 30,
|
|
};
|
|
}
|
|
}
|
|
|
|
const targetScreenshareState = !sharingScreen$.value;
|
|
logger.info(
|
|
`toggleScreenSharing called. Switching ${
|
|
targetScreenshareState ? "On" : "Off"
|
|
}`,
|
|
);
|
|
// If a connection is ready, toggle screen sharing.
|
|
// We deliberately do nothing in the case of a null connection because
|
|
// it looks nice for the call control buttons to all become available
|
|
// at once upon joining the call, rather than introducing a disabled
|
|
// state. The user can just click again.
|
|
// 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.
|
|
const participant = participant$.value;
|
|
if (!participant) return;
|
|
watchScreenShareToggle(
|
|
participant.setScreenShareEnabled(
|
|
targetScreenshareState,
|
|
screenshareSettings,
|
|
publishOptions,
|
|
),
|
|
targetScreenshareState,
|
|
logger,
|
|
(e) => screenShareError$.next(e),
|
|
);
|
|
};
|
|
}
|
|
|
|
return {
|
|
startTracks,
|
|
requestJoinAndPublish,
|
|
requestDisconnect,
|
|
localMemberState$,
|
|
participant$,
|
|
reconnecting$,
|
|
connected$: matrixAndLivekitConnected$,
|
|
disconnected$: scope.behavior(
|
|
homeserverConnected.rtsSession$.pipe(
|
|
map((state) => state === RTCSessionStatus.Disconnected),
|
|
),
|
|
),
|
|
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,
|
|
ParticipantEvent.TrackPublished,
|
|
ParticipantEvent.TrackUnpublished,
|
|
ParticipantEvent.LocalTrackPublished,
|
|
ParticipantEvent.LocalTrackUnpublished,
|
|
).pipe(map((p) => p.isScreenShareEnabled));
|
|
}
|
|
|
|
interface EnterRTCSessionOptions {
|
|
encryptMedia: boolean;
|
|
matrixRTCMode: MatrixRTCMode;
|
|
delayedLeaveTimings: ResolvedDelayedLeaveTimings;
|
|
/** Whether and what kind of notification to send when joining. */
|
|
sendNotificationType?: RTCNotificationType;
|
|
/** The kind of call being placed. */
|
|
callIntent?: RTCCallIntent;
|
|
}
|
|
|
|
/**
|
|
* Does the necessary steps to enter the RTC session on the matrix side:
|
|
* - Preparing the membership info (FOCUS to use, options)
|
|
* - Sends the matrix event to join the call, and starts the membership manager:
|
|
* - Delay events management
|
|
* - Handles retries (fails only after several attempts)
|
|
*
|
|
* @param rtcSession - The MatrixRTCSession to join.
|
|
* @param ownMembershipIdentity - Options for entering the RTC session.
|
|
* @param transport - The LivekitTransport to use for this session.
|
|
* @param options - `encryptMedia`: Whether to encrypt media. `matrixRTCMode`: The
|
|
* Matrix RTC mode to use. `delayedLeaveTimings`: The preferred timings for
|
|
* delayed leave events. `sendNotificationType`: Whether and what kind of
|
|
* notification to send on join. `callIntent`: The kind of call being placed.
|
|
* @throws If the host could not be told that we are joining.
|
|
*/
|
|
// Exported for unit testing
|
|
export function enterRTCSession(
|
|
rtcSession: MatrixRTCSession,
|
|
ownMembershipIdentity: CallMembershipIdentityParts,
|
|
transport: LivekitTransport,
|
|
{
|
|
encryptMedia,
|
|
matrixRTCMode,
|
|
delayedLeaveTimings,
|
|
sendNotificationType: notificationType,
|
|
callIntent,
|
|
}: EnterRTCSessionOptions,
|
|
): void {
|
|
PosthogAnalytics.instance.eventCallEnded.cacheStartCall(new Date());
|
|
PosthogAnalytics.instance.eventCallStarted.track(rtcSession.room.roomId);
|
|
|
|
// This must be called before we start trying to join the call, as we need to
|
|
// have started tracking by the time calls start getting created.
|
|
// groupCallOTelMembership?.onJoinCall();
|
|
|
|
const {
|
|
sync_disconnect_grace_period_ms: gracePeriod,
|
|
matrix_rtc_session: sessionConfig,
|
|
} = Config.get();
|
|
const retryInterval = sessionConfig.network_error_retry_ms;
|
|
|
|
// Set maximumNetworkErrorRetryCount such that we will consider the client
|
|
// disconnected as soon as either it fails to sync for longer than the grace
|
|
// period, or it is likely that a delayed leave event has been sent.
|
|
// Math.min is used to account for the respective worst case: /sync not available or leave event emitted.
|
|
const maxWaitTime = Math.min(gracePeriod, delayedLeaveTimings.delay_ms);
|
|
const maximumNetworkErrorRetryCount =
|
|
Math.ceil(maxWaitTime / retryInterval) + 1;
|
|
|
|
// Multi-sfu does not need a preferred foci list. just the focus that is actually used.
|
|
// TODO where/how do we track errors originating from the ongoing rtcSession?
|
|
|
|
rtcSession.join(ownMembershipIdentity, [transport], {
|
|
notificationType,
|
|
callIntent,
|
|
manageMediaKeys: encryptMedia,
|
|
delayedLeaveEventRestartMs: delayedLeaveTimings.restart_ms,
|
|
delayedLeaveEventDelayMs: delayedLeaveTimings.delay_ms,
|
|
delayedLeaveEventRestartLocalTimeoutMs:
|
|
delayedLeaveTimings.restart_timeout_ms,
|
|
networkErrorRetryMs: sessionConfig.network_error_retry_ms,
|
|
membershipEventExpiryMs: sessionConfig.membership_event_expiry_ms,
|
|
keyRotationParticipantLimit: sessionConfig.key_rotation_participant_limit,
|
|
unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0,
|
|
maximumNetworkErrorRetryCount: maximumNetworkErrorRetryCount,
|
|
});
|
|
}
|