use updated state logic in the sdk

This commit is contained in:
Timo K.
2026-10-14 10:23:29 +02:00
committed by Timo
parent cec5031f62
commit 598ec42390
5 changed files with 69 additions and 70 deletions
+6 -3
View File
@@ -147,8 +147,8 @@ export async function createMatrixRTCSdk(
vm.leave$ vm.leave$
.pipe( .pipe(
switchMap(() => switchMap(() =>
participation.status$.pipe( participation.state$.pipe(
filter((status) => status === "left"), filter((state) => state.kind === "left"),
take(1), take(1),
), ),
), ),
@@ -164,7 +164,10 @@ export async function createMatrixRTCSdk(
}, },
leave: (): void => vm?.leave(), leave: (): void => vm?.leave(),
stop: (): void => scope.end(), stop: (): void => scope.end(),
connected$: over((p) => p.connected$, false), connected$: over(
(p) => p.state$.pipe(map((state) => state.kind === "connected")),
false,
),
localMember$: over((p) => p.localMember$, null), localMember$: over((p) => p.localMember$, null),
remoteMembers$: over((p) => p.remoteMembers$, []), remoteMembers$: over((p) => p.remoteMembers$, []),
// Attested members only: the SDK has matched the sender // Attested members only: the SDK has matched the sender
@@ -261,7 +261,9 @@ describe.each(modes)("CallViewModel (%s mode)", (mode) => {
test("the selected audio output is handed to the client as it changes", () => { test("the selected audio output is handed to the client as it changes", () => {
const { mediaDevices, selected$ } = withAudioOutput(); const { mediaDevices, selected$ } = withAudioOutput();
withCallViewModel({ mediaDevices }, (_vm, _session, { rtcParticipation: participation }) => { withCallViewModel(
{ mediaDevices },
(_vm, _session, { rtcParticipation: participation }) => {
// Nothing selected yet, so the browser's choice stands // Nothing selected yet, so the browser's choice stands
expect(participation.setAudioOutputDeviceId).not.toHaveBeenCalled(); expect(participation.setAudioOutputDeviceId).not.toHaveBeenCalled();
selected$.next({ id: "speaker", virtualEarpiece: false }); selected$.next({ id: "speaker", virtualEarpiece: false });
@@ -273,7 +275,8 @@ describe.each(modes)("CallViewModel (%s mode)", (mode) => {
"headphones", "headphones",
); );
expect(participation.setAudioOutputDeviceId).toHaveBeenCalledTimes(2); expect(participation.setAudioOutputDeviceId).toHaveBeenCalledTimes(2);
}); },
);
}); });
test("a host that controls the audio devices keeps the output from the client", () => { test("a host that controls the audio devices keeps the output from the client", () => {
+17 -19
View File
@@ -295,7 +295,7 @@ export interface CallViewModel {
/** Call to initiate hangup. Use in conbination with reconnection state track the async hangup process. */ /** Call to initiate hangup. Use in conbination with reconnection state track the async hangup process. */
hangup: () => void; hangup: () => void;
/** Leaves the participation. The async leave can then be observed via connected$. */ /** Leaves the participation. The leave can then be observed via `leave$`. */
leave: () => void; leave: () => void;
// screen sharing // screen sharing
/** /**
@@ -442,9 +442,6 @@ export interface CallViewModel {
/** Connected once, and currently not: to the homeserver, the session or the media transport. */ /** Connected once, and currently not: to the homeserver, the session or the media transport. */
reconnecting$: Behavior<boolean>; reconnecting$: Behavior<boolean>;
/** Connected to the homeserver, the session and the media transport, all three. */
connected$: Behavior<boolean>;
} }
/** /**
@@ -627,8 +624,12 @@ export function createCallViewModel$(
localUser: { userId, deviceId }, localUser: { userId, deviceId },
}); });
const reconnecting$ = scope.behavior(
rtcParticipation.state$.pipe(map((state) => state.kind === "reconnecting")),
);
const handsRaised$ = scope.behavior( const handsRaised$ = scope.behavior(
handsRaisedSubject$.pipe(pauseWhen(rtcParticipation.reconnecting$)), handsRaisedSubject$.pipe(pauseWhen(reconnecting$)),
); );
const reactions$ = scope.behavior( const reactions$ = scope.behavior(
@@ -641,7 +642,7 @@ export function createCallViewModel$(
]), ]),
), ),
), ),
pauseWhen(rtcParticipation.reconnecting$), pauseWhen(reconnecting$),
), ),
); );
@@ -677,7 +678,7 @@ export function createCallViewModel$(
member.transport$.pipe(map((transport) => transport?.id)), member.transport$.pipe(map((transport) => transport?.id)),
), ),
mediaDevices, mediaDevices,
pretendToBeDisconnected$: rtcParticipation.reconnecting$, pretendToBeDisconnected$: reconnecting$,
displayName$: member.displayName$, displayName$: member.displayName$,
mxcAvatarUrl$: member.avatarUrl$, mxcAvatarUrl$: member.avatarUrl$,
handRaised$: scope.behavior( handRaised$: scope.behavior(
@@ -1509,14 +1510,10 @@ export function createCallViewModel$(
}; };
let reconnectStart: { time: number; reason: DisconnectReason } | null = null; let reconnectStart: { time: number; reason: DisconnectReason } | null = null;
rtcParticipation.disconnectReason$ rtcParticipation.state$.pipe(scope.bind()).subscribe((state) => {
.pipe(distinctUntilChanged(), pairwise(), scope.bind()) if (state.kind === "reconnecting") {
.subscribe(([prev, reason]) => { reconnectStart ??= { time: Date.now(), reason: state.reason };
if (reason !== null) { } else if (state.kind === "connected" && reconnectStart !== null) {
// Only the loss of a connection that existed counts as a reconnect,
// not the startup phase
if (prev === null) reconnectStart ??= { time: Date.now(), reason };
} else if (reconnectStart !== null) {
const reason = const reason =
reconnectStart.reason === "media" ? "livekit" : reconnectStart.reason; reconnectStart.reason === "media" ? "livekit" : reconnectStart.reason;
PosthogAnalytics.instance.eventCallReconnecting.track( PosthogAnalytics.instance.eventCallReconnecting.track(
@@ -1545,8 +1542,10 @@ export function createCallViewModel$(
unhoverScreen: (): void => screenUnhover$.next(), unhoverScreen: (): void => screenUnhover$.next(),
fatalError$: scope.behavior( fatalError$: scope.behavior(
rtcParticipation.fatalError$.pipe( rtcParticipation.state$.pipe(
map((error) => error && fromMatrixRTCError(error)), map((state) =>
state.kind === "failed" ? fromMatrixRTCError(state.error) : null,
),
), ),
), ),
participantCount$: participantCount$, participantCount$: participantCount$,
@@ -1582,8 +1581,7 @@ export function createCallViewModel$(
overflowing$, overflowing$,
earpieceMode$: earpieceMode$, earpieceMode$: earpieceMode$,
audioOutputSwitcher$: audioOutputSwitcher$, audioOutputSwitcher$: audioOutputSwitcher$,
reconnecting$: rtcParticipation.reconnecting$, reconnecting$,
connected$: rtcParticipation.connected$,
screenShareError$, screenShareError$,
dismissScreenShareError: (): void => screenShareError$.next(null), dismissScreenShareError: (): void => screenShareError$.next(null),
}; };
@@ -11,13 +11,14 @@ import {
constant, constant,
E2eeType, E2eeType,
type LocalRTCMember, type LocalRTCMember,
type ParticipationState,
type RTCParticipation, type RTCParticipation,
type MatrixRTCMode, type MatrixRTCMode,
type ObservableScope, type ObservableScope,
type RemoteRTCMember, type RemoteRTCMember,
} from "@element-hq/matrixrtc-sdk"; } from "@element-hq/matrixrtc-sdk";
import { SyncState } from "matrix-js-sdk/lib/sync"; import { SyncState } from "matrix-js-sdk/lib/sync";
import { BehaviorSubject, combineLatest, map, of, switchMap } from "rxjs"; import { BehaviorSubject, map, of, switchMap } from "rxjs";
import { ClientEvent, type RoomMember, type MatrixClient } from "matrix-js-sdk"; import { ClientEvent, type RoomMember, type MatrixClient } from "matrix-js-sdk";
import EventEmitter from "events"; import EventEmitter from "events";
@@ -244,11 +245,13 @@ export function withCallViewModel(mode: MatrixRTCMode) {
const rtcParticipation = mockRTCParticipation(scope, { const rtcParticipation = mockRTCParticipation(scope, {
localMember$, localMember$,
remoteMembers$, remoteMembers$,
connected$, state$: scope.behavior(
reconnecting$: scope.behavior( connected$.pipe(
combineLatest([connected$, of(initialSyncState)]).pipe( map((connected): ParticipationState =>
switchMap(() => connected$), connected
map((connected) => !connected), ? { kind: "connected" }
: { kind: "reconnecting", reason: "media" },
),
), ),
), ),
}); });
@@ -279,6 +282,11 @@ export function withCallViewModel(mode: MatrixRTCMode) {
); );
void mode; void mode;
continuation(vm, rtcSession, { raisedHands$, rtcParticipation }, setSyncState); continuation(
vm,
rtcSession,
{ raisedHands$, rtcParticipation },
setSyncState,
);
}; };
} }
+4 -17
View File
@@ -63,7 +63,6 @@ import {
type AudioMediaTrack, type AudioMediaTrack,
type Behavior, type Behavior,
constant, constant,
type DisconnectReason,
type EncryptionError, type EncryptionError,
type LocalAudioMediaTrack, type LocalAudioMediaTrack,
type LocalRTCMember, type LocalRTCMember,
@@ -71,7 +70,7 @@ import {
type RTCMember, type RTCMember,
type RTCParticipation, type RTCParticipation,
type RTCSlot, type RTCSlot,
type MatrixRTCError, type ParticipationState,
type PublishRequest, type PublishRequest,
type RemoteRTCMember, type RemoteRTCMember,
type TransportMetadata, type TransportMetadata,
@@ -494,10 +493,7 @@ export function mockRTCMember(
export interface MockParticipationInputs { export interface MockParticipationInputs {
localMember$?: Behavior<LocalRTCMember | null>; localMember$?: Behavior<LocalRTCMember | null>;
remoteMembers$?: Behavior<RemoteRTCMember[]>; remoteMembers$?: Behavior<RemoteRTCMember[]>;
connected$?: Behavior<boolean>; state$?: Behavior<ParticipationState>;
reconnecting$?: Behavior<boolean>;
disconnectReason$?: Behavior<DisconnectReason | null>;
fatalError$?: Behavior<MatrixRTCError | null>;
keyRotationSuppressed$?: Behavior<boolean>; keyRotationSuppressed$?: Behavior<boolean>;
connectedTransports$?: Behavior<TransportMetadata[]>; connectedTransports$?: Behavior<TransportMetadata[]>;
} }
@@ -511,10 +507,7 @@ export function mockRTCParticipation(
{ {
localMember$ = constant(null), localMember$ = constant(null),
remoteMembers$ = constant([]), remoteMembers$ = constant([]),
connected$ = constant(true), state$ = constant<ParticipationState>({ kind: "connected" }),
reconnecting$ = constant(false),
disconnectReason$ = constant(null),
fatalError$ = constant(null),
keyRotationSuppressed$ = constant(false), keyRotationSuppressed$ = constant(false),
connectedTransports$ = constant([]), connectedTransports$ = constant([]),
}: MockParticipationInputs = {}, }: MockParticipationInputs = {},
@@ -537,13 +530,7 @@ export function mockRTCParticipation(
const participation: RTCParticipation = { const participation: RTCParticipation = {
slot, slot,
leave: vi.fn(), leave: vi.fn(),
status$: scope.behavior( state$,
connected$.pipe(map((c) => (c ? "connected" : "connecting"))),
),
connected$,
reconnecting$,
disconnectReason$,
fatalError$,
localMember$, localMember$,
remoteMembers$, remoteMembers$,
publish: vi.fn(async (request: PublishRequest) => publish: vi.fn(async (request: PublishRequest) =>