Compare commits

...
Author SHA1 Message Date
Johannes Marbach 65e1b845a7 Apply suggestion from @Johennes 2026-09-07 13:18:00 +02:00
Matthew Hodgson eddd95ba41 Log track encryption flag and cryptor events for remote tracks
Extends the [RemoteTracks] logging from #4235 with the per-publication
encryption flag and the ParticipantEncryptionStatusChanged /
EncryptionError room events.

If a publisher encrypts frames while the subscriber believes the
publication is unencrypted, livekit-client bypasses the cryptor and hands
raw ciphertext to the decoder, which is audible as loud noise bursts.
The reverse mismatch (or a missing/invalid key) drops frames instead.
Neither case is visible in a rageshake today.
2026-09-07 11:53:01 +01:00
Johannes Marbach a376280c91 Merge pull request #4240 from element-hq/renovate/livekit-client
Update dependency livekit-client to v2.22.2
2026-09-07 08:07:34 +02:00
renovate[bot] 1e9a5cfa20 Update ghcr.io/element-hq/element-web:develop Docker digest to 6ed62b2 (#4239)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2026-09-07 02:04:08 +00:00
renovate[bot] d204b36b55 Update dependency livekit-client to v2.22.2 2026-09-07 01:56:21 +00:00
renovate[bot] a03f23e720 Update ghcr.io/element-hq/element-web:develop Docker digest to a4e9c8c (#4238)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2026-09-06 02:02:35 +00:00
renovate[bot] 210d94af18 Update ghcr.io/element-hq/element-web:develop Docker digest to 3221e92 (#4237)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2026-09-05 02:55:22 +00:00
renovate[bot]andRobin 70f6e785ce Update ghcr.io/element-hq/element-web:develop Docker digest to 724b468 (#4226)
* Update ghcr.io/element-hq/element-web:develop Docker digest to a9bd290

* Fix Element Web integration tests

---------

Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: Robin <robin@robin.town>
2026-09-04 09:18:19 +00:00
Johannes Marbach 0ad31cfc4d Merge pull request #4232 from element-hq/matthew/screenshare-watchdog
Surface screen share failures instead of silently doing nothing
2026-09-04 07:54:23 +02:00
Johannes Marbach bae8da3ddc Reformat 2026-09-04 07:44:16 +02:00
Johannes Marbach b22eba7af9 Add further tests 2026-09-04 07:39:58 +02:00
Robin cd7510e054 Merge pull request #4235 from element-hq/matthew/remote-track-logging
Log remote participant and track events per LiveKit connection
2026-09-03 20:22:27 +02:00
Robin 054913d17d Use testScope in ObservableScope tests 2026-09-03 20:11:18 +02:00
Matthew Hodgson dee4b05695 Diagnostics for duplicate participant matches and mid-delivery scope teardown
Warn when ConnectionManagerData merges participants from a second
connection to the same SFU URL, when more than one LiveKit participant
matches a member's backend identity, and when a scope is ended while one
of its behaviors is still delivering a value (which strands the later
bound subscribers on the old value). The per-member match log now also
names the connection the participant came from.
2026-09-03 18:43:18 +01:00
Matthew Hodgson b1393998a7 Report re-entrancy depth and tag torn behaviors
A single re-entrant emission can only strand a contiguous run of a
behavior's subscribers, so a tile that is both 'speaking' and 'waiting
for media' (a middle subscriber stranded) implies a nested re-entry.
Report the nesting depth and log each new depth once instead of only the
first re-entry, and tag splitBehavior-derived behaviors with their field
name so the warning identifies which behavior tore.
2026-09-03 18:08:53 +01:00
Matthew Hodgson 2ab2197c01 Warn when a behavior is re-entered while delivering a value
rxjs delivers a nested emission to every subscriber and then resumes
delivering the outer, older value to the remaining subscribers, which
leaves them permanently out of sync. Log the first occurrence per
behavior with a stack trace so the re-entrant path can be identified
from a rageshake.
2026-09-03 17:26:38 +01:00
Matthew Hodgson 0b95fd5a65 Log remote member participant matching and tile waiting state
A tile shows "Waiting for media" for as long as its MatrixRTC member
cannot be matched to a LiveKit participant. Log each transition of that
match and of the tile's waitingForMedia state so rageshakes can tie a
stuck tile to the LiveKit participant and track events.
2026-09-03 17:07:08 +01:00
Matthew Hodgson f8f5caaa09 Log remote participant and track events per LiveKit connection
Rageshakes contained nothing about the state of remote tracks, so a tile
showing the wrong mute or video state for a member could not be
diagnosed. Log connect/disconnect, publish/unpublish, subscribe/
unsubscribe, subscription failures, remote mute/unmute and stream state
changes on each Connection's logger, and remove the listeners when the
connection scope ends.
2026-09-03 16:50:30 +01:00
Matthew Hodgson bf5482db8c Surface screen share failures instead of silently doing nothing
toggleScreenSharing only had `.catch(logger.error)`, so a getDisplayMedia
request that hangs (element-call-rageshakes#17152: Element Desktop on
Windows, the user pressed the screen share button 14 times in 25 seconds
and the log shows nothing but the toggle lines and livekit-client's
"waiting for pending publication promise timed out") left the user with
a button that does nothing and us with no evidence of why.

Log when a toggle is requested and when it completes or fails, with the
elapsed time, so a hang is visible in the logs. Explicit failures other
than the user cancelling the picker show as a non-modal "Could not start
screen sharing" toast. Nothing is inferred from a toggle taking a long
time: the user may simply be choosing what to share.
2026-09-03 15:01:35 +01:00
18 changed files with 617 additions and 64 deletions
+2 -2
View File
@@ -13,7 +13,7 @@ services:
- ./backend/playwright_homeserver-othersite.yaml:/data/cfg/homeserver.yaml:Z
element-web:
# Pin to a SHA so that upstream cannot break our tests. Renovate handles regular updates.
image: ghcr.io/element-hq/element-web:develop@sha256:c107ec33fa349cb5c14fd03a4e13352d64ef264416dac96d273aa07b4004171f
image: ghcr.io/element-hq/element-web:develop@sha256:6ed62b2963ae098a1f79521eb516834dcf29452168e281cc359d9cfb4380adf6
element-web-1:
# Pin to a SHA so that upstream cannot break our tests. Renovate handles regular updates.
image: ghcr.io/element-hq/element-web:develop@sha256:c107ec33fa349cb5c14fd03a4e13352d64ef264416dac96d273aa07b4004171f
image: ghcr.io/element-hq/element-web:develop@sha256:6ed62b2963ae098a1f79521eb516834dcf29452168e281cc359d9cfb4380adf6
+1
View File
@@ -116,6 +116,7 @@
"peer_connection_timeout_description": "Connection to the media server timed out. Try switching to a different network or disabling your VPN. If the problem persists, see our <0>troubleshooting guide</0> or contact your server administrator.",
"room_creation_restricted": "Failed to create call",
"room_creation_restricted_description": "Call creation might be restricted to authorized users only. Try again later, or contact your server admin if the problem persists.",
"screen_share_failed": "Could not start screen sharing",
"sticky_events_required": "Homeserver does not support Matrix 2.0 calls",
"sticky_events_required_description": "This deployment is configured to use Matrix 2.0 call mode, but the homeserver does not advertise support for sticky events (MSC4354). Ask your server admin to upgrade, or switch the deployment to a compatible mode.",
"unexpected_ec_error": "An unexpected error occurred (<0>Error Code:</0> <1>{{ errorCode }}</1>). Please contact your server admin."
+1 -1
View File
@@ -153,7 +153,7 @@ export const widgetTest = test.extend<MyFixtures>({
// Accept the DM invite from brooks
// This how playwright record selects the DM invite in the room list
await ewPage2.getByRole("option", { name: "Open room" }).click();
await ewPage2.getByRole("button", { name: "Open room" }).click();
await ewPage2.getByRole("button", { name: "Start chatting" }).click();
}
+25 -18
View File
@@ -39,13 +39,13 @@ importers:
version: 11.7.12
'@livekit/components-core':
specifier: ^0.12.0
version: 0.12.15(livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22))(tslib@2.8.1)
version: 0.12.15(livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22))(tslib@2.8.1)
'@livekit/components-react':
specifier: ^2.0.0
version: 2.9.24(livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22))(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tslib@2.8.1)
version: 2.9.24(livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22))(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tslib@2.8.1)
'@livekit/track-processors':
specifier: ^0.7.1
version: 0.7.2(@types/dom-mediacapture-transform@0.1.11)(livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22))
version: 0.7.2(@types/dom-mediacapture-transform@0.1.11)(livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22))
'@mediapipe/tasks-vision':
specifier: ^0.10.18
version: 0.10.35
@@ -183,7 +183,7 @@ importers:
version: 5.88.1(@emnapi/core@1.11.2)(@emnapi/runtime@1.11.2)(@types/node@24.13.3)(typescript@5.9.3)
livekit-client:
specifier: ^2.18.1
version: 2.22.0(@types/dom-mediacapture-record@1.0.22)
version: 2.22.2(@types/dom-mediacapture-record@1.0.22)
lodash-es:
specifier: ^4.17.21
version: 4.18.1
@@ -4465,8 +4465,8 @@ packages:
resolution: {integrity: sha512-AC/7JofJvZGrrneWNaEnJeOLUx+JlGt7tNa0wZiRPT4MY1wmfKjt2+6O2p2uz2+skll8OZZmJMNqeke7kKbNgQ==}
hasBin: true
jose@6.2.10:
resolution: {integrity: sha512-iiW7J9qRFlGxvCOIBDBDxFePQSn7ZMAnrYGhrrOo6siO/MIqwfyilLR27pkfDgUk+raLuzADS8A3S/KLBisc0g==}
jose@6.2.12:
resolution: {integrity: sha512-9NiFmJEex0sy2Dk58j2UGBSHgUs2ypF9eZSu4L6vjOX3Dp96Sw1F3uL+H+D1sx02jZZdzUT0HgvCy59CuvXcWw==}
jose@6.2.9:
resolution: {integrity: sha512-XrchZOFZUl/T3vTwRe8XK+cJrGtMF4th1ARnDfwbBXFKThGhlsxEE4Zu03AD/bjJSt/9jT/mxrOCkJWOg77aPA==}
@@ -4614,8 +4614,8 @@ packages:
lines-and-columns@1.2.4:
resolution: {integrity: sha512-7ylylesZQ/PV29jhEDl3Ufjo6ZX7gCqJr5F7PKrqc93v7fzSymt1BpwEU8nAUXs8qzzvqhbjhK5QZg6Mt/HkBg==}
livekit-client@2.22.0:
resolution: {integrity: sha512-GLtYQfRh/RsvXaOX1x609bFZ17yyKmKWDqu9JmkMKn9vIFLi2GsapRv9gT8OJO/1R2dsitrkbGmloyUfxWQsaA==}
livekit-client@2.22.2:
resolution: {integrity: sha512-o3yLYoH2eHVR/DmIZzgC96FMvtR4+g/DjeqNwS5zp12EjqrqiRULxm6Oq/YsTE97WnA6fhy23+a2hGkbce5SPA==}
peerDependencies:
'@types/dom-mediacapture-record': ^1
@@ -4664,6 +4664,10 @@ packages:
resolution: {integrity: sha512-h5bgJWpxJNswbU7qCrV0tIKQCaS3blPDrqKWx+QxzuzL1zGUzij9XCWLrSLsJPu5t+eWA/ycetzYAO5IOMcWAQ==}
hasBin: true
machina@7.0.1:
resolution: {integrity: sha512-0Xkgs8NeBvO2hW6bjJT8ewLWrMfh1haMPstaghpASWjXkLjRhtDiaJG1Kb7mslaOHxZYxpH7PIjpmirRVYbMMA==}
engines: {node: '>=22.22'}
magic-string@0.30.21:
resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==}
@@ -7070,21 +7074,21 @@ snapshots:
'@jridgewell/resolve-uri': 3.1.2
'@jridgewell/sourcemap-codec': 1.5.5
'@livekit/components-core@0.12.15(livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22))(tslib@2.8.1)':
'@livekit/components-core@0.12.15(livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22))(tslib@2.8.1)':
dependencies:
'@floating-ui/dom': 1.7.6
livekit-client: 2.22.0(@types/dom-mediacapture-record@1.0.22)
livekit-client: 2.22.2(@types/dom-mediacapture-record@1.0.22)
loglevel: 1.9.1
rxjs: 7.8.2
tslib: 2.8.1
'@livekit/components-react@2.9.24(livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22))(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tslib@2.8.1)':
'@livekit/components-react@2.9.24(livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22))(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tslib@2.8.1)':
dependencies:
'@livekit/components-core': 0.12.15(livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22))(tslib@2.8.1)
'@livekit/components-core': 0.12.15(livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22))(tslib@2.8.1)
clsx: 2.1.1
events: 3.3.0
jose: 6.2.9
livekit-client: 2.22.0(@types/dom-mediacapture-record@1.0.22)
livekit-client: 2.22.2(@types/dom-mediacapture-record@1.0.22)
react: 19.2.8
react-dom: 19.2.8(react@19.2.8)
tslib: 2.8.1
@@ -7096,11 +7100,11 @@ snapshots:
dependencies:
'@bufbuild/protobuf': 1.10.1
'@livekit/track-processors@0.7.2(@types/dom-mediacapture-transform@0.1.11)(livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22))':
'@livekit/track-processors@0.7.2(@types/dom-mediacapture-transform@0.1.11)(livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22))':
dependencies:
'@mediapipe/tasks-vision': 0.10.35
'@types/dom-mediacapture-transform': 0.1.11
livekit-client: 2.22.0(@types/dom-mediacapture-record@1.0.22)
livekit-client: 2.22.2(@types/dom-mediacapture-record@1.0.22)
'@matrix-org/matrix-sdk-crypto-wasm@18.5.0': {}
@@ -10059,7 +10063,7 @@ snapshots:
jiti@2.7.0: {}
jose@6.2.10: {}
jose@6.2.12: {}
jose@6.2.9: {}
@@ -10203,14 +10207,15 @@ snapshots:
lines-and-columns@1.2.4: {}
livekit-client@2.22.0(@types/dom-mediacapture-record@1.0.22):
livekit-client@2.22.2(@types/dom-mediacapture-record@1.0.22):
dependencies:
'@livekit/mutex': 1.1.1
'@livekit/protocol': 1.50.4
'@types/dom-mediacapture-record': 1.0.22
events: 3.3.0
jose: 6.2.10
jose: 6.2.12
loglevel: 1.9.2
machina: 7.0.1
sdp-transform: 2.15.0
tslib: 2.8.1
typed-emitter: 2.1.0
@@ -10253,6 +10258,8 @@ snapshots:
lz-string@1.5.0: {}
machina@7.0.1: {}
magic-string@0.30.21:
dependencies:
'@jridgewell/sourcemap-codec': 1.5.5
+17
View File
@@ -270,6 +270,7 @@ export const InCallView: FC<InCallViewProps> = ({
const audioParticipants = useBehavior(vm.livekitRoomItems$);
const participantCount = useBehavior(vm.participantCount$);
const reconnecting = useBehavior(vm.reconnecting$);
const screenShareError = useBehavior(vm.screenShareError$);
const layout = useBehavior(vm.layout$);
const edgeToEdge = useBehavior(vm.edgeToEdge$);
const overflowing = useBehavior(vm.overflowing$);
@@ -399,6 +400,21 @@ export const InCallView: FC<InCallViewProps> = ({
);
}
const onDismissScreenShareToast = useCallback(
() => vm.dismissScreenShareError(),
[vm],
);
const screenShareToast = (
<Toast
onDismiss={onDismissScreenShareToast}
open={screenShareError !== null}
autoDismiss={5000}
modal={false}
>
{t("error.screen_share_failed")}
</Toast>
);
// The reconnecting toast cannot be dismissed
const onDismissReconnectingToast = useCallback(() => {}, []);
// We need to use a non-modal toast to avoid trapping focus within the toast.
@@ -634,6 +650,7 @@ export const InCallView: FC<InCallViewProps> = ({
<ReactionsAudioRenderer vm={vm} muted={muteAllAudio} />
<RingingAudioRenderer vm={ringingVm} muted={muteAllAudio} />
{reconnectingToast}
{screenShareToast}
{earpieceOverlay}
<ReactionsOverlay vm={vm} />
{footer}
+10 -10
View File
@@ -171,7 +171,7 @@ exports[`InCallView > rendering > renders 1`] = `
class="_settingsLogoContainer_20b7b4"
>
<button
aria-labelledby="_r_8_"
aria-labelledby="_r_b_"
class="_icon-button_1215g_8 _settingsOnlyShowWide_20b7b4"
data-kind="secondary"
data-testid="settings-bottom-left"
@@ -306,7 +306,7 @@ exports[`InCallView > rendering > renders 1`] = `
class="_buttons_20b7b4"
>
<button
aria-labelledby="_r_d_"
aria-labelledby="_r_g_"
class="_button_1nw83_8 _settingsOnlyShowNarrow_20b7b4 _has-icon_1nw83_60 _icon-only_1nw83_53"
data-kind="secondary"
data-size="lg"
@@ -331,7 +331,7 @@ exports[`InCallView > rendering > renders 1`] = `
aria-busy="false"
aria-checked="false"
aria-disabled="true"
aria-labelledby="_r_i_"
aria-labelledby="_r_l_"
class="_button_1nw83_8 _has-icon_1nw83_60 _icon-only_1nw83_53"
data-kind="primary"
data-size="lg"
@@ -356,7 +356,7 @@ exports[`InCallView > rendering > renders 1`] = `
aria-busy="false"
aria-checked="false"
aria-disabled="true"
aria-labelledby="_r_n_"
aria-labelledby="_r_q_"
class="_button_1nw83_8 _has-icon_1nw83_60 _icon-only_1nw83_53"
data-kind="primary"
data-size="lg"
@@ -381,7 +381,7 @@ exports[`InCallView > rendering > renders 1`] = `
aria-disabled="false"
aria-expanded="false"
aria-haspopup="true"
aria-labelledby="_r_s_"
aria-labelledby="_r_v_"
class="_button_1nw83_8 _raiseHand_20b7b4 _has-icon_1nw83_60 _icon-only_1nw83_53"
data-kind="secondary"
data-size="lg"
@@ -404,7 +404,7 @@ exports[`InCallView > rendering > renders 1`] = `
</svg>
</button>
<button
aria-labelledby="_r_14_"
aria-labelledby="_r_17_"
class="_button_1nw83_8 _has-icon_1nw83_60 _icon-only_1nw83_53 _destructive_1nw83_110"
data-kind="primary"
data-size="lg"
@@ -432,8 +432,8 @@ exports[`InCallView > rendering > renders 1`] = `
data-size="lg"
>
<input
aria-labelledby="_r_1a_"
name="_r_19_"
aria-labelledby="_r_1d_"
name="_r_1c_"
type="radio"
value="spotlight"
/>
@@ -452,9 +452,9 @@ exports[`InCallView > rendering > renders 1`] = `
/>
</svg>
<input
aria-labelledby="_r_1f_"
aria-labelledby="_r_1i_"
checked=""
name="_r_19_"
name="_r_1c_"
type="radio"
value="grid"
/>
+7
View File
@@ -267,6 +267,11 @@ export interface CallViewModel {
* Whether we are sharing our screen.
*/
sharingScreen$: Behavior<boolean>;
/**
* The last error from toggling screen sharing, until dismissed.
*/
screenShareError$: Behavior<Error | null>;
dismissScreenShareError: () => void;
// UI interactions
/**
@@ -1858,6 +1863,8 @@ export function createCallViewModel$(
reconnecting$: localMembership.reconnecting$,
livekitRoomItems$,
connected$: localMembership.connected$,
screenShareError$: localMembership.screenShareError$,
dismissScreenShareError: localMembership.dismissScreenShareError,
};
}
@@ -31,6 +31,7 @@ import {
flushPromises,
mockConfig,
mockLivekitRoom,
mockLocalParticipant,
mockMuteStates,
withTestScheduler,
ownMemberMock,
@@ -41,6 +42,7 @@ import {
enterRTCSession,
PublishState,
TrackState,
watchScreenShareToggle,
} from "./LocalMember";
import {
FailToGetOpenIdToken,
@@ -68,6 +70,31 @@ vi.mock("@livekit/components-core", () => ({
.mockReturnValue(of({ isScreenShareEnabled: false })),
}));
describe("watchScreenShareToggle", () => {
it("reports nothing when the toggle completes", async () => {
const onError = vi.fn();
watchScreenShareToggle(Promise.resolve(), true, logger, onError);
await flushPromises();
expect(onError).not.toHaveBeenCalled();
});
it("reports failures other than the user cancelling", async () => {
const onError = vi.fn();
const e = new Error("NotReadableError");
watchScreenShareToggle(Promise.reject(e), true, logger, onError);
await flushPromises();
expect(onError).toHaveBeenCalledWith(e);
});
it("does not report the user cancelling the picker", async () => {
const onError = vi.fn();
const cancelled = new DOMException("Permission denied", "NotAllowedError");
watchScreenShareToggle(Promise.reject(cancelled), true, logger, onError);
await flushPromises();
expect(onError).not.toHaveBeenCalled();
});
});
describe("LocalMembership", () => {
describe("enterRTCSession", () => {
it("It joins the correct Session", () => {
@@ -854,4 +881,98 @@ describe("LocalMembership", () => {
scope.end();
});
});
describe("toggleScreenSharing", () => {
let originalMediaDevices: MediaDevices | undefined;
beforeAll(() => {
mockConfig();
// Screen sharing is only offered when getDisplayMedia is available.
originalMediaDevices = navigator.mediaDevices;
Object.defineProperty(navigator, "mediaDevices", {
configurable: true,
value: { getDisplayMedia: vi.fn() },
});
});
afterAll(() => {
Object.defineProperty(navigator, "mediaDevices", {
configurable: true,
value: originalMediaDevices,
});
});
const createMembershipWithConnection = (
connection: Connection | null,
): {
scope: ObservableScope;
localMembership: ReturnType<typeof createLocalMembership$>;
} => {
const scope = new ObservableScope();
const connectionManagerData = new ConnectionManagerData();
if (connection) connectionManagerData.add(connection, []);
const localMembership = createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: new BehaviorSubject({
advertised$: new BehaviorSubject(aTransport),
active$: new BehaviorSubject(aTransportWithSFUConfig),
}),
});
return { scope, localMembership };
};
it("surfaces a failure and clears it on dismiss", async () => {
const error = new Error("NotReadableError");
const setScreenShareEnabled = vi.fn().mockRejectedValue(error);
const connection = {
state$: constant(ConnectionState.LivekitConnected),
transport: aTransport,
livekitRoom: mockLivekitRoom({
localParticipant: mockLocalParticipant({
isScreenShareEnabled: false,
setScreenShareEnabled,
}),
}),
} as unknown as Connection;
const { scope, localMembership } =
createMembershipWithConnection(connection);
await flushPromises();
expect(localMembership.toggleScreenSharing).not.toBeNull();
expect(localMembership.screenShareError$.value).toBeNull();
localMembership.toggleScreenSharing!();
await flushPromises();
expect(setScreenShareEnabled).toHaveBeenCalledWith(
true,
expect.any(Object),
undefined,
);
expect(localMembership.screenShareError$.value).toBe(error);
localMembership.dismissScreenShareError();
expect(localMembership.screenShareError$.value).toBeNull();
scope.end();
});
it("does nothing when there is no local participant", async () => {
// No connection means participant$ never resolves to a participant.
const { scope, localMembership } = createMembershipWithConnection(null);
await flushPromises();
expect(localMembership.toggleScreenSharing).not.toBeNull();
localMembership.toggleScreenSharing!();
await flushPromises();
expect(localMembership.screenShareError$.value).toBeNull();
scope.end();
});
});
});
@@ -196,6 +196,11 @@ export const createLocalMembership$ = ({
* Callback to toggle screen sharing. If null, screen sharing is not possible.
*/
toggleScreenSharing: (() => void) | null;
/**
* The last error from toggling screen sharing, until dismissed.
*/
screenShareError$: Behavior<Error | null>;
dismissScreenShareError: () => void;
// tracks$: Behavior<LocalTrack[]>;
participant$: Behavior<LocalParticipant | null>;
connection$: Behavior<Connection | null>;
@@ -706,6 +711,7 @@ export const createLocalMembership$ = ({
),
);
const screenShareError$ = new BehaviorSubject<Error | null>(null);
let toggleScreenSharing: (() => void) | null = null;
if (
"getDisplayMedia" in (navigator.mediaDevices ?? {}) &&
@@ -777,13 +783,18 @@ export const createLocalMembership$ = ({
// We also allow screen sharing to be toggled even if the connection
// is still initializing or publishing tracks, because there's no
// technical reason to disallow this. LiveKit will publish if it can.
participant$.value
?.setScreenShareEnabled(
const participant = participant$.value;
if (!participant) return;
watchScreenShareToggle(
participant.setScreenShareEnabled(
targetScreenshareState,
screenshareSettings,
publishOptions,
)
.catch(logger.error);
),
targetScreenshareState,
logger,
(e) => screenShareError$.next(e),
);
};
}
@@ -802,11 +813,44 @@ export const createLocalMembership$ = ({
),
sharingScreen$,
toggleScreenSharing,
screenShareError$,
dismissScreenShareError: () => screenShareError$.next(null),
connection$: localConnection$,
internalLoggerRef: logger,
};
};
/**
* Logs the outcome of a screen share toggle and reports failures.
*
* getDisplayMedia may legitimately take a long time (the user is choosing
* what to share) or never settle at all, so nothing is inferred from silence:
* the request and its completion are logged with the elapsed time so that a
* hang is visible in the logs, and only an explicit rejection is reported.
*
* The user cancelling the picker rejects with a NotAllowedError; that is
* logged but not reported.
*/
export function watchScreenShareToggle(
toggle: Promise<unknown>,
enable: boolean,
logger: Logger,
onError: (e: Error) => void,
): void {
const what = `Screen share ${enable ? "start" : "stop"}`;
const started = Date.now();
const elapsed = (): string => `${Date.now() - started} ms`;
logger.info(`${what} requested`);
toggle.then(
() => logger.info(`${what} completed in ${elapsed()}`),
(e: unknown) => {
logger.error(`${what} failed after ${elapsed()}:`, e);
if (e instanceof DOMException && e.name === "NotAllowedError") return;
onError(e instanceof Error ? e : new Error(String(e)));
},
);
}
export function observeSharingScreen$(p: Participant): Observable<boolean> {
return observeParticipantEvents(
p,
@@ -19,14 +19,17 @@ import {
import {
type LocalParticipant,
type RemoteParticipant,
type RemoteTrack,
type RemoteTrackPublication,
type Room as LivekitRoom,
RoomEvent,
Track,
ConnectionState as LivekitConnectionState,
} from "livekit-client";
import fetchMock from "fetch-mock";
import EventEmitter from "events";
import { type IOpenIDToken } from "matrix-js-sdk";
import { logger } from "matrix-js-sdk/lib/logger";
import { logger, type Logger } from "matrix-js-sdk/lib/logger";
import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
import {
@@ -509,6 +512,97 @@ describe("remote participants", () => {
});
});
describe("remote track logging", () => {
it("logs remote participant and track events on the connection logger", () => {
setupTest();
const info = vi.fn();
const warn = vi.fn();
const testLogger = {
getChild: (): unknown => testLogger,
info,
debug: vi.fn(),
warn,
error: vi.fn(),
} as unknown as Logger;
new Connection(
{
client,
roomId: ROOM_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
},
testLogger,
);
info.mockClear(); // drop the constructor log line
const bob = mockRemoteParticipant({
identity: "@bob:example.org:DEV111",
sid: "PA_bob",
});
const pub = {
kind: "audio",
source: "microphone",
trackSid: "TR_mic",
isMuted: false,
isEncrypted: true,
} as unknown as RemoteTrackPublication;
const messages = (): string[] => info.mock.calls.map((c) => c[0] as string);
fakeLivekitRoom.emit(RoomEvent.ParticipantConnected, bob);
fakeLivekitRoom.emit(
RoomEvent.TrackSubscribed,
{} as RemoteTrack,
pub,
bob,
);
fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, bob);
fakeLivekitRoom.emit(
RoomEvent.TrackStreamStateChanged,
pub,
Track.StreamState.Paused,
bob,
);
expect(messages()).toEqual([
"Participant connected: @bob:example.org:DEV111 (PA_bob)",
"Subscribed: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true muted=false",
"Muted: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true",
"Stream paused: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true",
]);
// Encryption status changes are logged; cryptor errors are warnings
fakeLivekitRoom.emit(
RoomEvent.ParticipantEncryptionStatusChanged,
false,
bob,
);
expect(messages().at(-1)).toBe(
"Encryption status of @bob:example.org:DEV111: encrypted=false",
);
const cryptorError = new Error("missing key at index 3");
fakeLivekitRoom.emit(RoomEvent.EncryptionError, cryptorError, bob);
expect(warn).toHaveBeenCalledWith(
"Encryption error for @bob:example.org:DEV111:",
cryptorError,
);
// Local mute events are already logged by the Publisher
fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, {
...fakeLocalParticipant,
isLocal: true,
} as unknown as LocalParticipant);
expect(messages().filter((m) => m.startsWith("Muted"))).toHaveLength(1);
// Listeners are removed when the scope ends
testScope.end();
fakeLivekitRoom.emit(RoomEvent.ParticipantDisconnected, bob);
expect(
messages().filter((m) => m.startsWith("Participant disconnected")),
).toHaveLength(0);
});
});
//
// NOT USED ANYMORE ?
//
@@ -13,8 +13,14 @@ import {
import {
ConnectionError,
ConnectionErrorReason,
type Participant,
type RemoteParticipant,
type RemoteTrackPublication,
type Room as LivekitRoom,
RoomEvent,
type SubscriptionError,
type Track,
type TrackPublication,
} from "livekit-client";
import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject, map } from "rxjs";
@@ -164,6 +170,7 @@ export class Connection {
// Only tracks remote participants
connectedParticipantsObserver(this.livekitRoom),
);
this.logRemoteTrackEvents();
scope.onEnd(() => {
this.logger.info(`Connection scope ended, stopping connection`);
@@ -171,6 +178,114 @@ export class Connection {
});
}
/**
* Logs the lifecycle of remote participants and their tracks as seen by
* livekit-client, so that rageshakes show what a tile should have been
* rendering (published, subscribed, muted, paused) for each remote member.
*/
private logRemoteTrackEvents(): void {
const room = this.livekitRoom;
const log = this.logger.getChild("[RemoteTracks]");
// The encryption flag matters: if the publisher encrypts but this client
// believes the track is unencrypted, the cryptor is bypassed and raw
// ciphertext reaches the decoder (audible as loud noise bursts).
const track = (pub: TrackPublication, p: Participant): string =>
`${pub.kind} ${pub.source} ${pub.trackSid} of ${p.identity} encrypted=${pub.isEncrypted}`;
const onParticipantConnected = (p: RemoteParticipant): void =>
log.info(`Participant connected: ${p.identity} (${p.sid})`);
const onParticipantDisconnected = (p: RemoteParticipant): void =>
log.info(`Participant disconnected: ${p.identity} (${p.sid})`);
const onTrackPublished = (
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Published: ${track(pub, p)} muted=${pub.isMuted}`);
const onTrackUnpublished = (
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Unpublished: ${track(pub, p)}`);
const onTrackSubscribed = (
_t: Track,
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Subscribed: ${track(pub, p)} muted=${pub.isMuted}`);
const onTrackUnsubscribed = (
_t: Track,
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Unsubscribed: ${track(pub, p)}`);
const onTrackSubscriptionFailed = (
trackSid: string,
p: RemoteParticipant,
reason?: SubscriptionError,
): void =>
log.warn(
`Subscription failed: ${trackSid} of ${p.identity} reason=${reason}`,
);
// Mute events also fire for the local participant, which the Publisher
// already logs.
const onTrackMuted = (pub: TrackPublication, p: Participant): void => {
if (!p.isLocal) log.info(`Muted: ${track(pub, p)}`);
};
const onTrackUnmuted = (pub: TrackPublication, p: Participant): void => {
if (!p.isLocal) log.info(`Unmuted: ${track(pub, p)}`);
};
const onTrackStreamStateChanged = (
pub: RemoteTrackPublication,
state: Track.StreamState,
p: RemoteParticipant,
): void => log.info(`Stream ${state}: ${track(pub, p)}`);
const onEncryptionStatusChanged = (
encrypted: boolean,
p?: Participant,
): void =>
log.info(
`Encryption status of ${p?.identity ?? "unknown participant"}: encrypted=${encrypted}`,
);
// livekit-client throttles these per cryptor; they indicate frames being
// dropped (missing/invalid key).
const onEncryptionError = (error: Error, p?: Participant): void =>
log.warn(
`Encryption error for ${p?.identity ?? "unknown participant"}:`,
error,
);
room
.on(RoomEvent.ParticipantConnected, onParticipantConnected)
.on(RoomEvent.ParticipantDisconnected, onParticipantDisconnected)
.on(RoomEvent.TrackPublished, onTrackPublished)
.on(RoomEvent.TrackUnpublished, onTrackUnpublished)
.on(RoomEvent.TrackSubscribed, onTrackSubscribed)
.on(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed)
.on(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed)
.on(RoomEvent.TrackMuted, onTrackMuted)
.on(RoomEvent.TrackUnmuted, onTrackUnmuted)
.on(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged)
.on(
RoomEvent.ParticipantEncryptionStatusChanged,
onEncryptionStatusChanged,
)
.on(RoomEvent.EncryptionError, onEncryptionError);
this.scope.onEnd(() => {
room
.off(RoomEvent.ParticipantConnected, onParticipantConnected)
.off(RoomEvent.ParticipantDisconnected, onParticipantDisconnected)
.off(RoomEvent.TrackPublished, onTrackPublished)
.off(RoomEvent.TrackUnpublished, onTrackUnpublished)
.off(RoomEvent.TrackSubscribed, onTrackSubscribed)
.off(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed)
.off(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed)
.off(RoomEvent.TrackMuted, onTrackMuted)
.off(RoomEvent.TrackUnmuted, onTrackUnmuted)
.off(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged)
.off(
RoomEvent.ParticipantEncryptionStatusChanged,
onEncryptionStatusChanged,
)
.off(RoomEvent.EncryptionError, onEncryptionError);
});
}
/**
* Starts the connection.
*
@@ -14,7 +14,7 @@ import { logger } from "matrix-js-sdk/lib/logger";
import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts";
import {
createConnectionManager$,
type ConnectionManagerData,
ConnectionManagerData,
} from "./ConnectionManager.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type Connection } from "./Connection.ts";
@@ -203,6 +203,24 @@ describe("connections$ stream", () => {
});
});
describe("ConnectionManagerData", () => {
test("warns when a second connection to the same URL is merged", () => {
const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
const data = new ConnectionManagerData(logger);
const connection = { transport: TRANSPORT_1 } as unknown as Connection;
const p = (identity: string): RemoteParticipant =>
({ identity }) as unknown as RemoteParticipant;
data.add(connection, [p("a")]);
data.add({ ...connection } as Connection, [p("b")]);
expect(warn).toHaveBeenCalledWith(
expect.stringContaining(
"second connection to https://lk.example.org: existing [a], adding [b]",
),
);
expect(data.getParticipantsForTransport(TRANSPORT_1)).toHaveLength(2);
});
});
describe("connectionManagerData$ stream", () => {
// Used in test to control fake connections' remoteParticipants$ streams
let fakeRemoteParticipantsStreams: Map<string, Behavior<RemoteParticipant[]>>;
@@ -30,7 +30,7 @@ export class ConnectionManagerData {
{ connection: Connection; participants: RemoteParticipant[] }
> = new Map();
public constructor() {}
public constructor(private readonly logger?: Logger) {}
public add(connection: Connection, participants: RemoteParticipant[]): void {
const key = this.getKey(connection.transport);
@@ -38,6 +38,11 @@ export class ConnectionManagerData {
if (!existing) {
this.store.set(key, { connection, participants });
} else {
// Transports are deduplicated by URL upstream, so this should never
// happen; if it does, members may be matched against the wrong room.
this.logger?.warn(
`Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`,
);
existing.participants.push(...participants);
}
}
@@ -239,7 +244,7 @@ export function createConnectionManager$({
lists.reduce((data, { connection, participants }) => {
data.add(connection, participants);
return data;
}, new ConnectionManagerData()),
}, new ConnectionManagerData(logger)),
epoch,
),
),
@@ -5,7 +5,8 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { describe, test, expect, beforeEach, afterEach } from "vitest";
import { describe, test, expect, beforeEach, afterEach, vi } from "vitest";
import { logger } from "matrix-js-sdk/lib/logger";
import {
type CallMembership,
type LivekitTransport,
@@ -77,6 +78,7 @@ function epochMeWith$<T, U>(
}
test("should signal participant not yet connected to livekit", async () => {
const info = vi.spyOn(logger, "info");
const mockedMemberships$ = new BehaviorSubject([bobMembership]);
const mockConnectionManagerData$ = new BehaviorSubject(
new ConnectionManagerData(),
@@ -108,6 +110,9 @@ test("should signal participant not yet connected to livekit", async () => {
return true;
},
);
expect(info).toHaveBeenCalledWith(
"[RemoteMatrixLivekitMembers] @bob:example.org:DEV000: LiveKit participant missing on no connection",
);
});
// Helper to create epoch'ed memberships$ and membershipsWithTransport$ from memberships observable.
@@ -11,6 +11,7 @@ import {
type LivekitTransportConfig,
} from "matrix-js-sdk/lib/matrixrtc";
import { combineLatest, filter, map } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
import { type Behavior } from "../../Behavior";
import { type IConnectionManager } from "./ConnectionManager";
@@ -110,13 +111,17 @@ export function createRemoteMatrixLivekitMembers$({
const participants = transport
? managerData.getParticipantsForTransport(transport)
: [];
const participant =
participants.find(
(p) => p.identity == membership.rtcBackendIdentity,
) ?? null;
const matches = participants.filter(
(p) => p.identity == membership.rtcBackendIdentity,
);
const participant = matches[0] ?? null;
const connection = transport
? managerData.getConnectionForTransport(transport)
: null;
if (matches.length > 1)
logger.warn(
`[RemoteMatrixLivekitMembers] ${membership.rtcBackendIdentity}: ${matches.length} LiveKit participants match (sids ${matches.map((p) => p.sid).join(", ")}), using ${participant?.sid}`,
);
yield {
// This could just be the backend identity without the other keys.
@@ -133,8 +138,16 @@ export function createRemoteMatrixLivekitMembers$({
}
},
// Each update where the key of the generator array do not change will result in updates to the `data$` behavior.
(scope, data$, userId, _deviceId, _memberId, _rtcBackendIdentity) => {
(scope, data$, userId, _deviceId, _memberId, rtcBackendIdentity) => {
const { participant$, ...rest } = scope.splitBehavior(data$);
// Log whether the member could be matched to a LiveKit participant,
// since a tile shows "waiting for media" for as long as it cannot.
participant$.pipe(scope.bind()).subscribe((p) => {
const url = data$.value.connection?.transport.livekit_service_url;
logger.info(
`[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"} on ${url ?? "no connection"}`,
);
});
// will only get called once per backend identity.
// updates to data$ and as a result to displayName$ and mxcAvatarUrl$ are more frequent.
return {
+59 -5
View File
@@ -16,7 +16,7 @@ import {
ObservableScope,
trackEpoch,
} from "./ObservableScope";
import { withTestScheduler } from "../utils/test";
import { testScope, withTestScheduler } from "../utils/test";
describe("Epoch", () => {
it("should map the value correctly", () => {
@@ -87,7 +87,7 @@ describe("Epoch", () => {
});
it("behavior test", () => {
const scope = new ObservableScope();
const scope = testScope();
const s$ = new Subject();
const behavior$ = scope.behavior(s$, 0);
behavior$.subscribe((value) => {
@@ -115,7 +115,7 @@ describe("Reconcile", () => {
it("should wait clean up before processing next", async () => {
vi.useFakeTimers();
const scope = new ObservableScope();
const scope = testScope();
const behavior$ = new BehaviorSubject<number>(0);
const setup = vi.fn().mockImplementation(async () => await sleep(100));
@@ -149,7 +149,7 @@ describe("Reconcile", () => {
it("should skip intermediates values that are not setup", async () => {
vi.useFakeTimers();
const scope = new ObservableScope();
const scope = testScope();
const behavior$ = new BehaviorSubject<number>(0);
const setup = vi
@@ -194,7 +194,7 @@ describe("Reconcile", () => {
it("should wait for setup to complete before starting cleanup", async () => {
vi.useFakeTimers();
const scope = new ObservableScope();
const scope = testScope();
const behavior$ = new BehaviorSubject<number>(0);
const setup = vi
@@ -237,3 +237,57 @@ describe("Reconcile", () => {
expect(setup).toHaveBeenCalledWith(1);
});
});
describe("behavior", () => {
it("warns with the tag and nesting depth, logging each new depth once", () => {
const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
const scope = testScope();
const source$ = new Subject<number>();
const behavior$ = scope.behavior(source$, 0, "tagged");
// Re-enter once on value 1 (depth 2), then a nested re-entry on value 2
// (depth 3), which is the shape needed to strand a middle subscriber.
behavior$.subscribe((v) => {
if (v === 1) source$.next(2);
});
behavior$.subscribe((v) => {
if (v === 2) source$.next(3);
});
const seen: number[] = [];
behavior$.subscribe((v) => seen.push(v));
source$.next(1);
const messages = warn.mock.calls.map((c) => c[0] as string);
expect(messages).toEqual([
expect.stringContaining("Behavior (tagged) re-entered at depth 2"),
expect.stringContaining("Behavior (tagged) re-entered at depth 3"),
]);
warn.mock.calls.forEach((c) => expect(c[1]).toEqual(expect.any(String)));
// The behavior settles on the newest value while the last subscriber is
// left stranded on the oldest.
expect(behavior$.value).toBe(3);
expect(seen.at(-1)).toBe(1);
});
it("warns when the scope is ended while a behavior is mid-delivery", () => {
const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
const scope = new ObservableScope();
const source$ = new Subject<number>();
const behavior$ = scope.behavior(source$, 0);
behavior$.subscribe((v) => {
if (v === 1) scope.end();
});
// A derived behavior is bound to the scope, so ending the scope
// mid-delivery unsubscribes it before the value reaches it.
const derived$ = scope.behavior(behavior$);
source$.next(1);
expect(warn).toHaveBeenCalledWith(
expect.stringContaining("Scope ended while 1 of its behaviors"),
expect.any(String),
);
expect(behavior$.value).toBe(1);
expect(derived$.value).toBe(0);
});
});
+50 -4
View File
@@ -20,6 +20,8 @@ import {
takeUntil,
} from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
import { type Behavior } from "./Behavior";
type MonoTypeOperator = <T>(o: Observable<T>) => Observable<T>;
@@ -34,6 +36,12 @@ const nothing = Symbol("nothing");
* A scope which limits the execution lifetime of its bound Observables.
*/
export class ObservableScope {
/**
* Number of this scope's behaviors currently mid-delivery. Used to detect a
* scope being torn down synchronously from within one of its own emissions.
*/
private delivering = 0;
private readonly ended$ = new BehaviorSubject(false);
private readonly bindImpl: MonoTypeOperator = takeUntil(
@@ -66,6 +74,9 @@ export class ObservableScope {
public behavior<T>(
setValue$: Observable<T>,
initialValue: T | typeof nothing = nothing,
// An optional label used only in the re-entrancy warning below, so that a
// torn behavior can be identified from a rageshake.
tag?: string,
): Behavior<T> {
const subject$ = new BehaviorSubject(initialValue);
// Push values from the Observable into the BehaviorSubject.
@@ -73,11 +84,37 @@ export class ObservableScope {
// they will no longer re-emit their current value upon subscription. We want
// to support Observables that complete (for example `of({})`), so we have to
// take care to not propagate the completion event.
// If a subscriber synchronously causes this same behavior to emit again,
// rxjs delivers the nested value to every subscriber first and then
// resumes delivering the outer (older) value to the remaining subscribers,
// leaving them permanently out of sync with the others. A single such
// re-entry can only strand a contiguous run of subscribers; stranding a
// subscriber in the middle of the list needs a nested (deeper) re-entry, so
// we report the depth and log each new depth (with a stack trace) rather
// than only the first occurrence, to make a nested re-entry visible.
let depth = 0;
let maxReportedDepth = 1;
setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({
next(value) {
subject$.next(value);
next: (value) => {
if (depth > 0 && depth + 1 > maxReportedDepth) {
maxReportedDepth = depth + 1;
logger.warn(
`Behavior${tag ? ` (${tag})` : ""} re-entered at depth ${
depth + 1
} while delivering a value; later subscribers will be left with a stale value`,
new Error().stack,
);
}
depth++;
this.delivering++;
try {
subject$.next(value);
} finally {
depth--;
this.delivering--;
}
},
error(err: unknown) {
error: (err: unknown) => {
subject$.error(err);
},
});
@@ -90,6 +127,11 @@ export class ObservableScope {
* Ends the scope, causing any bound Observables to complete.
*/
public end(): void {
if (this.delivering > 0)
logger.warn(
`Scope ended while ${this.delivering} of its behaviors were still delivering a value; later subscribers will be left with a stale value`,
new Error().stack,
);
this.ended$.next(true);
}
@@ -171,7 +213,11 @@ export class ObservableScope {
return Object.fromEntries(
Object.keys(input$.value).map((key) => [
`${key}$`,
this.behavior(input$.pipe(map((input) => input[key as keyof T]))),
this.behavior(
input$.pipe(map((input) => input[key as keyof T])),
nothing,
key,
),
]),
) as SplitBehavior<T>;
}
+16 -10
View File
@@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details.
import { type RemoteParticipant } from "livekit-client";
import { combineLatest, map, of, switchMap } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
import { type Behavior } from "../Behavior";
import { createVolumeControls, type VolumeControls } from "../VolumeControls";
@@ -45,6 +46,20 @@ export function createRemoteUserMedia(
statsType: "inbound-rtp",
});
const waitingForMedia$ = scope.behavior(
combineLatest(
[inputs.livekitRoom$, inputs.participant$],
(livekitRoom, participant) =>
// If livekitRoom is undefined, the user is not attempting to publish on
// any transport and so we shouldn't expect a participant. (They might
// be a subscribe-only bot for example.)
livekitRoom !== undefined && participant === null,
),
);
waitingForMedia$.pipe(scope.bind()).subscribe((waiting) => {
logger.info(`[RemoteUserMedia ${inputs.id}] waitingForMedia=${waiting}`);
});
return {
...baseUserMedia,
...createVolumeControls(scope, {
@@ -68,15 +83,6 @@ export function createRemoteUserMedia(
),
),
),
waitingForMedia$: scope.behavior(
combineLatest(
[inputs.livekitRoom$, inputs.participant$],
(livekitRoom, participant) =>
// If livekitRoom is undefined, the user is not attempting to publish on
// any transport and so we shouldn't expect a participant. (They might
// be a subscribe-only bot for example.)
livekitRoom !== undefined && participant === null,
),
),
waitingForMedia$,
};
}