mirror of
https://github.com/vector-im/element-call.git
synced 2026-10-16 23:45:50 +00:00
sdk implementation
This commit is contained in:
Timo K.
@@ -0,0 +1,133 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import {
|
||||
type BaseKeyProvider,
|
||||
Room as LivekitRoom,
|
||||
type RoomOptions,
|
||||
} from "livekit-client";
|
||||
// Inline so that the worker also loads when the SDK is served from another
|
||||
// origin than the page
|
||||
import E2EEWorker from "livekit-client/e2ee-worker?worker&inline";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
import { BehaviorSubject, combineLatest, map } from "rxjs";
|
||||
|
||||
import { buildLiveKitOptions } from "../../../src/livekit/options";
|
||||
import {
|
||||
type OpenIDClientParts,
|
||||
type SFUConfig,
|
||||
} from "../../../src/livekit/openIDSFU";
|
||||
import { type Behavior } from "../../../src/state/Behavior";
|
||||
import {
|
||||
Connection,
|
||||
type ConnectionOpts,
|
||||
ConnectionState,
|
||||
} from "../../../src/state/CallViewModel/remoteMembers/Connection";
|
||||
import { type ConnectionFactory } from "../../../src/state/CallViewModel/remoteMembers/ConnectionFactory";
|
||||
import { type ObservableScope } from "../../../src/state/ObservableScope";
|
||||
import { type LocalMediaInputs, type ResolvedTransport } from "../api";
|
||||
|
||||
/**
|
||||
* A connection that keeps what it fetched from the JWT service, so that the
|
||||
* transport can report it as `resolved$`. Nothing while the config is still
|
||||
* being fetched, and nothing again once the connection has stopped or failed.
|
||||
*/
|
||||
export class ResolvedConnection extends Connection {
|
||||
private readonly sfuConfig$: BehaviorSubject<SFUConfig | undefined>;
|
||||
public readonly resolved$: Behavior<ResolvedTransport | undefined>;
|
||||
|
||||
public constructor(opts: ConnectionOpts, logger: Logger) {
|
||||
super(opts, logger);
|
||||
this.sfuConfig$ = new BehaviorSubject(opts.existingSFUConfig);
|
||||
this.resolved$ = opts.scope.behavior(
|
||||
combineLatest([this.sfuConfig$, this.state$]).pipe(
|
||||
map(([config, state]) =>
|
||||
config === undefined ||
|
||||
state === ConnectionState.Stopped ||
|
||||
state instanceof Error
|
||||
? undefined
|
||||
: resolvedTransport(config),
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
protected override async getSFUConfigForRemoteConnection(): Promise<SFUConfig> {
|
||||
const config = await super.getSFUConfigForRemoteConnection();
|
||||
this.sfuConfig$.next(config);
|
||||
return config;
|
||||
}
|
||||
}
|
||||
|
||||
function resolvedTransport(config: SFUConfig): ResolvedTransport {
|
||||
return {
|
||||
type: "livekit",
|
||||
url: config.url,
|
||||
token: config.jwt,
|
||||
roomAlias: config.livekitAlias,
|
||||
identity: config.livekitIdentity,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a `ResolvedConnection` per transport, each with a LiveKit room of
|
||||
* its own that captures from the devices the host selected.
|
||||
*/
|
||||
export class LivekitConnectionFactory implements ConnectionFactory {
|
||||
public constructor(
|
||||
private readonly client: OpenIDClientParts,
|
||||
private readonly roomId: string,
|
||||
private readonly localMedia: LocalMediaInputs,
|
||||
private readonly keyProvider: BaseKeyProvider | undefined,
|
||||
) {}
|
||||
|
||||
public createConnection(
|
||||
scope: ObservableScope,
|
||||
transport: LivekitTransport,
|
||||
ownMembershipIdentity: CallMembershipIdentityParts,
|
||||
logger: Logger,
|
||||
sfuConfig?: SFUConfig,
|
||||
): Connection {
|
||||
return new ResolvedConnection(
|
||||
{
|
||||
existingSFUConfig: sfuConfig,
|
||||
roomId: this.roomId,
|
||||
transport,
|
||||
client: this.client,
|
||||
scope,
|
||||
livekitRoomFactory: () =>
|
||||
new LivekitRoom(roomOptions(this.localMedia, this.keyProvider)),
|
||||
ownMembershipIdentity,
|
||||
},
|
||||
logger,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function roomOptions(
|
||||
localMedia: LocalMediaInputs,
|
||||
keyProvider: BaseKeyProvider | undefined,
|
||||
): RoomOptions {
|
||||
const base = buildLiveKitOptions();
|
||||
return {
|
||||
...base,
|
||||
videoCaptureDefaults: {
|
||||
...base.videoCaptureDefaults,
|
||||
deviceId: localMedia.videoInputDeviceId$.value,
|
||||
processor: localMedia.videoProcessor$.value,
|
||||
},
|
||||
audioCaptureDefaults: {
|
||||
...base.audioCaptureDefaults,
|
||||
deviceId: localMedia.audioInputDeviceId$.value,
|
||||
},
|
||||
// Every room needs a worker of its own: one gets confused by streams from
|
||||
// several rooms
|
||||
e2ee: keyProvider && { keyProvider, worker: new E2EEWorker() },
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { type BaseKeyProvider, ExternalE2EEKeyProvider } from "livekit-client";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type MatrixRTCSession as JsSdkRtcSession } from "matrix-js-sdk/lib/matrixrtc";
|
||||
|
||||
import { E2eeType } from "../../../src/e2ee/e2eeType";
|
||||
import { MatrixKeyProvider } from "../../../src/e2ee/matrixKeyProvider";
|
||||
import { type EncryptionSystem } from "../../../src/e2ee/sharedKeyManagement";
|
||||
|
||||
/** The LiveKit key provider for the encryption the host asked for, or none. */
|
||||
export function createKeyProvider(
|
||||
encryptionSystem: EncryptionSystem,
|
||||
session: JsSdkRtcSession,
|
||||
logger: Logger,
|
||||
): BaseKeyProvider | undefined {
|
||||
switch (encryptionSystem.kind) {
|
||||
case E2eeType.NONE:
|
||||
return undefined;
|
||||
case E2eeType.PER_PARTICIPANT: {
|
||||
const keyProvider = new MatrixKeyProvider();
|
||||
keyProvider.setRTCSession(session);
|
||||
return keyProvider;
|
||||
}
|
||||
case E2eeType.SHARED_KEY: {
|
||||
const keyProvider = new ExternalE2EEKeyProvider();
|
||||
keyProvider
|
||||
.setKey(encryptionSystem.secret)
|
||||
.catch((e) => logger.error("Failed to set the shared E2EE key", e));
|
||||
return keyProvider;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,359 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import {
|
||||
type LocalParticipant,
|
||||
type ScreenShareCaptureOptions,
|
||||
} from "livekit-client";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type LivekitTransport,
|
||||
type MatrixRTCSession as JsSdkRtcSession,
|
||||
type Status as RTCSessionStatus,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { deepCompare } from "matrix-js-sdk/lib/utils";
|
||||
import {
|
||||
BehaviorSubject,
|
||||
catchError,
|
||||
combineLatest,
|
||||
concat,
|
||||
distinctUntilChanged,
|
||||
map,
|
||||
NEVER,
|
||||
type Observable,
|
||||
of,
|
||||
pairwise,
|
||||
race,
|
||||
Subject,
|
||||
switchMap,
|
||||
} from "rxjs";
|
||||
|
||||
import { type Behavior } from "../../../src/state/Behavior";
|
||||
import { type ObservableScope } from "../../../src/state/ObservableScope";
|
||||
import { type HomeserverConnected } from "../../../src/state/CallViewModel/localMember/HomeserverConnected";
|
||||
import { observeSharingScreen$ } from "../../../src/state/CallViewModel/localMember/LocalMember";
|
||||
import {
|
||||
type Connection,
|
||||
ConnectionState,
|
||||
} from "../../../src/state/CallViewModel/remoteMembers/Connection";
|
||||
import { type IConnectionManager } from "../../../src/state/CallViewModel/remoteMembers/ConnectionManager";
|
||||
import { type RtcSessionError } from "../api";
|
||||
import { toRtcSessionError } from "../utils/errors";
|
||||
import { type LocalTransport } from "./LocalTransport";
|
||||
import { type Publisher } from "./Publisher";
|
||||
|
||||
export enum TransportState {
|
||||
Waiting = "transport_waiting",
|
||||
}
|
||||
|
||||
export enum PublishState {
|
||||
WaitingForUser = "publish_waiting_for_user",
|
||||
Publishing = "publish_publishing",
|
||||
}
|
||||
|
||||
export type LocalMemberMediaState =
|
||||
| { connection: ConnectionState | RtcSessionError }
|
||||
| PublishState
|
||||
| RtcSessionError;
|
||||
|
||||
export type LocalMemberState =
|
||||
| RtcSessionError
|
||||
| TransportState.Waiting
|
||||
| {
|
||||
media: LocalMemberMediaState;
|
||||
matrix: RtcSessionError | RTCSessionStatus;
|
||||
};
|
||||
|
||||
interface Props {
|
||||
scope: ObservableScope;
|
||||
connectionManager: IConnectionManager;
|
||||
localTransport$: Observable<LocalTransport>;
|
||||
homeserverConnected: HomeserverConnected;
|
||||
createPublisher: (connection: Connection) => Publisher;
|
||||
joinMatrixRTC: (transport: LivekitTransport) => void;
|
||||
/** The membership manager giving up on keeping the membership alive. */
|
||||
membershipManagerError$: Observable<unknown>;
|
||||
matrixRTCSession: Pick<
|
||||
JsSdkRtcSession,
|
||||
"updateCallIntent" | "leaveRoomSession"
|
||||
>;
|
||||
cameraEnabled$: Behavior<boolean>;
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
export interface LocalMembership {
|
||||
requestJoinAndPublish: () => void;
|
||||
requestDisconnect: () => void;
|
||||
joinRequested$: Behavior<boolean>;
|
||||
state$: Behavior<LocalMemberState>;
|
||||
participant$: Behavior<LocalParticipant | null>;
|
||||
connection$: Behavior<Connection | null>;
|
||||
/** Fully connected: to the homeserver, the session and the transport. */
|
||||
connected$: Behavior<boolean>;
|
||||
/** Connected once, and currently not. */
|
||||
reconnecting$: Behavior<boolean>;
|
||||
sharingScreen$: Behavior<boolean>;
|
||||
toggleScreenSharing: (() => void) | null;
|
||||
screenShareError$: Behavior<Error | null>;
|
||||
dismissScreenShareError: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* The local member's state machine: waits for the transport, publishes on
|
||||
* its connection once asked to join, and enters and leaves the MatrixRTC
|
||||
* session in step.
|
||||
*/
|
||||
export function createLocalMembership$({
|
||||
scope,
|
||||
connectionManager,
|
||||
localTransport$: localTransportWithErrors$,
|
||||
homeserverConnected,
|
||||
createPublisher,
|
||||
joinMatrixRTC,
|
||||
membershipManagerError$,
|
||||
matrixRTCSession,
|
||||
cameraEnabled$,
|
||||
logger: parentLogger,
|
||||
}: Props): LocalMembership {
|
||||
const logger = parentLogger.getChild("[LocalMember]");
|
||||
|
||||
const fatalTransportError$ = new Subject<RtcSessionError>();
|
||||
const localTransport$ = localTransportWithErrors$.pipe(
|
||||
catchError((e: unknown) => {
|
||||
fatalTransportError$.next(toRtcSessionError(e));
|
||||
return NEVER;
|
||||
}),
|
||||
);
|
||||
const transport$ = scope.behavior<LocalTransport | null>(
|
||||
localTransport$,
|
||||
null,
|
||||
);
|
||||
|
||||
const connection$ = scope.behavior(
|
||||
combineLatest([
|
||||
connectionManager.connectionManagerData$,
|
||||
localTransport$,
|
||||
]).pipe(
|
||||
map(([{ value: connections }, { transport }]) =>
|
||||
connections.getConnectionForTransport(transport),
|
||||
),
|
||||
),
|
||||
null,
|
||||
);
|
||||
|
||||
const joinRequested$ = new BehaviorSubject(false);
|
||||
const publisher$ = new BehaviorSubject<Publisher | null>(null);
|
||||
const publishError$ = new BehaviorSubject<RtcSessionError | null>(null);
|
||||
const matrixError$ = new BehaviorSubject<RtcSessionError | null>(null);
|
||||
|
||||
scope.reconcile(connection$, async (connection) => {
|
||||
if (connection === null) return;
|
||||
const publisher = createPublisher(connection);
|
||||
publisher$.next(publisher);
|
||||
return Promise.resolve(async (): Promise<void> => {
|
||||
publisher$.next(null);
|
||||
await publisher.destroy();
|
||||
});
|
||||
});
|
||||
|
||||
scope.reconcile(
|
||||
scope.behavior(combineLatest([publisher$, joinRequested$])),
|
||||
async ([publisher, shouldPublish]) => {
|
||||
if (publisher === null) return;
|
||||
try {
|
||||
if (shouldPublish) {
|
||||
publisher.createAndSetupTracks();
|
||||
await publisher.startPublishing();
|
||||
} else if (publisher.shouldPublish) await publisher.stopPublishing();
|
||||
} catch (e) {
|
||||
if (publishError$.value === null)
|
||||
publishError$.next(toRtcSessionError(e));
|
||||
else logger.error("Another publish error", e);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
scope.reconcile(
|
||||
scope.behavior(combineLatest([transport$, joinRequested$])),
|
||||
async ([transport, shouldJoin]) => {
|
||||
if (transport === null || !shouldJoin) return;
|
||||
try {
|
||||
joinMatrixRTC(transport.transport);
|
||||
} catch (e) {
|
||||
logger.error("Failed to enter the session", e);
|
||||
if (matrixError$.value === null)
|
||||
matrixError$.next(toRtcSessionError(e));
|
||||
}
|
||||
return Promise.resolve(async (): Promise<void> => {
|
||||
try {
|
||||
await matrixRTCSession.leaveRoomSession(1000);
|
||||
} catch (e) {
|
||||
logger.error("Failed to leave the session", e);
|
||||
}
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
membershipManagerError$.pipe(scope.bind()).subscribe((e) => {
|
||||
logger.error("The membership manager stopped", e);
|
||||
if (matrixError$.value === null) matrixError$.next(toRtcSessionError(e));
|
||||
});
|
||||
|
||||
cameraEnabled$.pipe(scope.bind()).subscribe((videoEnabled) => {
|
||||
matrixRTCSession
|
||||
.updateCallIntent(videoEnabled ? "video" : "audio")
|
||||
.catch((e) => {
|
||||
// Expected before the join: the intent is sent with it instead
|
||||
if (e instanceof Error && e.message === "Not connected yet") return;
|
||||
logger.error("Failed to update the call intent", e);
|
||||
});
|
||||
});
|
||||
|
||||
const connectionState$ = connection$.pipe(
|
||||
switchMap((connection) => connection?.state$ ?? of(null)),
|
||||
);
|
||||
|
||||
const mediaState$ = scope.behavior<LocalMemberMediaState>(
|
||||
combineLatest([connectionState$, joinRequested$]).pipe(
|
||||
map(([connectionState, shouldPublish]) => {
|
||||
if (connectionState !== ConnectionState.LivekitConnected)
|
||||
return {
|
||||
connection:
|
||||
connectionState instanceof Error
|
||||
? toRtcSessionError(connectionState)
|
||||
: (connectionState ?? ConnectionState.Initialized),
|
||||
};
|
||||
return shouldPublish
|
||||
? PublishState.Publishing
|
||||
: PublishState.WaitingForUser;
|
||||
}),
|
||||
distinctUntilChanged(deepCompare),
|
||||
),
|
||||
);
|
||||
|
||||
const state$ = scope.behavior<LocalMemberState>(
|
||||
concat(
|
||||
of(TransportState.Waiting),
|
||||
race(
|
||||
fatalTransportError$,
|
||||
localTransport$.pipe(
|
||||
switchMap(() =>
|
||||
combineLatest(
|
||||
[
|
||||
mediaState$,
|
||||
homeserverConnected.rtsSession$,
|
||||
matrixError$,
|
||||
publishError$,
|
||||
],
|
||||
(media, sessionStatus, matrixError, publishError) => ({
|
||||
matrix: matrixError ?? sessionStatus,
|
||||
media: publishError ?? media,
|
||||
}),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
const connected$ = scope.behavior(
|
||||
combineLatest(
|
||||
[homeserverConnected.combined$, connectionState$],
|
||||
([homeserverConnected], connectionState) =>
|
||||
homeserverConnected &&
|
||||
connectionState === ConnectionState.LivekitConnected,
|
||||
),
|
||||
);
|
||||
|
||||
const reconnecting$ = scope.behavior(
|
||||
connected$.pipe(
|
||||
pairwise(),
|
||||
map(([was, is]) => was && !is),
|
||||
),
|
||||
false,
|
||||
);
|
||||
|
||||
const participant$ = scope.behavior(
|
||||
connection$.pipe(map((c) => c?.livekitRoom.localParticipant ?? null)),
|
||||
);
|
||||
|
||||
// Nothing leaves this device while it may already have been dropped from
|
||||
// the session: the member would show as away while still being heard
|
||||
combineLatest([participant$, homeserverConnected.combined$])
|
||||
.pipe(scope.bind())
|
||||
.subscribe(([participant, [connected]]) => {
|
||||
if (participant === null) return;
|
||||
for (const { track } of participant.trackPublications.values()) {
|
||||
if (!track) continue;
|
||||
if (connected && track.isUpstreamPaused)
|
||||
track.resumeUpstream().catch((e) => {
|
||||
logger.error(`Failed to resume the ${track.kind} track`, e);
|
||||
});
|
||||
else if (!connected && !track.isUpstreamPaused)
|
||||
track.pauseUpstream().catch((e) => {
|
||||
logger.error(`Failed to pause the ${track.kind} track`, e);
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
const sharingScreen$ = scope.behavior(
|
||||
participant$.pipe(
|
||||
switchMap((p) => (p === null ? of(false) : observeSharingScreen$(p))),
|
||||
),
|
||||
);
|
||||
const screenShareError$ = new BehaviorSubject<Error | null>(null);
|
||||
const toggleScreenSharing =
|
||||
"getDisplayMedia" in (navigator.mediaDevices ?? {})
|
||||
? (): void => {
|
||||
const participant = participant$.value;
|
||||
if (participant === null) return;
|
||||
const enable = !sharingScreen$.value;
|
||||
participant
|
||||
.setScreenShareEnabled(enable, screenShareCaptureOptions)
|
||||
.catch((e: unknown) => {
|
||||
logger.error(
|
||||
`Screen share ${enable ? "start" : "stop"} failed`,
|
||||
e,
|
||||
);
|
||||
// The user closing the picker is not an error worth showing
|
||||
if (e instanceof DOMException && e.name === "NotAllowedError")
|
||||
return;
|
||||
screenShareError$.next(
|
||||
e instanceof Error ? e : new Error(String(e)),
|
||||
);
|
||||
});
|
||||
}
|
||||
: null;
|
||||
|
||||
return {
|
||||
requestJoinAndPublish: () => joinRequested$.next(true),
|
||||
requestDisconnect: () => joinRequested$.next(false),
|
||||
joinRequested$,
|
||||
state$,
|
||||
participant$,
|
||||
connection$,
|
||||
connected$,
|
||||
reconnecting$,
|
||||
sharingScreen$,
|
||||
toggleScreenSharing,
|
||||
screenShareError$,
|
||||
dismissScreenShareError: () => screenShareError$.next(null),
|
||||
};
|
||||
}
|
||||
|
||||
const screenShareCaptureOptions: ScreenShareCaptureOptions = {
|
||||
// No echo cancellation: it would cancel the other members' voices out of
|
||||
// the shared audio
|
||||
audio: {
|
||||
autoGainControl: false,
|
||||
noiseSuppression: false,
|
||||
voiceIsolation: false,
|
||||
},
|
||||
selfBrowserSurface: "include",
|
||||
surfaceSwitching: "include",
|
||||
systemAudio: "include",
|
||||
};
|
||||
@@ -0,0 +1,66 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { type MatrixClient } from "matrix-js-sdk";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
|
||||
import {
|
||||
DEFAULT_CONFIG,
|
||||
type MatrixRTCMode,
|
||||
} from "../../../src/config/ConfigOptions";
|
||||
import { getSFUConfigWithOpenID } from "../../../src/livekit/openIDSFU";
|
||||
import { type LocalTransport } from "../../../src/state/CallViewModel/localMember/LocalTransport";
|
||||
import { RtcTransportAutoDiscovery } from "../../../src/state/CallViewModel/localMember/RtcTransportAutoDiscovery";
|
||||
import { MatrixRTCTransportMissingError } from "../../../src/utils/errors";
|
||||
|
||||
export type { LocalTransport };
|
||||
|
||||
interface Props {
|
||||
client: Pick<
|
||||
MatrixClient,
|
||||
| "getDomain"
|
||||
| "_unstable_getRTCTransports"
|
||||
| "getOpenIdToken"
|
||||
| "getDeviceId"
|
||||
>;
|
||||
ownMembershipIdentity: CallMembershipIdentityParts;
|
||||
roomId: string;
|
||||
matrixRTCMode: MatrixRTCMode;
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
/**
|
||||
* The transport the local member publishes on: the homeserver's preferred
|
||||
* one, authenticated with so that the session can be joined with a token in
|
||||
* hand. Only the homeserver is asked; the SDK has no configuration of its own
|
||||
* to fall back on.
|
||||
*/
|
||||
export async function getLocalTransport({
|
||||
client,
|
||||
ownMembershipIdentity,
|
||||
roomId,
|
||||
matrixRTCMode,
|
||||
logger,
|
||||
}: Props): Promise<LocalTransport> {
|
||||
const transport = await new RtcTransportAutoDiscovery({
|
||||
client,
|
||||
resolvedConfig: DEFAULT_CONFIG,
|
||||
logger,
|
||||
}).discoverPreferredTransport();
|
||||
if (transport === null)
|
||||
throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
|
||||
const sfuConfig = await getSFUConfigWithOpenID(
|
||||
client,
|
||||
ownMembershipIdentity,
|
||||
transport.livekit_service_url,
|
||||
roomId,
|
||||
{ matrixRTCMode },
|
||||
logger,
|
||||
);
|
||||
return { transport, sfuConfig };
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import {
|
||||
type LocalParticipant,
|
||||
type Participant,
|
||||
type Room as LivekitRoom,
|
||||
} from "livekit-client";
|
||||
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { combineLatest, distinctUntilChanged, map } from "rxjs";
|
||||
|
||||
import { type Behavior } from "../../../src/state/Behavior";
|
||||
import { type ObservableScope } from "../../../src/state/ObservableScope";
|
||||
import { type Connection } from "../../../src/state/CallViewModel/remoteMembers/Connection";
|
||||
import { type createMatrixMemberMetadata$ } from "../../../src/state/CallViewModel/remoteMembers/MatrixMemberMetadata";
|
||||
import { type RemoteMatrixLivekitMember } from "../../../src/state/CallViewModel/remoteMembers/MatrixLivekitMembers";
|
||||
import { type EncryptionSystem } from "../../../src/e2ee/sharedKeyManagement";
|
||||
import {
|
||||
type LocalRtcMember,
|
||||
type MemberMedia,
|
||||
type RemoteRtcMember,
|
||||
type RtcMember,
|
||||
} from "../api";
|
||||
import {
|
||||
createLivekitMemberMedia,
|
||||
createLocalLivekitMemberMedia,
|
||||
} from "../media/LivekitMemberMedia";
|
||||
import { mapScoped } from "../utils/mapScoped";
|
||||
import { type LocalMembership } from "./LocalMember";
|
||||
import { type TransportRegistry } from "./Transports";
|
||||
|
||||
export interface MemberContext {
|
||||
metadata: ReturnType<typeof createMatrixMemberMetadata$>;
|
||||
transports: TransportRegistry;
|
||||
encryptionSystem: EncryptionSystem;
|
||||
}
|
||||
|
||||
/** Everything that tells one membership from another, for keying items. */
|
||||
export function membershipKeys(
|
||||
membership: CallMembership,
|
||||
): [string, string, string, string] {
|
||||
return [
|
||||
membership.userId,
|
||||
membership.deviceId,
|
||||
membership.memberId,
|
||||
membership.rtcBackendIdentity,
|
||||
];
|
||||
}
|
||||
|
||||
export function createRemoteRtcMember(
|
||||
scope: ObservableScope,
|
||||
member: RemoteMatrixLivekitMember,
|
||||
context: MemberContext,
|
||||
): RemoteRtcMember {
|
||||
return {
|
||||
...createRtcMember(scope, member.membership$, context),
|
||||
local: false,
|
||||
media$: mediaFor(
|
||||
scope,
|
||||
member.participant.value$,
|
||||
member.connection$,
|
||||
(mediaScope, participant, room) =>
|
||||
createLivekitMemberMedia(
|
||||
mediaScope,
|
||||
participant,
|
||||
room,
|
||||
context.encryptionSystem,
|
||||
),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
export function createLocalRtcMember(
|
||||
scope: ObservableScope,
|
||||
membership$: Behavior<CallMembership>,
|
||||
localMembership: LocalMembership,
|
||||
context: MemberContext,
|
||||
): LocalRtcMember {
|
||||
return {
|
||||
...createRtcMember(scope, membership$, context),
|
||||
local: true,
|
||||
media$: mediaFor(
|
||||
scope,
|
||||
localMembership.participant$,
|
||||
localMembership.connection$,
|
||||
(mediaScope, participant, room) =>
|
||||
createLocalLivekitMemberMedia(
|
||||
mediaScope,
|
||||
participant,
|
||||
room,
|
||||
context.encryptionSystem,
|
||||
),
|
||||
),
|
||||
sharingScreen$: localMembership.sharingScreen$,
|
||||
toggleScreenSharing: localMembership.toggleScreenSharing,
|
||||
screenShareError$: localMembership.screenShareError$,
|
||||
dismissScreenShareError: localMembership.dismissScreenShareError,
|
||||
};
|
||||
}
|
||||
|
||||
function createRtcMember(
|
||||
scope: ObservableScope,
|
||||
membership$: Behavior<CallMembership>,
|
||||
{ metadata, transports }: MemberContext,
|
||||
): Omit<RtcMember, "local" | "media$"> {
|
||||
const { userId, deviceId, rtcBackendIdentity } = membership$.value;
|
||||
return {
|
||||
id: rtcBackendIdentity,
|
||||
userId,
|
||||
deviceId,
|
||||
membership$,
|
||||
displayName$: scope.behavior(
|
||||
metadata
|
||||
.createDisplayNameBehavior$(scope, userId)
|
||||
.pipe(map((name) => name ?? userId)),
|
||||
),
|
||||
avatarUrl$: metadata.createAvatarUrlBehavior$(scope, userId),
|
||||
transport$: scope.behavior(
|
||||
membership$.pipe(
|
||||
map((membership) => membership.getTransport()),
|
||||
distinctUntilChanged(),
|
||||
map((transport) => transport && transports.get(transport)),
|
||||
),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* The member's media once its participant is known, in a scope that ends when
|
||||
* the participant or the connection changes.
|
||||
*/
|
||||
function mediaFor<
|
||||
P extends Participant | LocalParticipant,
|
||||
M extends MemberMedia,
|
||||
>(
|
||||
scope: ObservableScope,
|
||||
participant$: Behavior<P | null>,
|
||||
connection$: Behavior<Connection | null>,
|
||||
factory: (scope: ObservableScope, participant: P, room: LivekitRoom) => M,
|
||||
): Behavior<M | null> {
|
||||
const source$ = scope.behavior(
|
||||
combineLatest([participant$, connection$]).pipe(
|
||||
map(([participant, connection]) =>
|
||||
participant && connection
|
||||
? { participant, room: connection.livekitRoom }
|
||||
: null,
|
||||
),
|
||||
distinctUntilChanged(
|
||||
(a, b) => a?.participant === b?.participant && a?.room === b?.room,
|
||||
),
|
||||
),
|
||||
);
|
||||
return scope.behavior(
|
||||
mapScoped(scope, source$, (mediaScope, { participant, room }) =>
|
||||
factory(mediaScope, participant, room),
|
||||
).pipe(map((media) => media ?? null)),
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,226 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { observeParticipantMedia } from "@livekit/components-core";
|
||||
import {
|
||||
ConnectionState as LivekitConnectionState,
|
||||
type LocalTrackPublication,
|
||||
LocalVideoTrack,
|
||||
ParticipantEvent,
|
||||
type Room as LivekitRoom,
|
||||
Track,
|
||||
} from "livekit-client";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { combineLatest, distinctUntilChanged, map, skip } from "rxjs";
|
||||
|
||||
import { type Behavior } from "../../../src/state/Behavior";
|
||||
import { ObservableScope } from "../../../src/state/ObservableScope";
|
||||
import { type LocalMediaInputs } from "../api";
|
||||
|
||||
/**
|
||||
* Publishes the local media on one LiveKit room, following `LocalMediaInputs`.
|
||||
*
|
||||
* LiveKit publishes a track the moment it is created, but a member must not be
|
||||
* heard before it has joined the MatrixRTC session, nor after it has left. The
|
||||
* tracks are therefore kept, and their upstream paused, while `shouldPublish`
|
||||
* is false: the local preview stays live while nothing reaches the room.
|
||||
*/
|
||||
export class Publisher {
|
||||
public shouldPublish = false;
|
||||
private tracksRequested = false;
|
||||
private readonly scope = new ObservableScope();
|
||||
private readonly room: LivekitRoom;
|
||||
|
||||
public constructor(
|
||||
room: LivekitRoom,
|
||||
private readonly inputs: LocalMediaInputs,
|
||||
private readonly logger: Logger,
|
||||
) {
|
||||
this.room = room;
|
||||
room.setE2EEEnabled(room.options.e2ee !== undefined)?.catch((e: Error) => {
|
||||
this.logger.error("Failed to enable E2EE on the room", e);
|
||||
});
|
||||
this.followVideoProcessor();
|
||||
this.followDevices();
|
||||
this.onLocalTrackPublished = this.onLocalTrackPublished.bind(this);
|
||||
room.localParticipant.on(
|
||||
ParticipantEvent.LocalTrackPublished,
|
||||
this.onLocalTrackPublished,
|
||||
);
|
||||
}
|
||||
|
||||
public async destroy(): Promise<void> {
|
||||
this.scope.end();
|
||||
this.room.localParticipant.off(
|
||||
ParticipantEvent.LocalTrackPublished,
|
||||
this.onLocalTrackPublished,
|
||||
);
|
||||
try {
|
||||
await this.stopTracks();
|
||||
} catch (e) {
|
||||
this.logger.error("Failed to stop the local tracks", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the microphone and camera tracks the inputs ask for, and keeps
|
||||
* them in step with the inputs from then on. Both are enabled in one call so
|
||||
* that the browser asks for permission once. Safe to call more than once.
|
||||
*/
|
||||
public createAndSetupTracks(): void {
|
||||
if (this.tracksRequested) return;
|
||||
this.tracksRequested = true;
|
||||
const participant = this.room.localParticipant;
|
||||
const audio = this.inputs.microphoneEnabled$.value;
|
||||
const video = this.inputs.cameraEnabled$.value;
|
||||
// LiveKit resolves these once the track is published, which may block on
|
||||
// the connection; LocalTrackPublished is what tells us a track exists.
|
||||
if (audio && video) void participant.enableCameraAndMicrophone();
|
||||
else if (audio) void participant.setMicrophoneEnabled(true);
|
||||
else if (video) void participant.setCameraEnabled(true);
|
||||
|
||||
this.follow(this.inputs.microphoneEnabled$, Track.Source.Microphone);
|
||||
this.follow(this.inputs.cameraEnabled$, Track.Source.Camera);
|
||||
}
|
||||
|
||||
public async startPublishing(): Promise<void> {
|
||||
if (this.shouldPublish) return;
|
||||
this.shouldPublish = true;
|
||||
// Enabling a track does not resume an upstream that was paused while it
|
||||
// was already enabled, so it is done explicitly
|
||||
await this.resumeUpstreams([Track.Source.Microphone, Track.Source.Camera]);
|
||||
}
|
||||
|
||||
public async stopPublishing(): Promise<void> {
|
||||
this.shouldPublish = false;
|
||||
await this.pauseUpstreams([
|
||||
Track.Source.Microphone,
|
||||
Track.Source.Camera,
|
||||
Track.Source.ScreenShare,
|
||||
]);
|
||||
}
|
||||
|
||||
private async stopTracks(): Promise<void> {
|
||||
const participant = this.room.localParticipant;
|
||||
for (const source of [
|
||||
Track.Source.Microphone,
|
||||
Track.Source.Camera,
|
||||
Track.Source.ScreenShare,
|
||||
]) {
|
||||
const track = participant.getTrackPublication(source)?.track;
|
||||
if (track) await participant.unpublishTrack(track, true);
|
||||
}
|
||||
}
|
||||
|
||||
private onLocalTrackPublished(publication: LocalTrackPublication): void {
|
||||
this.logger.info(`Local ${publication.source} track published`);
|
||||
if (!this.shouldPublish)
|
||||
this.pauseUpstreams([publication.source]).catch((e) => {
|
||||
this.logger.error("Failed to pause the upstream", e);
|
||||
});
|
||||
// The input may have changed while the track was being created
|
||||
const enabled =
|
||||
publication.source === Track.Source.Microphone
|
||||
? this.inputs.microphoneEnabled$.value
|
||||
: publication.source === Track.Source.Camera
|
||||
? this.inputs.cameraEnabled$.value
|
||||
: undefined;
|
||||
if (enabled === false) this.setEnabled(publication.source, false);
|
||||
}
|
||||
|
||||
private follow(
|
||||
enabled$: Behavior<boolean>,
|
||||
source: Track.Source.Microphone | Track.Source.Camera,
|
||||
): void {
|
||||
enabled$
|
||||
.pipe(skip(1), distinctUntilChanged(), this.scope.bind())
|
||||
.subscribe((enabled) => this.setEnabled(source, enabled));
|
||||
}
|
||||
|
||||
private setEnabled(source: Track.Source, enabled: boolean): void {
|
||||
const participant = this.room.localParticipant;
|
||||
const toggle =
|
||||
source === Track.Source.Microphone
|
||||
? participant.setMicrophoneEnabled(enabled)
|
||||
: participant.setCameraEnabled(enabled);
|
||||
toggle
|
||||
.then(async () => {
|
||||
// Unmuting restarts the upstream; until the member has joined, it
|
||||
// has to stay paused
|
||||
if (enabled && !this.shouldPublish) await this.pauseUpstreams([source]);
|
||||
})
|
||||
.catch((e) => {
|
||||
this.logger.error(`Failed to set ${source} enabled=${enabled}`, e);
|
||||
});
|
||||
}
|
||||
|
||||
private async pauseUpstreams(sources: Track.Source[]): Promise<void> {
|
||||
for (const source of sources) {
|
||||
const track =
|
||||
this.room.localParticipant.getTrackPublication(source)?.track;
|
||||
if (track && !track.isUpstreamPaused) await track.pauseUpstream();
|
||||
}
|
||||
}
|
||||
|
||||
private async resumeUpstreams(sources: Track.Source[]): Promise<void> {
|
||||
for (const source of sources) {
|
||||
const track =
|
||||
this.room.localParticipant.getTrackPublication(source)?.track;
|
||||
if (track?.isUpstreamPaused) await track.resumeUpstream();
|
||||
}
|
||||
}
|
||||
|
||||
private followDevices(): void {
|
||||
const sync = (
|
||||
kind: MediaDeviceKind,
|
||||
deviceId$: Behavior<string | undefined>,
|
||||
): void => {
|
||||
deviceId$.pipe(this.scope.bind()).subscribe((deviceId) => {
|
||||
if (
|
||||
deviceId === undefined ||
|
||||
this.room.state !== LivekitConnectionState.Connected ||
|
||||
this.room.getActiveDevice(kind) === deviceId
|
||||
)
|
||||
return;
|
||||
this.room
|
||||
.switchActiveDevice(kind, deviceId)
|
||||
.catch((e) => this.logger.error(`Failed to switch ${kind}`, e));
|
||||
});
|
||||
};
|
||||
sync("audioinput", this.inputs.audioInputDeviceId$);
|
||||
sync("videoinput", this.inputs.videoInputDeviceId$);
|
||||
}
|
||||
|
||||
private followVideoProcessor(): void {
|
||||
const participant = this.room.localParticipant;
|
||||
const cameraTrack$ = observeParticipantMedia(participant).pipe(
|
||||
map(() => {
|
||||
const track = participant.getTrackPublication(
|
||||
Track.Source.Camera,
|
||||
)?.track;
|
||||
return track instanceof LocalVideoTrack ? track : undefined;
|
||||
}),
|
||||
distinctUntilChanged(),
|
||||
);
|
||||
combineLatest([cameraTrack$, this.inputs.videoProcessor$])
|
||||
.pipe(this.scope.bind())
|
||||
.subscribe(([track, processor]) => {
|
||||
if (!track) return;
|
||||
if (processor && !track.getProcessor()) {
|
||||
// A processor cannot be built on a track that has already ended
|
||||
if (track.mediaStreamTrack.readyState === "ended") return;
|
||||
track.setProcessor(processor).catch((e) => {
|
||||
this.logger.warn("Failed to attach the video processor", e);
|
||||
});
|
||||
} else if (!processor && track.getProcessor()) {
|
||||
track.stopProcessor().catch((e) => {
|
||||
this.logger.warn("Failed to stop the video processor", e);
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,235 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { type MatrixClient, type Room } from "matrix-js-sdk";
|
||||
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
|
||||
import { MatrixRTCSessionEvent } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
import { v4 as uuidv4 } from "uuid";
|
||||
import { combineLatest, from, fromEvent, map } from "rxjs";
|
||||
|
||||
import { MatrixRTCMode } from "../../../src/config/ConfigOptions";
|
||||
import { type ObservableScope } from "../../../src/state/ObservableScope";
|
||||
import {
|
||||
createKeyRotationSuppressed$,
|
||||
createMemberships$,
|
||||
membershipsAndTransports$,
|
||||
} from "../../../src/state/SessionBehaviors";
|
||||
import { createHomeserverConnected$ } from "../../../src/state/CallViewModel/localMember/HomeserverConnected";
|
||||
import { createConnectionManager$ } from "../../../src/state/CallViewModel/remoteMembers/ConnectionManager";
|
||||
import { createRemoteMatrixLivekitMembers$ } from "../../../src/state/CallViewModel/remoteMembers/MatrixLivekitMembers";
|
||||
import {
|
||||
createMatrixMemberMetadata$,
|
||||
createRoomMembers$,
|
||||
} from "../../../src/state/CallViewModel/remoteMembers/MatrixMemberMetadata";
|
||||
import { filterBehavior, generateItems } from "../../../src/utils/observable";
|
||||
import {
|
||||
type LocalMediaInputs,
|
||||
type LocalRtcMember,
|
||||
type RemoteRtcMember,
|
||||
type RtcSession,
|
||||
RtcSessionError,
|
||||
type RtcSessionOptions,
|
||||
} from "../api";
|
||||
import { LivekitConnectionFactory } from "./ConnectionFactory";
|
||||
import { joinRtcSession, sessionTimings } from "./joinRtcSession";
|
||||
import { createKeyProvider } from "./KeyProvider";
|
||||
import { createLocalMembership$ } from "./LocalMember";
|
||||
import { getLocalTransport } from "./LocalTransport";
|
||||
import {
|
||||
createLocalRtcMember,
|
||||
createRemoteRtcMember,
|
||||
membershipKeys,
|
||||
} from "./Members";
|
||||
import { Publisher } from "./Publisher";
|
||||
import { mapScoped } from "../utils/mapScoped";
|
||||
import { fatalError, sessionStatus } from "./status";
|
||||
import { createTransportRegistry } from "./Transports";
|
||||
|
||||
/**
|
||||
* Takes the whole `MatrixClient` rather than a slice of it, and finds the
|
||||
* MatrixRTC session itself. The js-sdk is the MatrixRTC implementation today;
|
||||
* when the rust-rtc crate replaces it, the SDK has to bridge the crate to the
|
||||
* js-sdk client, and only the SDK knows what that bridge needs. Holding the
|
||||
* client keeps that change inside the SDK.
|
||||
*/
|
||||
export function createRtcSession(
|
||||
scope: ObservableScope,
|
||||
client: MatrixClient,
|
||||
room: Room,
|
||||
localMedia: LocalMediaInputs,
|
||||
options: RtcSessionOptions,
|
||||
): RtcSession {
|
||||
const logger = rootLogger.getChild("[RtcSession]");
|
||||
const userId = client.getUserId();
|
||||
const deviceId = client.getDeviceId();
|
||||
if (!(userId && deviceId))
|
||||
throw new RtcSessionError("The client has to be logged in");
|
||||
const { encryptionSystem, matrixRTCMode } = options;
|
||||
|
||||
const jsSdkSession = client.matrixRTC.getRoomSession(room);
|
||||
const keyProvider = createKeyProvider(encryptionSystem, jsSdkSession, logger);
|
||||
|
||||
const ownMembershipIdentity: CallMembershipIdentityParts = {
|
||||
userId,
|
||||
deviceId,
|
||||
// A pre-sticky membership names itself `${userId}:${deviceId}`, and the
|
||||
// key transport stamps this id into every key event, so a uuid there
|
||||
// would name a member no peer can resolve
|
||||
memberId:
|
||||
matrixRTCMode === MatrixRTCMode.Matrix_2_0
|
||||
? uuidv4()
|
||||
: `${userId}:${deviceId}`,
|
||||
};
|
||||
|
||||
const memberships$ = createMemberships$(scope, jsSdkSession);
|
||||
const { membershipsWithTransport$, transports$ } = membershipsAndTransports$(
|
||||
scope,
|
||||
memberships$,
|
||||
);
|
||||
|
||||
const localTransport$ = from(
|
||||
getLocalTransport({
|
||||
client,
|
||||
ownMembershipIdentity,
|
||||
roomId: room.roomId,
|
||||
matrixRTCMode,
|
||||
logger,
|
||||
}),
|
||||
);
|
||||
|
||||
const connectionManager = createConnectionManager$({
|
||||
scope,
|
||||
connectionFactory: new LivekitConnectionFactory(
|
||||
client,
|
||||
room.roomId,
|
||||
localMedia,
|
||||
keyProvider,
|
||||
),
|
||||
localTransport$,
|
||||
remoteTransports$: transports$,
|
||||
logger,
|
||||
ownMembershipIdentity,
|
||||
});
|
||||
const transports = createTransportRegistry(scope, connectionManager);
|
||||
|
||||
const localMembership = createLocalMembership$({
|
||||
scope,
|
||||
connectionManager,
|
||||
localTransport$,
|
||||
homeserverConnected: createHomeserverConnected$(
|
||||
scope,
|
||||
client,
|
||||
jsSdkSession,
|
||||
sessionTimings.syncDisconnectGracePeriodMs,
|
||||
),
|
||||
createPublisher: (connection) =>
|
||||
new Publisher(
|
||||
connection.livekitRoom,
|
||||
localMedia,
|
||||
logger.getChild(
|
||||
`[Publisher ${connection.transport.livekit_service_url}]`,
|
||||
),
|
||||
),
|
||||
joinMatrixRTC: (transport) =>
|
||||
joinRtcSession(jsSdkSession, ownMembershipIdentity, transport, {
|
||||
encryptMedia: keyProvider !== undefined,
|
||||
matrixRTCMode,
|
||||
sendNotificationType: options.sendNotificationType,
|
||||
callIntent: options.callIntent,
|
||||
}),
|
||||
membershipManagerError$: fromEvent(
|
||||
jsSdkSession,
|
||||
MatrixRTCSessionEvent.MembershipManagerError,
|
||||
),
|
||||
matrixRTCSession: jsSdkSession,
|
||||
cameraEnabled$: localMedia.cameraEnabled$,
|
||||
logger,
|
||||
});
|
||||
|
||||
const context = {
|
||||
metadata: createMatrixMemberMetadata$(
|
||||
scope,
|
||||
scope.behavior(memberships$.pipe(map(({ value }) => value))),
|
||||
createRoomMembers$(scope, room),
|
||||
),
|
||||
transports,
|
||||
encryptionSystem,
|
||||
};
|
||||
|
||||
const remoteMembers$ = scope.behavior<RemoteRtcMember[]>(
|
||||
createRemoteMatrixLivekitMembers$({
|
||||
scope,
|
||||
membershipsWithTransport$,
|
||||
connectionManager,
|
||||
localUser: { userId, deviceId },
|
||||
}).pipe(
|
||||
map(({ value }) => value),
|
||||
generateItems(
|
||||
"RtcSession remoteMembers",
|
||||
function* (members) {
|
||||
for (const member of members)
|
||||
yield {
|
||||
keys: membershipKeys(member.membership$.value),
|
||||
data: member,
|
||||
};
|
||||
},
|
||||
(memberScope, member$) =>
|
||||
createRemoteRtcMember(memberScope, member$.value, context),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
const localMembership$ = scope.behavior(
|
||||
memberships$.pipe(
|
||||
map(
|
||||
({ value }) =>
|
||||
value.find(
|
||||
(membership) =>
|
||||
membership.userId === userId && membership.deviceId === deviceId,
|
||||
) ?? null,
|
||||
),
|
||||
filterBehavior((membership) => membership !== null),
|
||||
),
|
||||
);
|
||||
const localMember$ = scope.behavior<LocalRtcMember | null>(
|
||||
mapScoped(scope, localMembership$, (memberScope, membership$) =>
|
||||
createLocalRtcMember(memberScope, membership$, localMembership, context),
|
||||
).pipe(map((member) => member ?? null)),
|
||||
);
|
||||
|
||||
const status$ = scope.behavior(
|
||||
combineLatest(
|
||||
[
|
||||
localMembership.state$,
|
||||
localMembership.joinRequested$,
|
||||
localMembership.connected$,
|
||||
localMembership.reconnecting$,
|
||||
],
|
||||
sessionStatus,
|
||||
),
|
||||
);
|
||||
|
||||
return {
|
||||
join: localMembership.requestJoinAndPublish,
|
||||
leave: localMembership.requestDisconnect,
|
||||
status$,
|
||||
connected$: localMembership.connected$,
|
||||
reconnecting$: localMembership.reconnecting$,
|
||||
fatalError$: scope.behavior(localMembership.state$.pipe(map(fatalError))),
|
||||
localMember$,
|
||||
remoteMembers$,
|
||||
memberCount$: scope.behavior(
|
||||
combineLatest(
|
||||
[localMember$, remoteMembers$],
|
||||
(local, remote) => remote.length + (local === null ? 0 : 1),
|
||||
),
|
||||
),
|
||||
keyRotationSuppressed$: createKeyRotationSuppressed$(scope, jsSdkSession),
|
||||
connectedTransports$: transports.connected$,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import {
|
||||
isLivekitTransport,
|
||||
type Transport,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { combineLatest, distinctUntilChanged, map, of, switchMap } from "rxjs";
|
||||
|
||||
import { constant, type Behavior } from "../../../src/state/Behavior";
|
||||
import { type ObservableScope } from "../../../src/state/ObservableScope";
|
||||
import { ConnectionState } from "../../../src/state/CallViewModel/remoteMembers/Connection";
|
||||
import { type IConnectionManager } from "../../../src/state/CallViewModel/remoteMembers/ConnectionManager";
|
||||
import { type TransportMetadata } from "../api";
|
||||
import { ResolvedConnection } from "./ConnectionFactory";
|
||||
|
||||
export interface TransportRegistry {
|
||||
/** The one `TransportMetadata` for a transport, however many memberships name it. */
|
||||
get(raw: Transport): TransportMetadata;
|
||||
/** Transports with a live connection. */
|
||||
connected$: Behavior<TransportMetadata[]>;
|
||||
}
|
||||
|
||||
export function createTransportRegistry(
|
||||
scope: ObservableScope,
|
||||
connectionManager: IConnectionManager,
|
||||
): TransportRegistry {
|
||||
const transports = new Map<string, TransportMetadata>();
|
||||
|
||||
const get = (raw: Transport): TransportMetadata => {
|
||||
const id = transportId(raw);
|
||||
let transport = transports.get(id);
|
||||
if (transport === undefined) {
|
||||
transport = createTransportMetadata(scope, connectionManager, raw, id);
|
||||
transports.set(id, transport);
|
||||
}
|
||||
return transport;
|
||||
};
|
||||
|
||||
const connected$ = scope.behavior(
|
||||
connectionManager.connectionManagerData$.pipe(
|
||||
switchMap(({ value }) => {
|
||||
const connections = value.getConnections();
|
||||
if (connections.length === 0) return of([]);
|
||||
return combineLatest(
|
||||
connections.map((connection) =>
|
||||
connection.state$.pipe(
|
||||
map((state) =>
|
||||
state === ConnectionState.LivekitConnected
|
||||
? get(connection.transport)
|
||||
: null,
|
||||
),
|
||||
),
|
||||
),
|
||||
).pipe(map((transports) => transports.filter((t) => t !== null)));
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
return { get, connected$ };
|
||||
}
|
||||
|
||||
function createTransportMetadata(
|
||||
scope: ObservableScope,
|
||||
connectionManager: IConnectionManager,
|
||||
raw: Transport,
|
||||
id: string,
|
||||
): TransportMetadata {
|
||||
const resolved$ = isLivekitTransport(raw)
|
||||
? scope.behavior(
|
||||
connectionManager.connectionManagerData$.pipe(
|
||||
map(({ value }) => value.getConnectionForTransport(raw)),
|
||||
distinctUntilChanged(),
|
||||
switchMap((connection) =>
|
||||
connection instanceof ResolvedConnection
|
||||
? connection.resolved$
|
||||
: of(undefined),
|
||||
),
|
||||
),
|
||||
)
|
||||
: constant(undefined);
|
||||
return { type: raw.type, id, raw, resolved$ };
|
||||
}
|
||||
|
||||
function transportId(raw: Transport): string {
|
||||
return isLivekitTransport(raw)
|
||||
? raw.livekit_service_url
|
||||
: JSON.stringify(raw);
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations 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,
|
||||
type MatrixRTCSession as JsSdkRtcSession,
|
||||
type RTCCallIntent,
|
||||
type RTCNotificationType,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
|
||||
import {
|
||||
DEFAULT_CONFIG,
|
||||
MatrixRTCMode,
|
||||
} from "../../../src/config/ConfigOptions";
|
||||
|
||||
interface Options {
|
||||
encryptMedia: boolean;
|
||||
matrixRTCMode: MatrixRTCMode;
|
||||
sendNotificationType?: RTCNotificationType;
|
||||
callIntent?: RTCCallIntent;
|
||||
}
|
||||
|
||||
/** The session's timing defaults; a later slice makes them an input. */
|
||||
export const sessionTimings = {
|
||||
syncDisconnectGracePeriodMs: DEFAULT_CONFIG.sync_disconnect_grace_period_ms,
|
||||
...DEFAULT_CONFIG.matrix_rtc_session,
|
||||
};
|
||||
|
||||
/**
|
||||
* Sends the membership and starts the membership manager, which keeps it
|
||||
* alive and retries on failure.
|
||||
* @throws If the join could not be started.
|
||||
*/
|
||||
export function joinRtcSession(
|
||||
session: JsSdkRtcSession,
|
||||
ownMembershipIdentity: CallMembershipIdentityParts,
|
||||
transport: LivekitTransport,
|
||||
{ encryptMedia, matrixRTCMode, sendNotificationType, callIntent }: Options,
|
||||
): void {
|
||||
const timings = sessionTimings.delayed_leave;
|
||||
// Give up on the network as soon as either the sync has been down for the
|
||||
// grace period or the delayed leave has probably been sent, whichever is
|
||||
// sooner
|
||||
const maxWaitMs = Math.min(
|
||||
sessionTimings.syncDisconnectGracePeriodMs,
|
||||
timings.delay_ms,
|
||||
);
|
||||
session.joinRTCSession(ownMembershipIdentity, [transport], {
|
||||
notificationType: sendNotificationType,
|
||||
callIntent,
|
||||
manageMediaKeys: encryptMedia,
|
||||
delayedLeaveEventRestartMs: timings.restart_ms,
|
||||
delayedLeaveEventDelayMs: timings.delay_ms,
|
||||
delayedLeaveEventRestartLocalTimeoutMs: timings.restart_timeout_ms,
|
||||
networkErrorRetryMs: sessionTimings.network_error_retry_ms,
|
||||
makeKeyDelay: sessionTimings.wait_for_key_rotation_ms,
|
||||
membershipEventExpiryMs: sessionTimings.membership_event_expiry_ms,
|
||||
keyRotationParticipantLimit: sessionTimings.key_rotation_participant_limit,
|
||||
unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0,
|
||||
maximumNetworkErrorRetryCount:
|
||||
Math.ceil(maxWaitMs / sessionTimings.network_error_retry_ms) + 1,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { Status } from "matrix-js-sdk/lib/matrixrtc";
|
||||
|
||||
import { ConnectionState } from "../../../src/state/CallViewModel/remoteMembers/Connection";
|
||||
import { RtcSessionError } from "../api";
|
||||
import {
|
||||
type LocalMemberState,
|
||||
PublishState,
|
||||
TransportState,
|
||||
} from "./LocalMember";
|
||||
import { fatalError, sessionStatus } from "./status";
|
||||
|
||||
const publishing: LocalMemberState = {
|
||||
media: PublishState.Publishing,
|
||||
matrix: Status.Connected,
|
||||
};
|
||||
const connecting: LocalMemberState = {
|
||||
media: { connection: ConnectionState.LivekitConnecting },
|
||||
matrix: Status.Connecting,
|
||||
};
|
||||
const error = new RtcSessionError("gone");
|
||||
|
||||
describe("sessionStatus", () => {
|
||||
it.each<[string, LocalMemberState, boolean, boolean, boolean, string]>([
|
||||
[
|
||||
"no transport yet",
|
||||
TransportState.Waiting,
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
"waitingForTransport",
|
||||
],
|
||||
["not asked to join", connecting, false, false, false, "disconnected"],
|
||||
["joining", connecting, true, false, false, "connecting"],
|
||||
["connected", publishing, true, true, false, "connected"],
|
||||
["dropped after connecting", connecting, true, false, true, "reconnecting"],
|
||||
["a transport error", error, true, false, false, "disconnected"],
|
||||
[
|
||||
"a matrix error",
|
||||
{ ...publishing, matrix: error },
|
||||
true,
|
||||
true,
|
||||
false,
|
||||
"disconnected",
|
||||
],
|
||||
])("%s", (_name, state, joinRequested, connected, reconnecting, expected) => {
|
||||
expect(sessionStatus(state, joinRequested, connected, reconnecting)).toBe(
|
||||
expected,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("fatalError", () => {
|
||||
it("is null while nothing is wrong", () => {
|
||||
expect(fatalError(TransportState.Waiting)).toBeNull();
|
||||
expect(fatalError(publishing)).toBeNull();
|
||||
});
|
||||
|
||||
it("finds the error wherever the state holds it", () => {
|
||||
expect(fatalError(error)).toBe(error);
|
||||
expect(fatalError({ ...publishing, matrix: error })).toBe(error);
|
||||
expect(
|
||||
fatalError({ media: { connection: error }, matrix: Status.Connected }),
|
||||
).toBe(error);
|
||||
});
|
||||
|
||||
it("does not treat a publish error as fatal", () => {
|
||||
expect(fatalError({ media: error, matrix: Status.Connected })).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,37 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { type RtcSessionError, type SessionConnectionStatus } from "../api";
|
||||
import { type LocalMemberState, TransportState } from "./LocalMember";
|
||||
|
||||
/** The local member's state, collapsed to what a host shows. */
|
||||
export function sessionStatus(
|
||||
state: LocalMemberState,
|
||||
joinRequested: boolean,
|
||||
connected: boolean,
|
||||
reconnecting: boolean,
|
||||
): SessionConnectionStatus {
|
||||
if (fatalError(state) !== null) return "disconnected";
|
||||
if (state === TransportState.Waiting) return "waitingForTransport";
|
||||
if (connected) return "connected";
|
||||
if (reconnecting) return "reconnecting";
|
||||
return joinRequested ? "connecting" : "disconnected";
|
||||
}
|
||||
|
||||
/** The error that stops the session, if the state holds one. */
|
||||
export function fatalError(state: LocalMemberState): RtcSessionError | null {
|
||||
if (state === TransportState.Waiting) return null;
|
||||
if (state instanceof Error) return state;
|
||||
if (state.matrix instanceof Error) return state.matrix;
|
||||
if (
|
||||
typeof state.media === "object" &&
|
||||
"connection" in state.media &&
|
||||
state.media.connection instanceof Error
|
||||
)
|
||||
return state.media.connection;
|
||||
return null;
|
||||
}
|
||||
Reference in New Issue
Block a user