[SDK] make stats$ a polled method and add audioLevel$ to audio tracks

This commit is contained in:
Timo K.
2026-10-08 16:36:25 +02:00
parent 5f914b8a01
commit 799619557b
8 changed files with 127 additions and 144 deletions
+11 -5
View File
@@ -599,11 +599,11 @@ export interface MediaTrack {
/** False when the SFU reports the track as unencrypted. Fixed for the life of the track. */ /** False when the SFU reports the track as unencrypted. Fixed for the life of the track. */
encrypted: boolean; encrypted: boolean;
/** /**
* Inbound statistics for a remote track, outbound for a local one. Polled * Inbound statistics for a remote track, outbound for a local one, polled
* once a second, but only while someone is subscribed, so a statistics * every `intervalMs` while the observable is subscribed and not at all
* panel costs nothing while closed. * otherwise, so a statistics panel costs nothing while closed.
*/ */
stats$: Behavior<MediaStreamStats>; stats$(intervalMs: number): Observable<MediaStreamStats>;
/** /**
* Rendering. The view hands its <video> or <audio> element over; the SDK * Rendering. The view hands its <video> or <audio> element over; the SDK
* sets its stream and, for video, registers the size and on-screen * sets its stream and, for video, registers the size and on-screen
@@ -624,6 +624,12 @@ export interface AudioMediaTrack extends MediaTrack {
* member reports the same value. * member reports the same value.
*/ */
isActive$: Behavior<boolean>; isActive$: Behavior<boolean>;
/**
* How loud the track is right now, 0 to 1, as the backend measures it. 0
* while muted. Not what decides `isActive$`: the backend applies its own
* threshold and hysteresis to that.
*/
audioLevel$: Behavior<number>;
/** Route playback through Web Audio, for earpiece pan and gain. Undefined resets. */ /** Route playback through Web Audio, for earpiece pan and gain. Undefined resets. */
setAudioContext(ctx: AudioContext | undefined, plugins?: AudioNode[]): void; setAudioContext(ctx: AudioContext | undefined, plugins?: AudioNode[]): void;
/** No-op for our own audio, which is never played back. */ /** No-op for our own audio, which is never played back. */
@@ -721,7 +727,7 @@ sdk/
metadata, transports, discovery, the js-sdk join metadata, transports, discovery, the js-sdk join
src/media-backend/ the MediaBackend interface in api.ts; livekit/ is the one src/media-backend/ the MediaBackend interface in api.ts; livekit/ is the one
backend: connections, publisher, JWT, keys, the tracks backend: connections, publisher, JWT, keys, the tracks
src/utils/ LazyBehavior, mapScoped, network retry, display names, test helpers src/utils/ mapScoped, network retry, display names, test helpers
dev/ the harness (below) dev/ the harness (below)
SdkArchitecture.md this document SdkArchitecture.md this document
SdkMigration.md how Element Call gets from CallViewModel to the client SdkMigration.md how Element Call gets from CallViewModel to the client
+1 -1
View File
@@ -269,7 +269,7 @@ function render(
"active", "active",
of$((t) => (t.kind === "audio" ? t.isActive$ : of(undefined))), of$((t) => (t.kind === "audio" ? t.isActive$ : of(undefined))),
); );
const stats$ = of$((t) => t.stats$); const stats$ = of$((t) => t.stats$(1000));
label("frameWidth", stats$.pipe(switchMap((s) => of(frames(s)?.frameWidth)))); label("frameWidth", stats$.pipe(switchMap((s) => of(frames(s)?.frameWidth))));
label("frames", stats$.pipe(switchMap((s) => of(frames(s)?.count)))); label("frames", stats$.pipe(switchMap((s) => of(frames(s)?.count))));
} }
+15 -10
View File
@@ -94,13 +94,12 @@ export interface MediaTrack {
* of the track: a publisher that changes its encryption publishes anew. * of the track: a publisher that changes its encryption publishes anew.
*/ */
encrypted: boolean; encrypted: boolean;
/** Polled while subscribed. */ /**
// REVIEW: we want the perf gain of not having it always run. * Inbound statistics for a remote track, outbound for a local one, polled
// * every `intervalMs` while the observable is subscribed and not at all
// proposals * otherwise. Each subscription runs its own poll, so a view shares one.
// - obs -> on subscribe start polling timer */
// - add public method the MediaTrack: for example: actiavteStatsPolling(ms) stats$(intervalMs: number): Observable<MediaStreamStats>;
stats$: Behavior<MediaStreamStats>;
/** /**
* Rendering. The view hands its `<video>` or `<audio>` element over; the SDK * Rendering. The view hands its `<video>` or `<audio>` element over; the SDK
* sets its stream and, for video, registers the size and on-screen observers * sets its stream and, for video, registers the size and on-screen observers
@@ -117,11 +116,17 @@ export interface AudioMediaTrack extends MediaTrack {
/** /**
* Whether the track carries sound right now, as the backend measures it. * Whether the track carries sound right now, as the backend measures it.
* False while muted. A call reads this on the microphone track and calls it * False while muted. A call reads this on the microphone track and calls it
* "speaking". LiveKit measures it per member, so every audio track of a * "speaking".
* member reports the same value. * This is different to audioLevel = 0. The backend might run some hysteresis on it.
*/ */
// REVIEW: audio Level (could even super-seed isActive "audioLevel != 0")
isActive$: Behavior<boolean>; isActive$: Behavior<boolean>;
/**
* How loud the track is right now, 0 to 1, as the backend measures it. 0
* while muted. Not what decides `isActive$`: the backend applies its own
* threshold and hysteresis to that, so a level above 0 may not be active and
* an active track may briefly sit at 0.
*/
audioLevel$: Behavior<number>;
/** Route playback through Web Audio, for earpiece pan and gain. Undefined resets. */ /** Route playback through Web Audio, for earpiece pan and gain. Undefined resets. */
setAudioContext(ctx: AudioContext | undefined, plugins?: AudioNode[]): void; setAudioContext(ctx: AudioContext | undefined, plugins?: AudioNode[]): void;
setVolume(volume: number): void; setVolume(volume: number): void;
@@ -12,6 +12,7 @@ import {
RemoteAudioTrack, RemoteAudioTrack,
type RemoteParticipant, type RemoteParticipant,
type Room as LivekitRoom, type Room as LivekitRoom,
RoomEvent,
Track, Track,
TrackEvent, TrackEvent,
type TrackPublication, type TrackPublication,
@@ -30,14 +31,18 @@ describe("createLivekitMediaTrack", () => {
beforeEach(() => { beforeEach(() => {
scope = new ObservableScope(); scope = new ObservableScope();
}); });
afterEach(() => scope.end()); afterEach(() => {
scope.end();
vi.useRealTimers();
});
it("describes the publication", () => { it("describes the publication", () => {
const { participant, publication } = fakes(); const { participant, publication, room } = fakes();
const track = createLivekitMediaTrack( const track = createLivekitMediaTrack(
scope, scope,
participant, participant,
publication as unknown as TrackPublication, publication as unknown as TrackPublication,
room,
); );
expect(track.kind).toBe("video"); expect(track.kind).toBe("video");
expect(track.source).toBe("camera"); expect(track.source).toBe("camera");
@@ -46,7 +51,7 @@ describe("createLivekitMediaTrack", () => {
}); });
it("attaches elements to the track, even one that arrives later", () => { it("attaches elements to the track, even one that arrives later", () => {
const { participant, publication, emitter } = fakes({ const { participant, publication, room, emitter } = fakes({
track: undefined, track: undefined,
}); });
const element = document.createElement("video"); const element = document.createElement("video");
@@ -54,6 +59,7 @@ describe("createLivekitMediaTrack", () => {
scope, scope,
participant, participant,
publication as unknown as TrackPublication, publication as unknown as TrackPublication,
room,
); );
track.attach(element); track.attach(element);
track.attach(element); track.attach(element);
@@ -69,13 +75,14 @@ describe("createLivekitMediaTrack", () => {
}); });
it("detaches everything when its scope ends", () => { it("detaches everything when its scope ends", () => {
const { participant, publication } = fakes(); const { participant, publication, room } = fakes();
const element = document.createElement("video"); const element = document.createElement("video");
const scope = new ObservableScope(); const scope = new ObservableScope();
const track = createLivekitMediaTrack( const track = createLivekitMediaTrack(
scope, scope,
participant, participant,
publication as unknown as TrackPublication, publication as unknown as TrackPublication,
room,
); );
track.attach(element); track.attach(element);
scope.end(); scope.end();
@@ -83,11 +90,12 @@ describe("createLivekitMediaTrack", () => {
}); });
it("follows the mute state", () => { it("follows the mute state", () => {
const { participant, publication, emitter } = fakes(); const { participant, publication, room, emitter } = fakes();
const track = createLivekitMediaTrack( const track = createLivekitMediaTrack(
scope, scope,
participant, participant,
publication as unknown as TrackPublication, publication as unknown as TrackPublication,
room,
); );
expect(track.muted$.value).toBe(false); expect(track.muted$.value).toBe(false);
publication.isMuted = true; publication.isMuted = true;
@@ -96,26 +104,28 @@ describe("createLivekitMediaTrack", () => {
}); });
it("scales a remote member's volume on the participant", () => { it("scales a remote member's volume on the participant", () => {
const { participant, publication } = fakes({ const { participant, publication, room } = fakes({
kind: Track.Kind.Audio, kind: Track.Kind.Audio,
}); });
const track = createLivekitMediaTrack( const track = createLivekitMediaTrack(
scope, scope,
participant, participant,
publication as unknown as TrackPublication, publication as unknown as TrackPublication,
room,
) as AudioMediaTrack; ) as AudioMediaTrack;
track.setVolume(0.5); track.setVolume(0.5);
expect(publication.track!.setVolume).toHaveBeenCalledWith(0.5); expect(publication.track!.setVolume).toHaveBeenCalledWith(0.5);
}); });
it("is active while the member speaks and the track is not muted", () => { it("is active while the member speaks and the track is not muted", () => {
const { participant, publication, emitter } = fakes({ const { participant, publication, room, emitter } = fakes({
kind: Track.Kind.Audio, kind: Track.Kind.Audio,
}); });
const track = createLivekitMediaTrack( const track = createLivekitMediaTrack(
scope, scope,
participant, participant,
publication as unknown as TrackPublication, publication as unknown as TrackPublication,
room,
) as AudioMediaTrack; ) as AudioMediaTrack;
expect(track.isActive$.value).toBe(false); expect(track.isActive$.value).toBe(false);
participant.isSpeaking = true; participant.isSpeaking = true;
@@ -126,6 +136,57 @@ describe("createLivekitMediaTrack", () => {
expect(track.isActive$.value).toBe(false); expect(track.isActive$.value).toBe(false);
}); });
it("reports the audio level the room announces, 0 while muted", () => {
const { participant, publication, room, emitter } = fakes({
kind: Track.Kind.Audio,
});
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
) as AudioMediaTrack;
expect(track.audioLevel$.value).toBe(0);
participant.audioLevel = 0.4;
(room as unknown as EventEmitter).emit(RoomEvent.ActiveSpeakersChanged, [
participant,
]);
expect(track.audioLevel$.value).toBe(0.4);
publication.isMuted = true;
emitter.emit(ParticipantEvent.TrackMuted, publication);
expect(track.audioLevel$.value).toBe(0);
});
it("polls statistics at the given interval, only while subscribed", async () => {
vi.useFakeTimers();
const { participant, publication, room } = fakes();
const stats = { type: "inbound-rtp" } as RTCInboundRtpStreamStats;
publication.track!.getRTCStatsReport.mockResolvedValue(
new Map([["s", stats]]),
);
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
);
const stats$ = track.stats$(500);
await vi.advanceTimersByTimeAsync(1000);
expect(publication.track!.getRTCStatsReport).not.toHaveBeenCalled();
const seen = vi.fn();
const subscription = stats$.subscribe(seen);
await vi.advanceTimersByTimeAsync(0);
expect(publication.track!.getRTCStatsReport).toHaveBeenCalledTimes(1);
expect(seen).toHaveBeenLastCalledWith(stats);
await vi.advanceTimersByTimeAsync(1000);
expect(publication.track!.getRTCStatsReport).toHaveBeenCalledTimes(3);
subscription.unsubscribe();
await vi.advanceTimersByTimeAsync(1000);
expect(publication.track!.getRTCStatsReport).toHaveBeenCalledTimes(3);
});
it("gives our own track its controls", async () => { it("gives our own track its controls", async () => {
const { participant, publication, room } = fakes({ const { participant, publication, room } = fakes({
kind: Track.Kind.Audio, kind: Track.Kind.Audio,
@@ -152,6 +213,7 @@ type FakeTrack = Track & {
attach: ReturnType<typeof vi.fn>; attach: ReturnType<typeof vi.fn>;
detach: ReturnType<typeof vi.fn>; detach: ReturnType<typeof vi.fn>;
setVolume: ReturnType<typeof vi.fn>; setVolume: ReturnType<typeof vi.fn>;
getRTCStatsReport: ReturnType<typeof vi.fn>;
}; };
type FakePublication = Omit<TrackPublication, "track" | "isMuted"> & { type FakePublication = Omit<TrackPublication, "track" | "isMuted"> & {
@@ -166,6 +228,7 @@ function fakeTrack(): FakeTrack {
setVolume: vi.fn(), setVolume: vi.fn(),
setAudioContext: vi.fn(), setAudioContext: vi.fn(),
setWebAudioPlugins: vi.fn(), setWebAudioPlugins: vi.fn(),
getRTCStatsReport: vi.fn().mockResolvedValue(undefined),
}) as FakeTrack; }) as FakeTrack;
} }
@@ -173,7 +236,7 @@ function fakes({
track = fakeTrack(), track = fakeTrack(),
kind = Track.Kind.Video, kind = Track.Kind.Video,
}: { track?: FakeTrack | undefined; kind?: Track.Kind } = {}): { }: { track?: FakeTrack | undefined; kind?: Track.Kind } = {}): {
participant: RemoteParticipant & { isSpeaking: boolean }; participant: RemoteParticipant & { isSpeaking: boolean; audioLevel: number };
publication: FakePublication; publication: FakePublication;
room: LivekitRoom; room: LivekitRoom;
emitter: EventEmitter; emitter: EventEmitter;
@@ -182,10 +245,14 @@ function fakes({
const participant = Object.assign(emitter, { const participant = Object.assign(emitter, {
isLocal: false, isLocal: false,
isSpeaking: false, isSpeaking: false,
audioLevel: 0,
identity: "@alice:example.org:DEVICE", identity: "@alice:example.org:DEVICE",
getTrackPublication: (): TrackPublication => getTrackPublication: (): TrackPublication =>
publication as unknown as TrackPublication, publication as unknown as TrackPublication,
}) as unknown as RemoteParticipant & { isSpeaking: boolean }; }) as unknown as RemoteParticipant & {
isSpeaking: boolean;
audioLevel: number;
};
const publication = Object.assign(emitter, { const publication = Object.assign(emitter, {
kind, kind,
source: source:
@@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details.
import { import {
observeParticipantEvents, observeParticipantEvents,
observeParticipantMedia, observeParticipantMedia,
roomEventSelector,
} from "@livekit/components-core"; } from "@livekit/components-core";
import { import {
facingModeFromLocalTrack, facingModeFromLocalTrack,
@@ -19,6 +20,7 @@ import {
RemoteAudioTrack, RemoteAudioTrack,
RemoteTrack, RemoteTrack,
type Room as LivekitRoom, type Room as LivekitRoom,
RoomEvent,
Track, Track,
TrackEvent, TrackEvent,
type TrackPublication, type TrackPublication,
@@ -27,13 +29,13 @@ import {
combineLatest, combineLatest,
distinctUntilChanged, distinctUntilChanged,
fromEvent, fromEvent,
interval,
map, map,
merge, merge,
type Observable,
of, of,
share,
startWith, startWith,
switchMap, switchMap,
timer,
} from "rxjs"; } from "rxjs";
import { type Behavior } from "../../reactive/Behavior"; import { type Behavior } from "../../reactive/Behavior";
@@ -48,7 +50,6 @@ import {
type VideoMediaTrack, type VideoMediaTrack,
} from "../../media-api"; } from "../../media-api";
import { MatrixRTCError } from "../../errors"; import { MatrixRTCError } from "../../errors";
import { LazyBehavior } from "../../utils/LazyBehavior";
import { convertToLivekitProcessor } from "./videoProcessor"; import { convertToLivekitProcessor } from "./videoProcessor";
export const mediaSources: Record<Track.Source, MediaSource> = { export const mediaSources: Record<Track.Source, MediaSource> = {
@@ -67,10 +68,6 @@ export const livekitSources: Record<MediaSource, Track.Source> = {
unknown: Track.Source.Unknown, unknown: Track.Source.Unknown,
}; };
// One timer for every track so that a large session does not keep hundreds of
// them, each firing a statistics request, in the event loop.
const refreshStats$ = interval(1000).pipe(startWith(0), share());
/** /**
* One publication of a participant as a `MediaTrack`. The publication is * One publication of a participant as a `MediaTrack`. The publication is
* fixed; the track behind it may come and go (a remote track arrives on * fixed; the track behind it may come and go (a remote track arrives on
@@ -80,6 +77,7 @@ export function createLivekitMediaTrack(
scope: ObservableScope, scope: ObservableScope,
participant: Participant, participant: Participant,
publication: TrackPublication, publication: TrackPublication,
room: LivekitRoom,
): AudioMediaTrack | VideoMediaTrack { ): AudioMediaTrack | VideoMediaTrack {
const mediaChanged$ = observeParticipantMedia(participant); const mediaChanged$ = observeParticipantMedia(participant);
const track$ = publicationTrack$(scope, participant, publication); const track$ = publicationTrack$(scope, participant, publication);
@@ -117,13 +115,11 @@ export function createLivekitMediaTrack(
id: publication.trackSid, id: publication.trackSid,
muted$, muted$,
encrypted: publication.isEncrypted, encrypted: publication.isEncrypted,
stats$: new LazyBehavior<MediaStreamStats>( stats$: (intervalMs): Observable<MediaStreamStats> =>
refreshStats$.pipe( timer(0, intervalMs).pipe(
switchMap(async () => rtpStreamStats(publication, participant.isLocal)), switchMap(async () => rtpStreamStats(publication, participant.isLocal)),
scope.bind(), scope.bind(),
), ),
undefined,
),
attach: (element) => { attach: (element) => {
if (attached.has(element)) return; if (attached.has(element)) return;
attached.add(element); attached.add(element);
@@ -154,6 +150,20 @@ export function createLivekitMediaTrack(
), ),
participant.isSpeaking && !publication.isMuted, participant.isSpeaking && !publication.isMuted,
), ),
// The level has no event of its own: it is written to the participant
// with each speaker update, which the room announces
audioLevel$: scope.behavior(
combineLatest(
[
roomEventSelector(room, RoomEvent.ActiveSpeakersChanged).pipe(
map(() => participant.audioLevel),
startWith(participant.audioLevel),
),
muted$,
],
(level, muted) => (muted ? 0 : level),
),
),
setAudioContext: (ctx, plugins = []) => { setAudioContext: (ctx, plugins = []) => {
audioContext = ctx; audioContext = ctx;
audioPlugins = plugins; audioPlugins = plugins;
@@ -181,7 +191,7 @@ export function createLocalLivekitMediaTrack(
room: LivekitRoom, room: LivekitRoom,
setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>, setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>,
): LocalAudioMediaTrack | LocalVideoMediaTrack { ): LocalAudioMediaTrack | LocalVideoMediaTrack {
const base = createLivekitMediaTrack(scope, participant, publication); const base = createLivekitMediaTrack(scope, participant, publication, room);
const local = { const local = {
setEnabled: async (enabled: boolean) => setEnabled(base.source, enabled), setEnabled: async (enabled: boolean) => setEnabled(base.source, enabled),
setDevice: async (deviceId: string): Promise<void> => { setDevice: async (deviceId: string): Promise<void> => {
@@ -54,8 +54,8 @@ export function createLivekitMemberMedia(
tracks$: memberTracks$( tracks$: memberTracks$(
scope, scope,
source$, source$,
(trackScope, participant, publication) => (trackScope, participant, publication, room) =>
createLivekitMediaTrack(trackScope, participant, publication), createLivekitMediaTrack(trackScope, participant, publication, room),
), ),
encryptionError$: encryptionErrors$(source$, encryptionSystem), encryptionError$: encryptionErrors$(source$, encryptionSystem),
}; };
-52
View File
@@ -1,52 +0,0 @@
/*
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, vi } from "vitest";
import { Observable, Subject } from "rxjs";
import { LazyBehavior } from "./LazyBehavior";
describe("LazyBehavior", () => {
it("runs its source only while subscribed", () => {
const teardown = vi.fn();
let starts = 0;
const values$ = new Subject<number>();
const source$ = new Observable<number>((subscriber) => {
starts++;
const subscription = values$.subscribe(subscriber);
return (): void => {
teardown();
subscription.unsubscribe();
};
});
const lazy$ = new LazyBehavior(source$, 0);
expect(starts).toBe(0);
expect(lazy$.value).toBe(0);
const seenByFirst: number[] = [];
const first = lazy$.subscribe((v) => seenByFirst.push(v));
expect(starts).toBe(1);
values$.next(1);
expect(seenByFirst).toEqual([0, 1]);
expect(lazy$.value).toBe(1);
const seenBySecond: number[] = [];
const second = lazy$.subscribe((v) => seenBySecond.push(v));
expect(starts).toBe(1);
expect(seenBySecond).toEqual([1]);
first.unsubscribe();
expect(teardown).not.toHaveBeenCalled();
second.unsubscribe();
expect(teardown).toHaveBeenCalledTimes(1);
values$.next(2);
expect(lazy$.value).toBe(1);
lazy$.subscribe().unsubscribe();
expect(starts).toBe(2);
});
});
-53
View File
@@ -1,53 +0,0 @@
/*
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 {
BehaviorSubject,
type Observable,
type Subscriber,
type Subscription,
} from "rxjs";
/**
* A behavior that only runs its source while someone is subscribed. The source
* is subscribed with the first subscriber and unsubscribed with the last, so a
* value that is expensive to keep current, such as polled statistics, costs
* nothing while nobody looks at it. `value` is whatever the source last
* produced, or the initial value.
*/
export class LazyBehavior<T> extends BehaviorSubject<T> {
private subscribers = 0;
private upstream: Subscription | undefined;
public constructor(
private readonly source$: Observable<T>,
initialValue: T,
) {
super(initialValue);
}
// Every subscription of an Observable goes through this hook, which rxjs
// declares as internal and so leaves out of its types. The base class's
// version replays the current value and registers the subscriber; it is
// wrapped rather than replaced.
protected _subscribe(subscriber: Subscriber<T>): Subscription {
if (this.subscribers++ === 0)
this.upstream = this.source$.subscribe((value) => this.next(value));
const subscription = (
BehaviorSubject.prototype as unknown as {
_subscribe(this: BehaviorSubject<T>, s: Subscriber<T>): Subscription;
}
)._subscribe.call(this, subscriber);
subscription.add(() => {
if (--this.subscribers === 0) {
this.upstream?.unsubscribe();
this.upstream = undefined;
}
});
return subscription;
}
}