From 9b67bc1aa067a288b02f55952132ba64a3cd5765 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A5vard=20Holvik?= Date: Wed, 17 Dec 2025 15:34:43 +0100 Subject: [PATCH 1/2] media: Replace rtcStats singleton with factory function --- .changeset/breezy-pens-work.md | 12 + packages/media/src/webrtc/BandwidthTester.ts | 7 +- packages/media/src/webrtc/P2pRtcManager.ts | 40 +-- .../media/src/webrtc/RtcManagerDispatcher.ts | 5 + packages/media/src/webrtc/Session.ts | 14 +- packages/media/src/webrtc/VegaConnection.ts | 14 +- .../media/src/webrtc/VegaConnectionManager.ts | 22 +- .../VegaRtcManager/__tests__/index.spec.ts | 5 + .../media/src/webrtc/VegaRtcManager/index.ts | 73 +++-- .../webrtc/__tests__/rtcStatsService.spec.ts | 265 ++++++++++++++++++ packages/media/src/webrtc/index.ts | 2 +- packages/media/src/webrtc/rtcStatsService.ts | 90 +++--- packages/media/src/webrtc/sdpModifier.ts | 2 - .../IssueMonitor/__tests__/index.spec.ts | 6 +- .../src/webrtc/stats/IssueMonitor/index.ts | 6 +- .../StatsMonitor/__tests__/index.spec.ts | 16 +- .../__tests__/peerConnection.spec.ts | 14 +- .../webrtc/stats/StatsMonitor/collectStats.ts | 7 +- .../src/webrtc/stats/StatsMonitor/index.ts | 13 +- .../stats/StatsMonitor/peerConnection.ts | 4 +- packages/media/src/webrtc/types.ts | 2 + .../media/tests/webrtc/P2pRtcManager.spec.ts | 10 +- .../tests/webrtc/RtcManagerDispatcher.spec.ts | 5 +- packages/media/tests/webrtc/Session.spec.ts | 9 +- .../webrtc/VegaConnectionManager.spec.ts | 72 ++--- packages/media/tests/webrtc/webRtcHelpers.ts | 34 +++ 26 files changed, 584 insertions(+), 165 deletions(-) create mode 100644 .changeset/breezy-pens-work.md create mode 100644 packages/media/src/webrtc/__tests__/rtcStatsService.spec.ts diff --git a/.changeset/breezy-pens-work.md b/.changeset/breezy-pens-work.md new file mode 100644 index 000000000..34c628844 --- /dev/null +++ b/.changeset/breezy-pens-work.md @@ -0,0 +1,12 @@ +--- +"@whereby.com/media": major +--- + +Removes rtcStatsService default singleton export. + +- Removed rtcStatsService default export + - **What changed**: The default singleton export has been removed and replaced with a named export `createRtcStatsConnection()`. + * `P2pRtcManager`, `VegaRtcManager` and `RtcManagerDispatcher` now require `rtcStats` in their constructors + * `startStatsMonitor`, `subscribeStats` and `subscribeIssues` have gotten a required `rtcStats` argument + - **How to adapt**: Consuming apps will have to create rtc stats connections using the new export and pass it on to any other module depending on this functionality. + - **Why**: Having a singleton connection made it harder to test this functionality, and it gave consumers limited possibility to inject options, like overriding the base url for the connection. diff --git a/packages/media/src/webrtc/BandwidthTester.ts b/packages/media/src/webrtc/BandwidthTester.ts index cd3e38dc5..db68b7181 100644 --- a/packages/media/src/webrtc/BandwidthTester.ts +++ b/packages/media/src/webrtc/BandwidthTester.ts @@ -5,6 +5,7 @@ import { getMediaSettings, modifyMediaCapabilities } from "../utils/mediaSetting import Logger from "../utils/Logger"; import { getMediasoupDeviceAsync } from "../utils/getMediasoupDevice"; import { ClearableTimeout } from "../utils"; +import { RtcStatsConnection } from "./rtcStatsService"; const logger = new Logger(); @@ -27,8 +28,9 @@ export default class BandwidthTester extends EventEmitter { _canvas: any; _drawInterval: any; _resultTimeout: ClearableTimeout | null; + _rtcStats: RtcStatsConnection; - constructor({ features }: { features?: any } = {}) { + constructor(rtcStats: RtcStatsConnection, { features }: { features?: any } = {}) { super(); this.closed = false; @@ -59,6 +61,7 @@ export default class BandwidthTester extends EventEmitter { this._drawInterval = null; this._resultTimeout = null; + this._rtcStats = rtcStats; } // This is the public API for this class @@ -84,7 +87,7 @@ export default class BandwidthTester extends EventEmitter { const host = this._features.sfuServerOverrideHost || "any.sfu.svc.whereby.com"; const wsUrl = `wss://${host}`; - this._vegaConnection = new VegaConnection(wsUrl, { protocol: "whereby-sfu#bw-test-v1" }); + this._vegaConnection = new VegaConnection(wsUrl, this._rtcStats, { protocol: "whereby-sfu#bw-test-v1" }); this._vegaConnection.on("open", () => this._start()); this._vegaConnection.on("close", () => this.close(true)); this._vegaConnection.on("message", (message: any) => this._onMessage(message)); diff --git a/packages/media/src/webrtc/P2pRtcManager.ts b/packages/media/src/webrtc/P2pRtcManager.ts index b2bf13490..269fd3179 100644 --- a/packages/media/src/webrtc/P2pRtcManager.ts +++ b/packages/media/src/webrtc/P2pRtcManager.ts @@ -1,4 +1,4 @@ -import rtcStats from "./rtcStatsService"; +import { RtcStatsConnection } from "./rtcStatsService"; import Session from "./Session"; import { MEDIA_JITTER_BUFFER_TARGET } from "./constants"; import * as webrtcBugDetector from "./bugDetector"; @@ -128,8 +128,9 @@ export default class P2pRtcManager implements RtcManager { _rtcStatsDisconnectTimeout?: ReturnType; _webcamPaused?: boolean; _videoTrackIdByStreamId: Record; + _rtcStats: RtcStatsConnection; - constructor({ selfId, room, emitter, serverSocket, webrtcProvider, features }: RtcManagerOptions) { + constructor({ selfId, room, rtcStats, emitter, serverSocket, webrtcProvider, features }: RtcManagerOptions) { const { name, session, iceServers, turnServers, mediaserverConfigTtlSeconds } = room; this._selfId = selfId; @@ -146,6 +147,7 @@ export default class P2pRtcManager implements RtcManager { this._isAudioOnlyMode = false; this._closed = false; this._videoTrackIdByStreamId = {}; + this._rtcStats = rtcStats; // Timeouts this._fetchMediaServersTimer = null; @@ -159,13 +161,13 @@ export default class P2pRtcManager implements RtcManager { // One of them is getting unplugged. The other is the Chrome audio // process crashing. The third is the tab being closed. // https://bugs.chromium.org/p/chromium/issues/detail?id=1050008 - rtcStats.sendEvent("audio_ended", { unloading }); + this._rtcStats.sendEvent("audio_ended", { unloading }); this._emit(rtcManagerEvents.MICROPHONE_STOPPED_WORKING, {}); this.analytics.micTrackEndedCount++; }; this._videoTrackOnEnded = () => { - rtcStats.sendEvent("video_ended", { unloading }); + this._rtcStats.sendEvent("video_ended", { unloading }); this._emit(rtcManagerEvents.CAMERA_STOPPED_WORKING, {}); this.analytics.camTrackEndedCount++; }; @@ -400,7 +402,7 @@ export default class P2pRtcManager implements RtcManager { ) { logger.info("We have asked for an initial offer, ignoring all others"); this.analytics.P2POffendingInitialOffer++; - rtcStats.sendEvent("P2POffendingInitialOffer", { clientId: session.clientId }); + this._rtcStats.sendEvent("P2POffendingInitialOffer", { clientId: session.clientId }); return; } session @@ -450,32 +452,31 @@ export default class P2pRtcManager implements RtcManager { } sendAudioMutedStats(muted: boolean) { - rtcStats.sendEvent("audio_muted", { muted }); + this._rtcStats.sendEvent("audio_muted", { muted }); } sendVideoMutedStats(muted: boolean) { - rtcStats.sendEvent("video_muted", { muted }); + this._rtcStats.sendEvent("video_muted", { muted }); } sendStatsCustomEvent(eventName: string, data: any) { - rtcStats.sendEvent(eventName, data); + this._rtcStats.sendEvent(eventName, data); } rtcStatsConnect() { - if (!rtcStats.server.connected) { - rtcStats.server.connect(); + if (!this._rtcStats.server.connected) { + this._rtcStats.server.connect(); } } rtcStatsDisconnect() { clearTimeout(this._rtcStatsDisconnectTimeout); - - rtcStats.server.close(); + this._rtcStats.server.close(); } rtcStatsReconnect() { - if (!rtcStats.server.connected && rtcStats.server.attemptedConnectedAtLeastOnce) { - rtcStats.server.connect(); + if (!this._rtcStats.server.connected && this._rtcStats.server.attemptedConnectedAtLeastOnce) { + this._rtcStats.server.connect(); } } @@ -561,7 +562,7 @@ export default class P2pRtcManager implements RtcManager { return; } if (this._features.awaitJoinRoomFinished && !this._serverSocket.joinRoomFinished) { - rtcStats.sendEvent("skip_emitting_server_message", { eventName }); + this._rtcStats.sendEvent("skip_emitting_server_message", { eventName }); } else { this._serverSocket.emit(eventName, data); } @@ -610,6 +611,7 @@ export default class P2pRtcManager implements RtcManager { deprioritizeH264Encoding, incrementAnalyticMetric: (metric: P2PAnalyticMetric) => this.analytics[metric]++, mediaPrefs: this._remoteClientMediaPrefs[clientId], + rtcStats: this._rtcStats, }); this.peerConnections[clientId] = session; @@ -737,7 +739,7 @@ export default class P2pRtcManager implements RtcManager { const stream = event.streams[0]; if (!stream) { this.analytics.P2POnTrackNoStream++; - rtcStats.sendEvent("P2POnTrackNoStream", { + this._rtcStats.sendEvent("P2POnTrackNoStream", { trackKind: event.track.kind, trackId: event.track.id, }); @@ -803,7 +805,7 @@ export default class P2pRtcManager implements RtcManager { // We did not gather any relay candidates and we were never connected. // At that point we consider it a local network problem. this.analytics.P2PLocalNetworkFailed++; - rtcStats.sendEvent("P2PLocalNetworkFailed", { + this._rtcStats.sendEvent("P2PLocalNetworkFailed", { from: "iceConnectionStateChange", clientId, }); @@ -828,7 +830,7 @@ export default class P2pRtcManager implements RtcManager { webrtcBugDetector.detectMicrophoneNotWorking(session.pc).then((failureDirection) => { if (failureDirection) { this.analytics.P2PMicNotWorking++; - rtcStats.sendEvent("P2PMicNotWorking", { clientId, failureDirection }); + this._rtcStats.sendEvent("P2PMicNotWorking", { clientId, failureDirection }); // TODO: Decide if we want to act on this or not. // this._emit(rtcManagerEvents.MICROPHONE_NOT_WORKING, { @@ -851,7 +853,7 @@ export default class P2pRtcManager implements RtcManager { // We did not gather any relay candidates and we were never connected. // At that point we consider it a local network problem. this.analytics.P2PLocalNetworkFailed++; - rtcStats.sendEvent("P2PLocalNetworkFailed", { + this._rtcStats.sendEvent("P2PLocalNetworkFailed", { from: "connectionStateChange", clientId, }); diff --git a/packages/media/src/webrtc/RtcManagerDispatcher.ts b/packages/media/src/webrtc/RtcManagerDispatcher.ts index 0150916d9..91bdeb511 100644 --- a/packages/media/src/webrtc/RtcManagerDispatcher.ts +++ b/packages/media/src/webrtc/RtcManagerDispatcher.ts @@ -4,6 +4,7 @@ import * as CONNECTION_STATUS from "../model/connectionStatusConstants"; import VegaRtcManager from "./VegaRtcManager"; import { RoomJoinedEvent, ServerSocket } from "../utils"; import { RtcManager, RtcEvents, RtcEventEmitter, WebRTCProvider } from "./types"; +import { RtcStatsConnection } from "./rtcStatsService"; export default class RtcManagerDispatcher { emitter: { emit: (eventName: K, args?: RtcEvents[K]) => void }; @@ -14,11 +15,13 @@ export default class RtcManagerDispatcher { serverSocket, webrtcProvider, features, + rtcStats, }: { emitter: RtcEventEmitter; serverSocket: ServerSocket; webrtcProvider: WebRTCProvider; features: any; + rtcStats: RtcStatsConnection; }) { this.emitter = emitter; this.currentManager = null; @@ -34,6 +37,7 @@ export default class RtcManagerDispatcher { webrtcProvider, features, eventClaim, + rtcStats, }; const isSfu = !!room.sfuServer; roomMode = isSfu ? "group" : "normal"; @@ -42,6 +46,7 @@ export default class RtcManagerDispatcher { if (this.currentManager.setEventClaim && eventClaim) { this.currentManager.setEventClaim(eventClaim); } + return; } this.currentManager.disconnectAll(); diff --git a/packages/media/src/webrtc/Session.ts b/packages/media/src/webrtc/Session.ts index 5b5108e1d..cb270ed94 100644 --- a/packages/media/src/webrtc/Session.ts +++ b/packages/media/src/webrtc/Session.ts @@ -2,8 +2,8 @@ import * as sdpModifier from "./sdpModifier"; import { setVideoBandwidthUsingSetParameters } from "./rtcrtpsenderHelper"; import adapterRaw from "webrtc-adapter"; import Logger from "../utils/Logger"; -import rtcStats from "./rtcStatsService"; import { MediaPrefs, SignalRTCSessionDescription } from "./types"; +import { RtcStatsConnection } from "./rtcStatsService"; import { P2PIncrementAnalyticMetric } from "./P2pRtcManager"; import { trackAnnotations } from "../utils/annotations"; import { type ConnectionStatus } from "../model"; @@ -19,6 +19,7 @@ interface P2PSessionOptions { deprioritizeH264Encoding: boolean; incrementAnalyticMetric: P2PIncrementAnalyticMetric; mediaPrefs?: MediaPrefs; + rtcStats: RtcStatsConnection; } export default class Session { @@ -48,6 +49,7 @@ export default class Session { srdComplete?: ReturnType; _incrementAnalyticMetric: P2PIncrementAnalyticMetric; pendingReplaceTrackActions: (() => Promise)[]; + _rtcStats: RtcStatsConnection; constructor({ clientId, @@ -56,6 +58,7 @@ export default class Session { deprioritizeH264Encoding, incrementAnalyticMetric, mediaPrefs, + rtcStats, }: P2PSessionOptions) { this.relayCandidateSeen = false; this.serverReflexiveCandidateSeen = false; @@ -66,6 +69,7 @@ export default class Session { this.mdnsHostCandidateSeen = false; this.pendingReplaceTrackActions = []; this._mediaPrefs = mediaPrefs; + this._rtcStats = rtcStats; // Create PC. this.peerConnectionConfig = peerConnectionConfig; @@ -219,7 +223,7 @@ export default class Session { this.pc.signalingState, ); this._incrementAnalyticMetric("P2PStaleAnswerIgnored"); - rtcStats.sendEvent("P2PStaleAnswerIgnored", { + this._rtcStats.sendEvent("P2PStaleAnswerIgnored", { clientId: this.clientId, signalingState: this.pc.signalingState, }); @@ -252,7 +256,7 @@ export default class Session { this.pc.addIceCandidate(candidate).catch((e: any) => { logger.warn("Failed to add ICE candidate ('%s'): %s", candidate ? candidate.candidate : null, e); this._incrementAnalyticMetric("P2PAddIceCandidateFailure"); - rtcStats.sendEvent("P2PAddIceCandidateFailure", { + this._rtcStats.sendEvent("P2PAddIceCandidateFailure", { clientId: this.clientId, errorName: e?.name, errorMessage: e?.message, @@ -325,7 +329,7 @@ export default class Session { let stream = this.streams.find((s) => s.getTracks().find((t) => t.id === newTrack.id)); if (!stream) { // This check was here from before, let's see if it ever happens. - rtcStats.sendEvent("P2PReplaceTrackNewTrackNotInStream", { + this._rtcStats.sendEvent("P2PReplaceTrackNewTrackNotInStream", { oldTrackId: oldTrack?.id, oldTrackKind: oldTrack?.kind, oldTrackIsEffect: oldTrack && trackAnnotations(oldTrack).isEffectTrack, @@ -338,7 +342,7 @@ export default class Session { // TODO: Don't depend on array index for the stream. stream = this.streams[0]; if (!stream) { - rtcStats.sendEvent("P2PReplaceTrackNoStream", {}); + this._rtcStats.sendEvent("P2PReplaceTrackNoStream", {}); this._incrementAnalyticMetric("P2PReplaceTrackNoStream"); throw new Error("replaceTrack: No stream?"); } diff --git a/packages/media/src/webrtc/VegaConnection.ts b/packages/media/src/webrtc/VegaConnection.ts index e26031d22..cb97f0106 100644 --- a/packages/media/src/webrtc/VegaConnection.ts +++ b/packages/media/src/webrtc/VegaConnection.ts @@ -2,8 +2,8 @@ import SfuV2Parser from "./SfuV2Parser"; import { EventEmitter } from "events"; import Logger from "../utils/Logger"; import { VegaIncrementAnalyticMetric } from "./VegaRtcManager/types"; -import rtcStats from "./rtcStatsService"; import { VegaConnectionOptions } from "./types"; +import { RtcStatsConnection } from "./rtcStatsService"; const logger = new Logger(); @@ -13,13 +13,19 @@ export default class VegaConnection extends EventEmitter { socket: WebSocket | null = null; sents: Map; incrementAnalyticMetric?: VegaIncrementAnalyticMetric; + rtcStats: RtcStatsConnection; - constructor(wsUrl: string, { protocol = "whereby-sfu#v4", incrementAnalyticMetric }: VegaConnectionOptions) { + constructor( + wsUrl: string, + rtcStatsConnection: RtcStatsConnection, + { protocol = "whereby-sfu#v4", incrementAnalyticMetric }: VegaConnectionOptions, + ) { super(); this.incrementAnalyticMetric = incrementAnalyticMetric; this.wsUrl = wsUrl; this.protocol = protocol; + this.rtcStats = rtcStatsConnection; // This is the map of sent requests that are waiting for a response this.sents = new Map(); @@ -87,7 +93,7 @@ export default class VegaConnection extends EventEmitter { // If an SFU response arrive after timeout it's already deleted. logger.warn(`Received unknown message with id ${socketMessage.id} from SFU.`); this.incrementAnalyticMetric?.("vegaUnknownResponse"); - rtcStats.sendEvent("VegaUnknownResponse", { + this.rtcStats.sendEvent("VegaUnknownResponse", { id: socketMessage.id, code: socketMessage.errorCode, reason: socketMessage.errorReason, @@ -139,7 +145,7 @@ export default class VegaConnection extends EventEmitter { }, timer: setTimeout(() => { this.incrementAnalyticMetric?.("vegaRequestTimeout"); - rtcStats.sendEvent("VegaRequestTimeout", { + this.rtcStats.sendEvent("VegaRequestTimeout", { id: request.id, method: request.method, }); diff --git a/packages/media/src/webrtc/VegaConnectionManager.ts b/packages/media/src/webrtc/VegaConnectionManager.ts index 0de07950d..26c111fc2 100644 --- a/packages/media/src/webrtc/VegaConnectionManager.ts +++ b/packages/media/src/webrtc/VegaConnectionManager.ts @@ -1,3 +1,4 @@ +import { RtcStatsConnection } from "./rtcStatsService"; import VegaConnection from "./VegaConnection"; import { VegaIncrementAnalyticMetric } from "./VegaRtcManager/types"; @@ -47,14 +48,17 @@ export function convertToProperHostList(hostList: string | HostListEntryOptional } // responsible for checking and nominating (delivering) 1 working sfu websocket connection // it will use a prioritized list of hosts to try -export function createVegaConnectionManager(config: { - initialHostList: string | any[]; - getUrlForHost?: (host: string) => string; - onConnected?: (vegaConnection: VegaConnection, info: ConnectionInfo) => void; - onDisconnected?: () => void; - onFailed?: () => void; - onAttemptFailed?: (info: { host: string; dc: string }) => void; -}) { +export function createVegaConnectionManager( + rtcStats: RtcStatsConnection, + config: { + initialHostList: string | any[]; + getUrlForHost?: (host: string) => string; + onConnected?: (vegaConnection: VegaConnection, info: ConnectionInfo) => void; + onDisconnected?: () => void; + onFailed?: () => void; + onAttemptFailed?: (info: { host: string; dc: string }) => void; + }, +) { let connectionInfo: ConnectionInfo; let lastDisconnectTime: number | undefined; let lastNetworkUpTime: number | undefined; @@ -165,7 +169,7 @@ export function createVegaConnectionManager(config: { return; } - const vegaConnection = new VegaConnection(config.getUrlForHost?.(host) || host, { + const vegaConnection = new VegaConnection(config.getUrlForHost?.(host) || host, rtcStats, { protocol: "whereby-sfu#v4", incrementAnalyticMetric, }); diff --git a/packages/media/src/webrtc/VegaRtcManager/__tests__/index.spec.ts b/packages/media/src/webrtc/VegaRtcManager/__tests__/index.spec.ts index 2623de929..12c7378a2 100644 --- a/packages/media/src/webrtc/VegaRtcManager/__tests__/index.spec.ts +++ b/packages/media/src/webrtc/VegaRtcManager/__tests__/index.spec.ts @@ -6,6 +6,7 @@ import { MockTransport, MockProducer } from "../../../../tests/webrtc/webRtcHelp import WS from "jest-websocket-mock"; import { setTimeout } from "timers/promises"; import { GetConstraintsOptions, WebRTCProvider } from "../../types"; +import { RtcStatsConnection } from "../../rtcStatsService"; jest.mock("../../../utils/getMediasoupDevice"); const { getMediasoupDeviceAsync } = jest.requireMock("../../../utils/getMediasoupDevice"); @@ -31,6 +32,7 @@ describe("VegaRtcManager", () => { let sfuWebsocketServerUrl: string; let mockSendTransport: MockTransport; + let rtcStatsConnectionStub: RtcStatsConnection; beforeEach(() => { mediaConstraints = { @@ -54,6 +56,7 @@ describe("VegaRtcManager", () => { getMediaOptions: () => mediaConstraints, }; mockSendTransport = new MockTransport(); + rtcStatsConnectionStub = helpers.createRtcStatsConnectionStub(); emitter = helpers.createEmitterStub(); @@ -100,6 +103,7 @@ describe("VegaRtcManager", () => { webrtcProvider, features: {}, eventClaim: helpers.randomString("/claim-"), + rtcStats: rtcStatsConnectionStub, }); }); @@ -139,6 +143,7 @@ describe("VegaRtcManager", () => { serverSocket, webrtcProvider, features: { isNodeSdk: true }, + rtcStats: rtcStatsConnectionStub, }); expect(getMediasoupDeviceAsync).toHaveBeenCalledWith({ isNodeSdk: true }); diff --git a/packages/media/src/webrtc/VegaRtcManager/index.ts b/packages/media/src/webrtc/VegaRtcManager/index.ts index de55467cd..2efeb89f7 100644 --- a/packages/media/src/webrtc/VegaRtcManager/index.ts +++ b/packages/media/src/webrtc/VegaRtcManager/index.ts @@ -3,7 +3,7 @@ import adapterRaw from "webrtc-adapter"; import { v4 as uuidv4 } from "uuid"; import rtcManagerEvents from "../rtcManagerEvents"; -import rtcStats from "../rtcStatsService"; +import type { RtcStatsConnection } from "../rtcStatsService"; import createMicAnalyser from "../VegaMicAnalyser"; import { AddCameraStreamOptions, @@ -131,8 +131,18 @@ export default class VegaRtcManager implements RtcManager { _networkIsDetectedUpBySignal: boolean; _cpuOveruseDetected: boolean; analytics: VegaAnalytics; - - constructor({ selfId, room, emitter, serverSocket, webrtcProvider, features, eventClaim }: VegaRtcManagerOptions) { + _rtcStats: RtcStatsConnection; + + constructor({ + selfId, + room, + emitter, + serverSocket, + webrtcProvider, + features, + eventClaim, + rtcStats, + }: VegaRtcManagerOptions) { const { session, iceServers, turnServers, sfuServer, mediaserverConfigTtlSeconds } = room; this._selfId = selfId; @@ -143,6 +153,7 @@ export default class VegaRtcManager implements RtcManager { this._webrtcProvider = webrtcProvider; this._features = features || {}; this._eventClaim = eventClaim; + this._rtcStats = rtcStats; this._vegaConnection = null; @@ -199,12 +210,12 @@ export default class VegaRtcManager implements RtcManager { // One of them is getting unplugged. The other is the Chrome audio // process crashing. The third is the tab being closed. // https://bugs.chromium.org/p/chromium/issues/detail?id=1050008 - rtcStats.sendEvent("audio_ended", { unloading }); + this._rtcStats.sendEvent("audio_ended", { unloading }); this._emitToPWA(rtcManagerEvents.MICROPHONE_STOPPED_WORKING, {}); this.analytics.micTrackEndedCount++; }; this._videoTrackOnEnded = () => { - rtcStats.sendEvent("video_ended", { unloading }); + this._rtcStats.sendEvent("video_ended", { unloading }); this._emitToPWA(rtcManagerEvents.CAMERA_STOPPED_WORKING, {}); this.analytics.camTrackEndedCount++; }; @@ -373,7 +384,7 @@ export default class VegaRtcManager implements RtcManager { _connect() { if (this._isConnectingOrConnected) { // TEMP investigation event for COB-2771 — remove once understood - rtcStats.sendEvent("SfuConnectEarlyBail", { + this._rtcStats.sendEvent("SfuConnectEarlyBail", { readyState: this._vegaConnection?.socket?.readyState, bufferedAmount: this._vegaConnection?.socket?.bufferedAmount, }); @@ -400,7 +411,7 @@ export default class VegaRtcManager implements RtcManager { this._features.sfuServerOverrideHost || this._sfuServer?.url; - this._vegaConnectionManager = createVegaConnectionManager({ + this._vegaConnectionManager = createVegaConnectionManager(this._rtcStats, { initialHostList: hostList, getUrlForHost: (host) => { const searchParams = new URLSearchParams({ @@ -434,7 +445,7 @@ export default class VegaRtcManager implements RtcManager { this._onClose(); }, onAttemptFailed: ({ host, dc }) => { - rtcStats.sendEvent("SfuConnectAttemptFailed", { host, dc }); + this._rtcStats.sendEvent("SfuConnectAttemptFailed", { host, dc }); }, }); } @@ -461,19 +472,19 @@ export default class VegaRtcManager implements RtcManager { this._qualityMonitor.close(); this._emitToPWA(rtcManagerEvents.SFU_CONNECTION_CLOSED); - rtcStats.sendEvent("SfuConnectionClosed", {}); + this._rtcStats.sendEvent("SfuConnectionClosed", {}); } async _join() { logger.info("_join()"); this._emitToPWA(rtcManagerEvents.SFU_CONNECTION_OPEN); - rtcStats.sendEvent("SfuConnectionOpened", {}); + this._rtcStats.sendEvent("SfuConnectionOpened", {}); try { if (!this._vegaConnection) { logger.error("_join() No VegaConnection found"); this.analytics.vegaJoinWithoutVegaConnection++; - rtcStats.sendEvent("JoinWithoutVegaConnection", {}); + this._rtcStats.sendEvent("JoinWithoutVegaConnection", {}); throw new Error("No VegaConnection found"); } // We need to always do this, as this triggers the join logic on the SFU @@ -516,7 +527,7 @@ export default class VegaRtcManager implements RtcManager { } catch (error) { logger.error("_join() [error:%o]", error); this.analytics.vegaJoinFailed++; - rtcStats.sendEvent("VegaJoinFailed", { error }); + this._rtcStats.sendEvent("VegaJoinFailed", { error }); // TODO: handle this error, rejoin? } } @@ -525,7 +536,7 @@ export default class VegaRtcManager implements RtcManager { if (!this._vegaConnection) { logger.error("_createTransport() No VegaConnection found"); this.analytics.vegaCreateTransportWithoutVegaConnection++; - rtcStats.sendEvent("CreateTransportWithoutVegaConnection", {}); + this._rtcStats.sendEvent("CreateTransportWithoutVegaConnection", {}); throw new Error("No VegaConnection found"); } const creator = send ? "createSendTransport" : "createRecvTransport"; @@ -829,7 +840,7 @@ export default class VegaRtcManager implements RtcManager { if (this._micPaused !== this._micProducer.paused) this._pauseResumeMic(); } catch (error) { this.analytics.vegaMicProducerFailed++; - rtcStats.sendEvent("VegaMicProducerFailed", { error }); + this._rtcStats.sendEvent("VegaMicProducerFailed", { error }); logger.error("micProducer failed:%o", error); } finally { this._micProducerPromise = null; @@ -837,7 +848,7 @@ export default class VegaRtcManager implements RtcManager { // Has the track disappeared while we were waiting to be executed? if (!this._micTrack) { this.analytics.vegaMicProducerClosed++; - rtcStats.sendEvent("VegaMicProducerClosed", {}); + this._rtcStats.sendEvent("VegaMicProducerClosed", {}); this._stopProducer(this._micProducer); this._micProducer = null; } @@ -999,7 +1010,7 @@ export default class VegaRtcManager implements RtcManager { const targetMaxSpatialLayer = simulcastLayer3ShouldBeActive ? 2 : 1; if (this._webcamProducer.maxSpatialLayer !== targetMaxSpatialLayer) { this._webcamProducer.setMaxSpatialLayer(targetMaxSpatialLayer); - rtcStats.sendEvent("simulcast_layer_activation_changed", { + this._rtcStats.sendEvent("simulcast_layer_activation_changed", { layerIndex: 2, active: simulcastLayer3ShouldBeActive, }); @@ -1048,7 +1059,7 @@ export default class VegaRtcManager implements RtcManager { ? addProducerCpuOveruseWatch({ producer, onOveruse: () => { - rtcStats.sendEvent("producer_cpuoveruse_detected", {}); + this._rtcStats.sendEvent("producer_cpuoveruse_detected", {}); // we stop monitoring once we detect cleanUpCpuWatch(); @@ -1083,7 +1094,7 @@ export default class VegaRtcManager implements RtcManager { if (this._webcamPaused !== this._webcamProducer.paused) this._pauseResumeWebcam(); } catch (error) { this.analytics.vegaWebcamProducerFailed++; - rtcStats.sendEvent("VegaWebcamProducerFailed", { error }); + this._rtcStats.sendEvent("VegaWebcamProducerFailed", { error }); logger.error("webcamProducer failed:%o", error); } finally { this._webcamProducerPromise = null; @@ -1110,7 +1121,7 @@ export default class VegaRtcManager implements RtcManager { if (!this._webcamProducer && (!this._webcamTrack || !this._webcamTrack.enabled)) { this.analytics.vegaReplaceTrackNoProducerNoEnabledTrack++; - rtcStats.sendEvent("VegaReplaceTrackNoProducerNoEnabledTrack", { + this._rtcStats.sendEvent("VegaReplaceTrackNoProducerNoEnabledTrack", { hasWebcamTrack: !!this._webcamTrack, }); } @@ -1213,7 +1224,7 @@ export default class VegaRtcManager implements RtcManager { if (this._screenVideoTrack !== this._screenVideoProducer.track) await this._replaceScreenVideoTrack(); } catch (error) { this.analytics.vegaScreenVideoProducerFailed++; - rtcStats.sendEvent("VegaScreenVideoProducerFailed", { error }); + this._rtcStats.sendEvent("VegaScreenVideoProducerFailed", { error }); logger.error("screenVideoProducer failed:%o", error); } finally { this._screenVideoProducerPromise = null; @@ -1294,7 +1305,7 @@ export default class VegaRtcManager implements RtcManager { if (this._screenAudioTrack !== this._screenAudioProducer.track) await this._replaceScreenAudioTrack(); } catch (error) { this.analytics.vegaScreenAudioProducerFailed++; - rtcStats.sendEvent("VegaScreenAudioProducerFailed", { error }); + this._rtcStats.sendEvent("VegaScreenAudioProducerFailed", { error }); logger.error("screenAudioProducer failed:%o", error); } finally { this._screenAudioProducerPromise = null; @@ -1393,7 +1404,7 @@ export default class VegaRtcManager implements RtcManager { this._syncMicAnalyser(); - rtcStats.sendEvent("colocation_changed", { colocation }); + this._rtcStats.sendEvent("colocation_changed", { colocation }); } /** @@ -1814,30 +1825,30 @@ export default class VegaRtcManager implements RtcManager { } sendAudioMutedStats(muted: boolean) { - rtcStats.sendEvent("audio_muted", { muted }); + this._rtcStats.sendEvent("audio_muted", { muted }); } sendVideoMutedStats(muted: boolean) { - rtcStats.sendEvent("video_muted", { muted }); + this._rtcStats.sendEvent("video_muted", { muted }); } sendStatsCustomEvent(eventName: string, data?: any) { - rtcStats.sendEvent(eventName, data); + this._rtcStats.sendEvent(eventName, data); } rtcStatsConnect() { - if (!rtcStats.server.connected) { - rtcStats.server.connect(); + if (!this._rtcStats.server.connected) { + this._rtcStats.server.connect(); } } rtcStatsDisconnect() { - rtcStats.server.close(); + this._rtcStats.server.close(); } rtcStatsReconnect() { - if (!rtcStats.server.connected && rtcStats.server.attemptedConnectedAtLeastOnce) { - rtcStats.server.connect(); + if (!this._rtcStats.server.connected && this._rtcStats.server.attemptedConnectedAtLeastOnce) { + this._rtcStats.server.connect(); } } @@ -1902,7 +1913,7 @@ export default class VegaRtcManager implements RtcManager { consumer = await this._receiveTransport.consume(options); } catch (error) { this.analytics.vegaConsumerCreationFailed++; - rtcStats.sendEvent("VegaConsumerCreationFailed", { producerId: options.producerId, error }); + this._rtcStats.sendEvent("VegaConsumerCreationFailed", { producerId: options.producerId, error }); throw error; } diff --git a/packages/media/src/webrtc/__tests__/rtcStatsService.spec.ts b/packages/media/src/webrtc/__tests__/rtcStatsService.spec.ts new file mode 100644 index 000000000..e33c5383f --- /dev/null +++ b/packages/media/src/webrtc/__tests__/rtcStatsService.spec.ts @@ -0,0 +1,265 @@ +import { mockWebSocketConstructor, randomString } from "../../../tests/webrtc/webRtcHelpers"; +import { RtcStatsConnection } from "../rtcStatsService"; +import rtcstats from "@whereby.com/rtcstats"; + +jest.mock("@whereby.com/rtcstats"); + +describe("rtcStatsService", () => { + describe("rtcStats connection", () => { + const MOCK_DATE = new Date(); + const baseUrl = "wss://rtcstats.test"; + const roomName = "/some-room"; + + let resetDelta: jest.Func; + let rtcStatsConnection: RtcStatsConnection; + let originalLocation: typeof window.location; + let originalWebSocket: typeof window.WebSocket; + let WebSocketConstructor: ReturnType; + + function currentWebSocket() { + const current = WebSocketConstructor.mock.instances.at(-1); + + if (!current) { + throw new Error("No websocket found!"); + } + return current; + } + + function rtcStatsConnected() { + const webSocket = currentWebSocket(); + + // @ts-ignore + webSocket.readyState = WebSocket.OPEN; + webSocket.onopen?.(new Event("opened")); + } + + function rtcStatsDisconnected() { + const webSocket = currentWebSocket(); + + // @ts-ignore + webSocket.readyState = WebSocket.CLOSED; + webSocket.onclose?.(new CloseEvent("closed")); + } + + beforeEach(() => { + jest.useFakeTimers(); + jest.setSystemTime(MOCK_DATE); + + resetDelta = jest.fn(); + jest.mocked(rtcstats).mockReturnValue({ resetDelta }); + + rtcStatsConnection = new RtcStatsConnection({ url: baseUrl, getStatsBufferSize: 2 }); + + // Replace global location object with mock + originalLocation = window.location; + const mockLocation = { + ...originalLocation, + pathname: roomName, + }; + + Object.defineProperty(window, "location", { + configurable: true, + value: mockLocation, + writable: true, + }); + + // Replace WebSocket class with mock + originalWebSocket = window.WebSocket; + WebSocketConstructor = mockWebSocketConstructor(); + Object.defineProperty(window, "WebSocket", { + configurable: true, + value: WebSocketConstructor, + writable: true, + }); + }); + + afterEach(() => { + jest.useRealTimers(); + + // Restore original location + Object.defineProperty(window, "location", { + configurable: true, + value: originalLocation, + writable: true, + }); + + // Restore original WebSocket + Object.defineProperty(window, "WebSocket", { + configurable: true, + value: originalWebSocket, + writable: true, + }); + }); + + it("should initially have clientInfo with connectionNumber 0", () => { + expect(rtcStatsConnection.clientInfo.connectionNumber).toEqual(0); + }); + + it("should set up to trace getstats every 10 seconds", () => { + expect(rtcstats).toHaveBeenCalledWith(rtcStatsConnection.server.trace, 10000, [""]); + }); + + describe("initial connection", () => { + it("should create websocket with correct url and protocol", () => { + rtcStatsConnection.server.connect(); + + expect(WebSocket).toHaveBeenCalledWith(baseUrl + roomName, "1.0"); + }); + + it("should send clientInfo once connected", () => { + rtcStatsConnection.server.connect(); + rtcStatsConnected(); + + expect(rtcStatsConnection.clientInfo.connectionNumber).toEqual(1); + expect(currentWebSocket().send).toHaveBeenCalledWith( + JSON.stringify(["clientInfo", null, rtcStatsConnection.clientInfo]), + ); + }); + + it("should send buffered events once connected", () => { + const preConnectEvents = [ + { event: ["customEvent", null, { type: "test", value: randomString() }], buffered: true }, + { event: ["customEvent", null, { type: "organizationId", value: randomString() }], buffered: true }, + { event: ["some", "this"], buffered: true }, + // insightsStats should not be buffered + { event: ["customEvent", null, { type: "insightsStats", value: randomString() }], buffered: false }, + { event: ["getstats", randomString()], buffered: true }, + { event: ["getstats", randomString()], buffered: true }, + // we have set a getStats buffer of 2, the third one should be ignored + { event: ["getstats", randomString()], buffered: false }, + ]; + preConnectEvents.forEach(({ event }) => rtcStatsConnection.server.trace(...event)); + + rtcStatsConnection.server.connect(); + rtcStatsConnected(); + + preConnectEvents.forEach(({ event, buffered }) => { + const expectFn = buffered ? expect(currentWebSocket().send) : expect(currentWebSocket().send).not; + expectFn.toHaveBeenCalledWith(JSON.stringify([...event, MOCK_DATE.getTime()])); + }); + }); + }); + + describe("when connected", () => { + beforeEach(() => { + rtcStatsConnection.server.connect(); + rtcStatsConnected(); + }); + + it("sendEvent should send customEvent and value", () => { + const type = randomString("customEventName"); + const value = randomString("customEventData"); + + rtcStatsConnection.sendEvent(type, value); + + expect(currentWebSocket().send).toHaveBeenCalledWith( + JSON.stringify(["customEvent", null, { type, value }, MOCK_DATE.getTime()]), + ); + }); + + it("sendAudioMuted() should send correct payload", () => { + const muted = true; + + rtcStatsConnection.sendAudioMuted(muted); + + expect(currentWebSocket().send).toHaveBeenCalledWith( + JSON.stringify([ + "customEvent", + null, + { type: "audio_muted", value: { muted } }, + MOCK_DATE.getTime(), + ]), + ); + }); + + it("sendVideoMuted() should send correct payload", () => { + const muted = true; + + rtcStatsConnection.sendVideoMuted(muted); + + expect(currentWebSocket().send).toHaveBeenCalledWith( + JSON.stringify([ + "customEvent", + null, + { type: "video_muted", value: { muted } }, + MOCK_DATE.getTime(), + ]), + ); + }); + + describe("cached events", () => { + it.each([ + "clientId", + "organizationId", + "displayName", + "userRole", + "deviceId", + "roomProduct", + "roomMode", + "sfuServer", + "featureFlags", + ])("should re-send %s on subsequenct connects", (eventType) => { + const value = randomString(); + const expectedPayload = JSON.stringify([ + "customEvent", + null, + { type: eventType, value }, + MOCK_DATE.getTime(), + ]); + + // Send once on the initial connection + rtcStatsConnection.sendEvent(eventType, value); + + // Willingly close and re-open the connection + rtcStatsConnection.server.close(); + rtcStatsDisconnected(); + rtcStatsConnection.server.connect(); + rtcStatsConnected(); + + // Verify the same message is sent on both connections + expect(WebSocketConstructor.mock.instances[0].send).toHaveBeenCalledWith(expectedPayload); + expect(WebSocketConstructor.mock.instances[1].send).toHaveBeenCalledWith(expectedPayload); + }); + }); + + it("should schedule opening new connection when dropped and send buffered events on re-open", () => { + const bufferedType = randomString("bufferedType"); + const bufferedValue = randomString("bufferedValue"); + + // Simulate connection drop + rtcStatsDisconnected(); + + // Send event while disconnected + rtcStatsConnection.sendEvent(bufferedType, bufferedValue); + + // Wait for a second and simulate websocket connecting + jest.advanceTimersByTime(1000); + rtcStatsConnected(); + + expect(WebSocketConstructor.mock.instances[0].close).toHaveBeenCalledTimes(1); + expect(WebSocketConstructor.mock.instances[1].send).toHaveBeenCalledWith( + JSON.stringify([ + "customEvent", + null, + { type: bufferedType, value: bufferedValue }, + MOCK_DATE.getTime(), + ]), + ); + }); + + it("should close webSocket when disconnecting", () => { + rtcStatsConnection.server.close(); + rtcStatsDisconnected(); + + expect(currentWebSocket().close).toHaveBeenCalled(); + }); + + it("should reset delta when disconnected", () => { + rtcStatsConnection.server.close(); + rtcStatsDisconnected(); + + expect(resetDelta).toHaveBeenCalled(); + }); + }); + }); +}); diff --git a/packages/media/src/webrtc/index.ts b/packages/media/src/webrtc/index.ts index 51576d210..09e4d017d 100644 --- a/packages/media/src/webrtc/index.ts +++ b/packages/media/src/webrtc/index.ts @@ -14,7 +14,7 @@ export { default as P2pRtcManager } from "./P2pRtcManager"; export { default as RtcManagerDispatcher } from "./RtcManagerDispatcher"; export { default as rtcManagerEvents } from "./rtcManagerEvents"; export * from "./rtcrtpsenderHelper"; -export { default as rtcStats } from "./rtcStatsService"; +export * from "./rtcStatsService"; export * from "./sdpModifier"; export { default as Session } from "./Session"; export { default as SfuV2Parser } from "./SfuV2Parser"; diff --git a/packages/media/src/webrtc/rtcStatsService.ts b/packages/media/src/webrtc/rtcStatsService.ts index 1ad329afe..187183633 100644 --- a/packages/media/src/webrtc/rtcStatsService.ts +++ b/packages/media/src/webrtc/rtcStatsService.ts @@ -14,16 +14,19 @@ const RTCSTATS_PROTOCOL_VERSION = "1.0"; // as they are delta compressed and we need the initial properties const GETSTATS_BUFFER_SIZE = 20; -const clientInfo = { - id: uuidv4(), // shared id across rtcstats reconnects - connectionNumber: 0, -}; - const noop = () => {}; -let resetDelta = noop; - // Inlined version of rtcstats/trace-ws with improved disconnect handling. -function rtcStatsConnection(wsURL: string, logger: any = console) { +function rtcStatsConnection({ + url, + clientInfo, + getStatsBufferSize, + logger = console, +}: { + url: string; + clientInfo: any; + getStatsBufferSize: number; + logger?: any; +}) { const buffer: any = []; let ws: WebSocket; let organizationId: number; @@ -44,6 +47,7 @@ function rtcStatsConnection(wsURL: string, logger: any = console) { const connection = { connected: false, attemptedConnectedAtLeastOnce: false, + resetDelta: noop, trace: (...args: any) => { args.push(Date.now()); @@ -88,7 +92,7 @@ function rtcStatsConnection(wsURL: string, logger: any = console) { } else if (args[0] === "getstats") { // only buffer getStats for a while // we don't want this to pile up, but we need at least the initial reports - if (getStatsBufferUsed < GETSTATS_BUFFER_SIZE) { + if (getStatsBufferUsed < getStatsBufferSize) { getStatsBufferUsed++; buffer.push(args); } @@ -118,7 +122,7 @@ function rtcStatsConnection(wsURL: string, logger: any = console) { ws?.close(); connection.connected = true; connection.attemptedConnectedAtLeastOnce = true; - ws = new WebSocket(wsURL + window.location.pathname, RTCSTATS_PROTOCOL_VERSION); + ws = new WebSocket(url + window.location.pathname, RTCSTATS_PROTOCOL_VERSION); ws.onerror = (e: Event) => { connection.connected = false; @@ -127,7 +131,7 @@ function rtcStatsConnection(wsURL: string, logger: any = console) { ws.onclose = (e: CloseEvent) => { connection.connected = false; logger.info(`[RTCSTATS] Closed ${e.code}`); - resetDelta(); + connection.resetDelta(); }; ws.onopen = () => { // send client info after each connection, so analysis tools can handle reconnections @@ -178,29 +182,47 @@ function rtcStatsConnection(wsURL: string, logger: any = console) { return connection; } -const RTCSTATS_URL = process.env.RTCSTATS_URL; -const server = rtcStatsConnection(RTCSTATS_URL || "wss://rtcstats.srv.whereby.com"); -const stats = rtcstats( - server.trace, - 10000, // query once every 10 seconds. - [""], // only shim unprefixed RTCPeerConnecion. -); -// on node clients this function can be undefined -resetDelta = stats?.resetDelta || noop; - -const rtcStats = { - sendEvent: (type: any, value: any) => { - server.trace("customEvent", null, { +export class RtcStatsConnection { + public clientInfo: { id: string; connectionNumber: number }; + public server: ReturnType; + + constructor({ + url, + getStatsBufferSize = GETSTATS_BUFFER_SIZE, + getStatsInterval = 10000, + }: { + url: string; + getStatsBufferSize?: number; + getStatsInterval?: number; + }) { + this.clientInfo = { + id: uuidv4(), // shared id across rtcstats reconnects + connectionNumber: 0, + }; + + this.server = rtcStatsConnection({ url, clientInfo: this.clientInfo, getStatsBufferSize }); + + const stats = rtcstats( + this.server.trace, + getStatsInterval, + [""], // only shim unprefixed RTCPeerConnecion. + ); + // on node clients this function can be undefined + this.server.resetDelta = stats?.resetDelta || noop; + } + + public sendEvent(type: any, value: any) { + this.server.trace("customEvent", null, { type, value, }); - }, - sendAudioMuted: (muted: boolean) => { - rtcStats.sendEvent("audio_muted", { muted }); - }, - sendVideoMuted: (muted: boolean) => { - rtcStats.sendEvent("video_muted", { muted }); - }, - server, -}; -export default rtcStats; + } + + public sendAudioMuted(muted: boolean) { + this.sendEvent("audio_muted", { muted }); + } + + public sendVideoMuted(muted: boolean) { + this.sendEvent("video_muted", { muted }); + } +} diff --git a/packages/media/src/webrtc/sdpModifier.ts b/packages/media/src/webrtc/sdpModifier.ts index b0f7156a4..afe2d385b 100644 --- a/packages/media/src/webrtc/sdpModifier.ts +++ b/packages/media/src/webrtc/sdpModifier.ts @@ -2,8 +2,6 @@ import SDPUtils from "sdp"; import adapterRaw from "webrtc-adapter"; import * as sdpTransform from "sdp-transform"; import Logger from "../utils/Logger"; -import { P2PIncrementAnalyticMetric } from "./P2pRtcManager"; -import rtcStats from "./rtcStatsService"; // @ts-ignore const adapter = adapterRaw.default ?? adapterRaw; diff --git a/packages/media/src/webrtc/stats/IssueMonitor/__tests__/index.spec.ts b/packages/media/src/webrtc/stats/IssueMonitor/__tests__/index.spec.ts index e5be1bb95..31d4bc00c 100644 --- a/packages/media/src/webrtc/stats/IssueMonitor/__tests__/index.spec.ts +++ b/packages/media/src/webrtc/stats/IssueMonitor/__tests__/index.spec.ts @@ -1,9 +1,11 @@ import { issueDetectorOrMetricEnabled, Metric, subscribeIssues } from ".."; import { createMockedMediaStreamTrack, + createRtcStatsConnectionStub, mockCheckData, mockStatsClient, } from "../../../../../tests/webrtc/webRtcHelpers"; +import { RtcStatsConnection } from "../../../rtcStatsService"; import { setClientProvider } from "../../StatsMonitor"; import { setPeerConnectionsForTests } from "../../StatsMonitor/peerConnectionTracker"; import { StatsClient } from "../../types"; @@ -142,6 +144,7 @@ describe("IssueMonitor", () => { }); let onUpdatedIssues: jest.Mock; + let rtcStatsConnectionStub: RtcStatsConnection; const statsIntervalTick = async () => { jest.advanceTimersByTime(2000); @@ -163,6 +166,7 @@ describe("IssueMonitor", () => { beforeEach(() => { onUpdatedIssues = jest.fn(); + rtcStatsConnectionStub = createRtcStatsConnectionStub(); localCam = createMockClient("localcam", true); remoteCam1 = createMockClient("remotecam1", false); @@ -175,7 +179,7 @@ describe("IssueMonitor", () => { setClientProvider(() => [localCam, remoteCam1, remoteCam2]); - stopSubscription = subscribeIssues({ onUpdatedIssues }).stop; + stopSubscription = subscribeIssues({ onUpdatedIssues }, rtcStatsConnectionStub).stop; }); afterEach(async () => { diff --git a/packages/media/src/webrtc/stats/IssueMonitor/index.ts b/packages/media/src/webrtc/stats/IssueMonitor/index.ts index 13af4251c..73c206dfd 100644 --- a/packages/media/src/webrtc/stats/IssueMonitor/index.ts +++ b/packages/media/src/webrtc/stats/IssueMonitor/index.ts @@ -1,6 +1,7 @@ import { IssueCheckData, IssueDetector, issueDetectors } from "./issueDetectors"; import { subscribeStats } from "../StatsMonitor"; import { RenderedDimensionsReport, StatsClient, ViewStats } from "../types"; +import { RtcStatsConnection } from "../../rtcStatsService"; type IssueSubscription = { onUpdatedIssues: ( @@ -507,11 +508,12 @@ function onUpdatedStats( ); } -export function subscribeIssues(subscription: IssueSubscription): { stop: () => void } { +export function subscribeIssues(subscription: IssueSubscription, rtcStats: RtcStatsConnection): { stop: () => void } { + if (!rtcStats) throw new Error("rtcStats connection must be passed to `subscribeIssues`"); subscriptions.push(subscription); // start the stats on first subscription - if (!stopStats) stopStats = subscribeStats({ onUpdatedStats }).stop; + if (!stopStats) stopStats = subscribeStats({ onUpdatedStats }, rtcStats).stop; return { stop() { diff --git a/packages/media/src/webrtc/stats/StatsMonitor/__tests__/index.spec.ts b/packages/media/src/webrtc/stats/StatsMonitor/__tests__/index.spec.ts index 027347599..b17cbdb18 100644 --- a/packages/media/src/webrtc/stats/StatsMonitor/__tests__/index.spec.ts +++ b/packages/media/src/webrtc/stats/StatsMonitor/__tests__/index.spec.ts @@ -1,4 +1,6 @@ import { StatsMonitorOptions, StatsMonitorState, subscribeStats } from ".."; +import { createRtcStatsConnectionStub } from "../../../../../tests/webrtc/webRtcHelpers"; +import { RtcStatsConnection } from "../../../rtcStatsService"; import { collectStats } from "../collectStats"; import { startCpuObserver } from "../cpuObserver"; @@ -10,6 +12,7 @@ jest.useFakeTimers(); describe("subscribeStats", () => { let options: StatsMonitorOptions; let baseState: StatsMonitorState; + let rtcStatsConnectionStub: RtcStatsConnection; beforeEach(() => { baseState = { @@ -22,6 +25,7 @@ describe("subscribeStats", () => { renderedDimensionsByTrack: {}, }; options = { interval: 2000, logger: { debug: jest.fn(), error: jest.fn(), info: jest.fn(), warn: jest.fn() } }; + rtcStatsConnectionStub = createRtcStatsConnectionStub(); }); describe("with current monitor", () => { @@ -35,7 +39,7 @@ describe("subscribeStats", () => { const state = { ...baseState, currentMonitor }; const subscription = { onUpdatedStats: jest.fn() }; - subscribeStats(subscription, options, state); + subscribeStats(subscription, rtcStatsConnectionStub, options, state); expect(state.subscriptions).toEqual([subscription]); }); @@ -44,7 +48,7 @@ describe("subscribeStats", () => { const state = { ...baseState, currentMonitor }; const subscription = { onUpdatedStats: jest.fn() }; - subscribeStats(subscription, options, state).stop(); + subscribeStats(subscription, rtcStatsConnectionStub, options, state).stop(); expect(state.subscriptions).toEqual([]); expect(currentMonitor?.stop).toHaveBeenCalled(); @@ -57,7 +61,7 @@ describe("subscribeStats", () => { const state = { ...baseState }; const subscription = { onUpdatedStats: jest.fn() }; - subscribeStats(subscription, options, state); + subscribeStats(subscription, rtcStatsConnectionStub, options, state); expect(startCpuObserver).toHaveBeenCalled(); }); @@ -66,7 +70,7 @@ describe("subscribeStats", () => { const state = { ...baseState }; const subscription = { onUpdatedStats: jest.fn() }; - subscribeStats(subscription, options, state); + subscribeStats(subscription, rtcStatsConnectionStub, options, state); jest.advanceTimersByTime(options.interval); expect(collectStats).toHaveBeenCalled(); @@ -76,7 +80,7 @@ describe("subscribeStats", () => { const state = { ...baseState, currentMonitor: null }; const subscription = { onUpdatedStats: jest.fn() }; - subscribeStats(subscription, options, state); + subscribeStats(subscription, rtcStatsConnectionStub, options, state); expect(state.currentMonitor).toMatchObject({ getUpdatedStats: expect.any(Function), @@ -90,7 +94,7 @@ describe("subscribeStats", () => { (startCpuObserver as jest.Mock).mockReturnValueOnce({ stop }); const subscription = { onUpdatedStats: jest.fn() }; - subscribeStats(subscription, options, state).stop(); + subscribeStats(subscription, rtcStatsConnectionStub, options, state).stop(); expect(stop).toHaveBeenCalled(); }); diff --git a/packages/media/src/webrtc/stats/StatsMonitor/__tests__/peerConnection.spec.ts b/packages/media/src/webrtc/stats/StatsMonitor/__tests__/peerConnection.spec.ts index 0348cb43a..46a9efeba 100644 --- a/packages/media/src/webrtc/stats/StatsMonitor/__tests__/peerConnection.spec.ts +++ b/packages/media/src/webrtc/stats/StatsMonitor/__tests__/peerConnection.spec.ts @@ -2,13 +2,19 @@ import { createMockedMediaStreamTrack, createRTCPeerConnectionStub, createRTCTrancieverStub, + createRtcStatsConnectionStub, } from "../../../../../tests/webrtc/webRtcHelpers"; import { PCData } from "../../types"; +import { RtcStatsConnection } from "../../../rtcStatsService"; import { getPeerConnectionsWithStatsReports } from "../peerConnection"; import { setPeerConnectionsForTests } from "../peerConnectionTracker"; describe("peerConnection", () => { describe("getPeerConnectionsWithStatsReports", () => { + let rtcStatsConnectionStub: RtcStatsConnection; + beforeEach(() => { + rtcStatsConnectionStub = createRtcStatsConnectionStub(); + }); it("should return correct amount of peer connections", async () => { const pcStats = [new Map(), new Map(), new Map()]; const existingPeerConnections = [ @@ -18,7 +24,7 @@ describe("peerConnection", () => { ]; setPeerConnectionsForTests(existingPeerConnections); - const result = await getPeerConnectionsWithStatsReports(); + const result = await getPeerConnectionsWithStatsReports(rtcStatsConnectionStub); result.forEach(({ pc, report }, index) => { expect(pc).toEqual(existingPeerConnections[index]); @@ -32,7 +38,7 @@ describe("peerConnection", () => { const existingPeerConnections = [peerConnection]; setPeerConnectionsForTests(existingPeerConnections); - const result = await getPeerConnectionsWithStatsReports(); + const result = await getPeerConnectionsWithStatsReports(rtcStatsConnectionStub); existingPeerConnections.forEach((_, index) => { expect(result[index].report).toEqual(new Map()); @@ -141,9 +147,11 @@ describe("peerConnection", () => { const [ { + pc: _resultPc, + report: _resultReport, pcData: { ssrcToTrackId }, }, - ] = await getPeerConnectionsWithStatsReports(pcDataByPc); + ] = await getPeerConnectionsWithStatsReports(rtcStatsConnectionStub, pcDataByPc); expect(ssrcToTrackId).toEqual(expectedSsrcTrackMappings); }); diff --git a/packages/media/src/webrtc/stats/StatsMonitor/collectStats.ts b/packages/media/src/webrtc/stats/StatsMonitor/collectStats.ts index c49bb39cb..fd03c84c4 100644 --- a/packages/media/src/webrtc/stats/StatsMonitor/collectStats.ts +++ b/packages/media/src/webrtc/stats/StatsMonitor/collectStats.ts @@ -9,7 +9,7 @@ import { import { getPeerConnectionsWithStatsReports } from "./peerConnection"; import { getPeerConnectionIndex, removePeerConnection } from "./peerConnectionTracker"; import { RenderedDimensionsReport, StatsClient, ViewStats } from "../types"; -import rtcStats from "../../rtcStatsService"; +import type { RtcStatsConnection } from "../../rtcStatsService"; const getOrCreateSsrcMetricsContainer = ( statsByView: Record, @@ -84,9 +84,10 @@ const DEFAULT_CLIENT: StatsClient = { export async function collectStats( state: StatsMonitorState, { logger, interval }: StatsMonitorOptions, + rtcStats: RtcStatsConnection, immediate: boolean, ): Promise | undefined> { - const collectStatsBound = collectStats.bind(null, state, { interval, logger }); + const collectStatsBound = collectStats.bind(null, state, { interval, logger }, rtcStats); try { // refresh provided clients before each run @@ -118,7 +119,7 @@ export async function collectStats( state.lastUpdateTime = Date.now(); // loop through current peer connections - (await getPeerConnectionsWithStatsReports()).forEach(({ pc, report, pcData }) => { + (await getPeerConnectionsWithStatsReports(rtcStats)).forEach(({ pc, report, pcData }) => { // each new peer connection will get +1, to be able to see/count/correlate data const pcIndex = getPeerConnectionIndex(pc); diff --git a/packages/media/src/webrtc/stats/StatsMonitor/index.ts b/packages/media/src/webrtc/stats/StatsMonitor/index.ts index 573253cb3..0ef60c010 100644 --- a/packages/media/src/webrtc/stats/StatsMonitor/index.ts +++ b/packages/media/src/webrtc/stats/StatsMonitor/index.ts @@ -3,6 +3,7 @@ import { PressureObserver, startCpuObserver } from "./cpuObserver"; import { numFailedTrackSsrcLookups, numMissingTrackSsrcLookups } from "./peerConnection"; import { PressureRecord, StatsClient, ViewStats, RenderedDimensionsReport } from "../types"; +import type { RtcStatsConnection } from "../../rtcStatsService"; import Logger from "../../../utils/Logger"; interface StatsMonitor { @@ -69,8 +70,12 @@ export const updateRenderedDimensions = (trackId: string, data: RenderedDimensio STATE.renderedDimensionsByTrack[trackId] = data; }; -function startStatsMonitor(state: StatsMonitorState, { interval, logger }: StatsMonitorOptions) { - const collectStatsBound = collectStats.bind(null, state, { interval, logger }); +function startStatsMonitor( + state: StatsMonitorState, + { interval, logger }: StatsMonitorOptions, + rtcStats: RtcStatsConnection, +) { + const collectStatsBound = collectStats.bind(null, state, { interval, logger }, rtcStats); let cpuObserver: ReturnType; @@ -96,13 +101,15 @@ function startStatsMonitor(state: StatsMonitorState, { interval, logger }: Stats export function subscribeStats( subscription: StatsSubscription, + rtcStats: RtcStatsConnection, options: StatsMonitorOptions = OPTIONS, state: StatsMonitorState = STATE, ) { + if (!rtcStats) throw new Error("rtcStats connection must be passed to `subscribeStats`"); state.subscriptions.push(subscription); // start the monitor on first subscription - if (!state.currentMonitor) state.currentMonitor = startStatsMonitor(state, options); + if (!state.currentMonitor) state.currentMonitor = startStatsMonitor(state, options, rtcStats); return { stop() { diff --git a/packages/media/src/webrtc/stats/StatsMonitor/peerConnection.ts b/packages/media/src/webrtc/stats/StatsMonitor/peerConnection.ts index 6005c3468..61fb8e331 100644 --- a/packages/media/src/webrtc/stats/StatsMonitor/peerConnection.ts +++ b/packages/media/src/webrtc/stats/StatsMonitor/peerConnection.ts @@ -1,5 +1,5 @@ -import rtcStats from "../../rtcStatsService"; import { PCData } from "../types"; +import { RtcStatsConnection } from "../../rtcStatsService"; import { getCurrentPeerConnections } from "./peerConnectionTracker"; // peer connection related data @@ -7,7 +7,7 @@ const PC_DATA_BY_PC = new WeakMap(); export let numMissingTrackSsrcLookups = 0; export let numFailedTrackSsrcLookups = 0; -export const getPeerConnectionsWithStatsReports = (pcDataByPc = PC_DATA_BY_PC) => +export const getPeerConnectionsWithStatsReports = (rtcStats: RtcStatsConnection, pcDataByPc = PC_DATA_BY_PC) => Promise.all( getCurrentPeerConnections().map(async (pc: RTCPeerConnection) => { let pcData = pcDataByPc.get(pc); diff --git a/packages/media/src/webrtc/types.ts b/packages/media/src/webrtc/types.ts index 7a7bfb4ba..31378f806 100644 --- a/packages/media/src/webrtc/types.ts +++ b/packages/media/src/webrtc/types.ts @@ -1,4 +1,5 @@ import { SignalRoom, ServerSocket } from "../utils"; +import { RtcStatsConnection } from "./rtcStatsService"; import { VegaIncrementAnalyticMetric } from "./VegaRtcManager/types"; export enum RtcEventNames { @@ -11,6 +12,7 @@ export type RtcEventEmitter = { emit: (eventName: K, export interface RtcManagerOptions { selfId: string; room: SignalRoom; + rtcStats: RtcStatsConnection; emitter: RtcEventEmitter; serverSocket: ServerSocket; webrtcProvider: WebRTCProvider; diff --git a/packages/media/tests/webrtc/P2pRtcManager.spec.ts b/packages/media/tests/webrtc/P2pRtcManager.spec.ts index 98e49cbde..80c2c9a00 100644 --- a/packages/media/tests/webrtc/P2pRtcManager.spec.ts +++ b/packages/media/tests/webrtc/P2pRtcManager.spec.ts @@ -5,8 +5,6 @@ jest.mock("webrtc-adapter", () => { }); import assert from "node:assert"; -import rtcStats from "../../src/webrtc/rtcStatsService"; - import * as helpers from "./webRtcHelpers"; import * as CONNECTION_STATUS from "../../src/model/connectionStatusConstants"; import P2pRtcManager, { ICE_RESTART_DELAY } from "../../src/webrtc/P2pRtcManager"; @@ -14,6 +12,7 @@ import rtcManagerEvents from "../../src/webrtc/rtcManagerEvents"; import { RELAY_MESSAGES, PROTOCOL_REQUESTS, PROTOCOL_RESPONSES } from "../../src/model/protocol"; import { CAMERA_STREAM_ID, GetConstraintsOptions, WebRTCProvider } from "../../src"; +import { RtcStatsConnection } from "../../src"; const originalNavigator = global.navigator; @@ -49,6 +48,7 @@ describe("P2pRtcManager", () => { let mediaConstraints: GetConstraintsOptions; let rtcManager: P2pRtcManager; let mediaserverConfigTtlSeconds: number; + let rtcStatsConnectionStub: RtcStatsConnection; beforeEach(() => { jest.useFakeTimers(); @@ -74,6 +74,7 @@ describe("P2pRtcManager", () => { emitterStub = helpers.createEmitterStub(); iceServers = helpers.createIceServersConfig(); clientId = helpers.randomString("client-"); + rtcStatsConnectionStub = helpers.createRtcStatsConnectionStub(); navigator = { mediaDevices: { @@ -131,6 +132,7 @@ describe("P2pRtcManager", () => { serverSocket: _serverSocket, webrtcProvider, features, + rtcStats: rtcStatsConnectionStub, }); } @@ -477,10 +479,10 @@ describe("P2pRtcManager", () => { }); it("should disconnect rtcStats", () => { - jest.spyOn(rtcStats.server, "close"); + jest.spyOn(rtcStatsConnectionStub.server, "close"); rtcManager.disconnectAll(); - expect(rtcStats.server.close).toHaveBeenCalled(); + expect(rtcStatsConnectionStub.server.close).toHaveBeenCalled(); }); }); diff --git a/packages/media/tests/webrtc/RtcManagerDispatcher.spec.ts b/packages/media/tests/webrtc/RtcManagerDispatcher.spec.ts index ac8e5c0fa..cfd83d9e9 100644 --- a/packages/media/tests/webrtc/RtcManagerDispatcher.spec.ts +++ b/packages/media/tests/webrtc/RtcManagerDispatcher.spec.ts @@ -8,6 +8,7 @@ import * as CONNECTION_STATUS from "../../src/model/connectionStatusConstants"; import { EventEmitter } from "events"; import { v4 as uuidv4 } from "uuid"; import { GetConstraintsOptions, WebRTCProvider } from "../../src"; +import { RtcStatsConnection } from "../../src"; jest.mock("mediasoup-client", () => ({ Device: jest.fn(), @@ -18,6 +19,7 @@ describe("RtcManagerDispatcher", () => { let emitter: any; let serverSocketStub: any; let webrtcProvider: WebRTCProvider; + let rtcStatsConnectionStub: RtcStatsConnection; const features: any = {}; beforeEach(() => { @@ -30,8 +32,9 @@ describe("RtcManagerDispatcher", () => { }; const serverSocket = serverSocketStub.socket; + rtcStatsConnectionStub = helpers.createRtcStatsConnectionStub(); - new RtcManagerDispatcher({ emitter, serverSocket, webrtcProvider, features }); + new RtcManagerDispatcher({ emitter, serverSocket, webrtcProvider, features, rtcStats: rtcStatsConnectionStub }); }); function mockEmitRoomJoined({ diff --git a/packages/media/tests/webrtc/Session.spec.ts b/packages/media/tests/webrtc/Session.spec.ts index 547ce58d5..d901a4a0f 100644 --- a/packages/media/tests/webrtc/Session.spec.ts +++ b/packages/media/tests/webrtc/Session.spec.ts @@ -1,4 +1,4 @@ -import { Session } from "../../src"; +import { RtcStatsConnection, Session } from "../../src"; import { trackAnnotations } from "../../src/utils/annotations"; import * as helpers from "./webRtcHelpers"; @@ -21,18 +21,22 @@ describe("Session", () => { let session: Session; let bandwidth: number; let peerConnectionConfig: RTCConfiguration; + let rtcStatsConnectionStub: RtcStatsConnection; beforeEach(() => { // @ts-ignore window.globalThis.RTCPeerConnection = RtcPeerConnection; bandwidth = 1; peerConnectionConfig = {}; + rtcStatsConnectionStub = helpers.createRtcStatsConnectionStub(); + session = new Session({ clientId, peerConnectionConfig, bandwidth, deprioritizeH264Encoding: false, incrementAnalyticMetric: jest.fn(), + rtcStats: rtcStatsConnectionStub, }); }); @@ -76,6 +80,7 @@ describe("Session", () => { deprioritizeH264Encoding: false, incrementAnalyticMetric: jest.fn(), mediaPrefs: { wantsVideo: false }, + rtcStats: rtcStatsConnectionStub, }); const stream = helpers.createMockedMediaStream(); const audioTrack = stream.getAudioTracks()[0]; @@ -128,6 +133,7 @@ describe("Session", () => { deprioritizeH264Encoding: false, incrementAnalyticMetric: jest.fn(), mediaPrefs: { wantsVideo: false }, + rtcStats: rtcStatsConnectionStub, }); }); @@ -304,6 +310,7 @@ describe("Session", () => { deprioritizeH264Encoding: false, incrementAnalyticMetric: jest.fn(), mediaPrefs: { wantsVideo: false }, + rtcStats: rtcStatsConnectionStub, }); const tracksAddedToPC: MediaStreamTrack[] = []; diff --git a/packages/media/tests/webrtc/VegaConnectionManager.spec.ts b/packages/media/tests/webrtc/VegaConnectionManager.spec.ts index 10a3c9f39..edae3cbd7 100644 --- a/packages/media/tests/webrtc/VegaConnectionManager.spec.ts +++ b/packages/media/tests/webrtc/VegaConnectionManager.spec.ts @@ -1,6 +1,8 @@ import { convertToProperHostList, createVegaConnectionManager } from "../../src/webrtc/VegaConnectionManager"; import VegaConnection from "../../src/webrtc/VegaConnection"; import { EventEmitter } from "events"; +import { RtcStatsConnection } from "../../src"; +import * as helpers from "./webRtcHelpers"; describe("convertToProperHostList", () => { it("converts a serialized string of hosts to an array of hosts", () => { @@ -47,8 +49,11 @@ jest.mock("../../src/webrtc/VegaConnection.ts", () => { }); describe("createVegaConnectionManager", () => { + let rtcStatsConnectionStub: RtcStatsConnection; + beforeEach(() => { jest.useFakeTimers(); + rtcStatsConnectionStub = helpers.createRtcStatsConnectionStub(); }); afterEach(() => { jest.useRealTimers(); @@ -57,7 +62,7 @@ describe("createVegaConnectionManager", () => { it("works sending a single host as a string", () => { const onConnected = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: "host1 open:100", onConnected, }); @@ -75,7 +80,7 @@ describe("createVegaConnectionManager", () => { it("works sending a multiple hosts as a string", () => { const onConnected = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: "host1 close:100,host2 open:100", onConnected, }); @@ -92,7 +97,7 @@ describe("createVegaConnectionManager", () => { }); it("tries connecting to hosts in order provided with delays between them", () => { - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1", dc: "a" }, { host: "host2", dc: "b" }, @@ -102,19 +107,19 @@ describe("createVegaConnectionManager", () => { connect(); jest.advanceTimersToNextTimer(); expect(VegaConnection).toHaveBeenCalledTimes(1); - expect(VegaConnection).toHaveBeenNthCalledWith(1, "host1", { + expect(VegaConnection).toHaveBeenNthCalledWith(1, "host1", rtcStatsConnectionStub, { protocol: "whereby-sfu#v4", incrementAnalyticMetric: undefined, }); jest.advanceTimersToNextTimer(); expect(VegaConnection).toHaveBeenCalledTimes(2); - expect(VegaConnection).toHaveBeenNthCalledWith(2, "host2", { + expect(VegaConnection).toHaveBeenNthCalledWith(2, "host2", rtcStatsConnectionStub, { protocol: "whereby-sfu#v4", incrementAnalyticMetric: undefined, }); jest.advanceTimersToNextTimer(); expect(VegaConnection).toHaveBeenCalledTimes(3); - expect(VegaConnection).toHaveBeenNthCalledWith(3, "host3", { + expect(VegaConnection).toHaveBeenNthCalledWith(3, "host3", rtcStatsConnectionStub, { protocol: "whereby-sfu#v4", incrementAnalyticMetric: undefined, }); @@ -122,7 +127,7 @@ describe("createVegaConnectionManager", () => { it("emits successful connection and doesn't attempt to connect to other hosts after if fast enough", () => { const onConnected = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1", dc: "a" }, { host: "host2 open:100", dc: "b" }, @@ -145,7 +150,7 @@ describe("createVegaConnectionManager", () => { it("tries multiple hosts in parallel if slow, but only emits the one that opens first (even if all opens)", () => { const onConnected = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1 open:7000", dc: "a" }, { host: "host2 open:5000", dc: "b" }, @@ -169,7 +174,7 @@ describe("createVegaConnectionManager", () => { it("emits failed if all hosts fail", () => { const onConnected = jest.fn(); const onFailed = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1 close:100", dc: "a" }, { host: "host2 close:100", dc: "b" }, @@ -189,7 +194,7 @@ describe("createVegaConnectionManager", () => { it("doesn't run multiple times if called while in progress, if one of the hosts connect", () => { const onConnected = jest.fn(); const onFailed = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1 close:100", dc: "a" }, { host: "host2 close:100", dc: "b" }, @@ -214,7 +219,7 @@ describe("createVegaConnectionManager", () => { it("will run (only) once more, if connect() is called again while first round is in progress, if all fails, so hostlist or network can change followed by a new connect()", () => { const onConnected = jest.fn(); const onFailed = jest.fn(); - const { connect, updateHostList } = createVegaConnectionManager({ + const { connect, updateHostList } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1 close:100", dc: "a" }, { host: "host2 close:100", dc: "b" }, @@ -251,7 +256,7 @@ describe("createVegaConnectionManager", () => { it("will treat a close shortly after open as a failed connection", () => { const onConnected = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1", dc: "a" }, { host: "host2 open:100 close:150", dc: "b" }, // failed connection @@ -276,7 +281,7 @@ describe("createVegaConnectionManager", () => { const onConnected = jest.fn(); const onDisconnected = jest.fn(); const onFailed = jest.fn(); - const { connect } = createVegaConnectionManager({ + const { connect } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1", dc: "a" }, { host: "host2 open:100 close:5000", dc: "b" }, // successful connection, closes later @@ -304,7 +309,7 @@ describe("createVegaConnectionManager", () => { it("will retry the last connected host shortly after network problems, the old or updated hostlist after a while", () => { const onConnected = jest.fn(); const onDisconnected = jest.fn(); - const { connect, updateHostList } = createVegaConnectionManager({ + const { connect, updateHostList } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "badhost1 close:100", dc: "a" }, { host: "goodhost1 open:100 close:5000", dc: "b" }, @@ -370,23 +375,26 @@ describe("createVegaConnectionManager", () => { const onDisconnected = jest.fn(); const onFailed = jest.fn(); - const { connect, updateHostList, networkIsPossiblyDown, networkIsUp } = createVegaConnectionManager({ - initialHostList: [ - { host: "badhost1 close:1000", dc: "a" }, // this is "simulating" the network problem - { host: "goodhost1 open:2000", dc: "b" }, - { host: "goodhost2 open:3000", dc: "c" }, - { host: "goodhost3 open:3000", dc: "c" }, - { host: "goodhost4 open:3000", dc: "c" }, - { host: "goodhost5 open:3000", dc: "c" }, - { host: "goodhost6 open:3000", dc: "c" }, - { host: "goodhost7 open:3000", dc: "c" }, - { host: "goodhost8 open:3000", dc: "c" }, - { host: "goodhost9 open:3000", dc: "c" }, - ], - onConnected, - onDisconnected, - onFailed, - }); + const { connect, updateHostList, networkIsPossiblyDown, networkIsUp } = createVegaConnectionManager( + rtcStatsConnectionStub, + { + initialHostList: [ + { host: "badhost1 close:1000", dc: "a" }, // this is "simulating" the network problem + { host: "goodhost1 open:2000", dc: "b" }, + { host: "goodhost2 open:3000", dc: "c" }, + { host: "goodhost3 open:3000", dc: "c" }, + { host: "goodhost4 open:3000", dc: "c" }, + { host: "goodhost5 open:3000", dc: "c" }, + { host: "goodhost6 open:3000", dc: "c" }, + { host: "goodhost7 open:3000", dc: "c" }, + { host: "goodhost8 open:3000", dc: "c" }, + { host: "goodhost9 open:3000", dc: "c" }, + ], + onConnected, + onDisconnected, + onFailed, + }, + ); connect(); jest.advanceTimersByTime(300); networkIsPossiblyDown(); // network goes down before goodhost1 is scheduled @@ -409,7 +417,7 @@ describe("createVegaConnectionManager", () => { it("will provide analytics, updated and aggregated across multiple calls to connect()", () => { const onConnected = jest.fn(); - const { connect, updateHostList } = createVegaConnectionManager({ + const { connect, updateHostList } = createVegaConnectionManager(rtcStatsConnectionStub, { initialHostList: [ { host: "host1 close:100", dc: "a" }, { host: "host2 open:100 close:5000", dc: "b" }, diff --git a/packages/media/tests/webrtc/webRtcHelpers.ts b/packages/media/tests/webrtc/webRtcHelpers.ts index 93d7fd95c..4edb285ba 100644 --- a/packages/media/tests/webrtc/webRtcHelpers.ts +++ b/packages/media/tests/webrtc/webRtcHelpers.ts @@ -3,6 +3,7 @@ import WS from "jest-websocket-mock"; import { ServerSocket } from "../../src/utils/ServerSocket"; import { SsrcStats, StatsClient, TrackStats, ViewStats } from "../../src/webrtc/stats/types"; import { IssueCheckData } from "../../src/webrtc/stats/IssueMonitor/issueDetectors"; +import { RtcStatsConnection } from "../../src"; jest.mock("../../src/utils/ServerSocket"); export function createServerSocketStub() { @@ -54,6 +55,15 @@ export function createServerSocketStub() { }; } +export function createRtcStatsConnectionStub() { + return { + sendEvent: jest.fn(), + sendAudioMuted: jest.fn(), + sendVideoMuted: jest.fn(), + server: { connect: jest.fn(), close: jest.fn(), trace: jest.fn() }, + } as unknown as RtcStatsConnection; +} + export function createRTCPeerConnectionStub( { receivers = [], @@ -326,3 +336,27 @@ export function mockCheckData(args?: Partial): IssueCheckData { ...args, }; } + +// INFO: The websocket mocks are very incomplete, only added for rtcstats tests +export function mockWebSocket(): WebSocket { + return { + close: jest.fn(), + send: jest.fn(), + } as unknown as WebSocket; +} + +export function mockWebSocketConstructor() { + const ClassMock: jest.Mock = jest.fn().mockImplementation(function () { + // @ts-ignore + Object.assign(this, mockWebSocket()); + + // IMPORTANT! Dont return anything here to allow mock.instances to be populated + }); + + // @ts-ignore + ClassMock.CLOSED = "CLOSED"; + // @ts-ignore + ClassMock.OPEN = "OPEN"; + + return ClassMock; +} From 629bf255f09cadd644df0d35fe00039c8174a719 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A5vard=20Holvik?= Date: Wed, 1 Jul 2026 21:57:16 +0300 Subject: [PATCH 2/2] core: Use new major media version, adopt rtcStats integration Make the rtcConnection slice initialize a fresh rtcConnection before passing it into the media functions which requires this. --- .changeset/sour-mangos-slide.md | 8 ++ .../src/redux/slices/connectionMonitor.ts | 77 +++++++++++-------- .../src/redux/slices/rtcConnection/index.ts | 18 +++-- packages/core/src/redux/tests/store.setup.ts | 23 +++++- .../redux/tests/store/cameraEffects.spec.ts | 7 +- .../tests/store/connectionMonitor.spec.ts | 12 --- .../src/redux/tests/store/localMedia.spec.ts | 1 + .../redux/tests/store/rtcConnection.spec.ts | 2 - .../tests/store/signalConnection.spec.ts | 9 --- 9 files changed, 91 insertions(+), 66 deletions(-) create mode 100644 .changeset/sour-mangos-slide.md diff --git a/.changeset/sour-mangos-slide.md b/.changeset/sour-mangos-slide.md new file mode 100644 index 000000000..20234de76 --- /dev/null +++ b/.changeset/sour-mangos-slide.md @@ -0,0 +1,8 @@ +--- +"@whereby.com/core": minor +--- + +Initialize rtcstats connection in rtcConnection slice. + +Due to breaking change in media package, we need to ensure that the rtcstats connection +is initialized outside the media package itself. diff --git a/packages/core/src/redux/slices/connectionMonitor.ts b/packages/core/src/redux/slices/connectionMonitor.ts index a67777e0c..fa13fcb42 100644 --- a/packages/core/src/redux/slices/connectionMonitor.ts +++ b/packages/core/src/redux/slices/connectionMonitor.ts @@ -4,7 +4,7 @@ import { RootState } from "../store"; import { createAppThunk } from "../thunk"; import { createReactor } from "../listenerMiddleware"; import { selectRoomConnectionStatus } from "./roomConnection/selectors"; -import { selectRtcManager } from "./rtcConnection"; +import { selectRtcManager, selectRtcStatsConnection } from "./rtcConnection"; import { selectAllClientViews } from "./room"; /** @@ -50,6 +50,12 @@ export const connectionMonitorSlice = createSlice({ export const { connectionMonitorStarted, connectionMonitorStopped } = connectionMonitorSlice.actions; export const doStartConnectionMonitor = createAppThunk(() => (dispatch, getState) => { + const rtcStatsConnection = selectRtcStatsConnection(getState()); + + if (!rtcStatsConnection) { + return; + } + setClientProvider(() => { const state = getState(); @@ -72,40 +78,43 @@ export const doStartConnectionMonitor = createAppThunk(() => (dispatch, getState return clientViews; }); - const issueMonitorSubscription = subscribeIssues({ - onUpdatedIssues: (issuesAndMetricsByClients) => { - const state = getState(); - const rtcManager = selectRtcManager(state); - - if (!rtcManager) { - return; - } - - // gather a selection of aggregate metrics - let lossSend = 0; - let lossReceive = 0; - let bitrateSend = 0; - let bitrateReceive = 0; - Object.entries(issuesAndMetricsByClients.aggregated.metrics).forEach( - ([key, value]: [string, { curMax: number; curSum: number }]) => { - if (/loc.*packetloss/.test(key)) lossSend = Math.max(lossSend, value.curMax); - if (/rem.*packetloss/.test(key)) lossReceive = Math.max(lossReceive, value.curMax); - if (/loc.*bitrate/.test(key)) bitrateSend += value.curSum; - if (/rem.*bitrate/.test(key)) bitrateReceive += value.curSum; - }, - ); - - // send the selection of aggregate metrics to rtcstats - rtcManager.sendStatsCustomEvent("insightsStats", { - ls: Math.round(lossSend * 1000) / 1000, - lr: Math.round(lossReceive * 1000) / 1000, - bs: Math.round(bitrateSend), - br: Math.round(bitrateReceive), - cpu: issuesAndMetricsByClients.aggregated.metrics["global-cpu-pressure"]?.curSum, - _time: Date.now(), - }); + const issueMonitorSubscription = subscribeIssues( + { + onUpdatedIssues: (issuesAndMetricsByClients) => { + const state = getState(); + const rtcManager = selectRtcManager(state); + + if (!rtcManager) { + return; + } + + // gather a selection of aggregate metrics + let lossSend = 0; + let lossReceive = 0; + let bitrateSend = 0; + let bitrateReceive = 0; + Object.entries(issuesAndMetricsByClients.aggregated.metrics).forEach( + ([key, value]: [string, { curMax: number; curSum: number }]) => { + if (/loc.*packetloss/.test(key)) lossSend = Math.max(lossSend, value.curMax); + if (/rem.*packetloss/.test(key)) lossReceive = Math.max(lossReceive, value.curMax); + if (/loc.*bitrate/.test(key)) bitrateSend += value.curSum; + if (/rem.*bitrate/.test(key)) bitrateReceive += value.curSum; + }, + ); + + // send the selection of aggregate metrics to rtcstats + rtcManager.sendStatsCustomEvent("insightsStats", { + ls: Math.round(lossSend * 1000) / 1000, + lr: Math.round(lossReceive * 1000) / 1000, + bs: Math.round(bitrateSend), + br: Math.round(bitrateReceive), + cpu: issuesAndMetricsByClients.aggregated.metrics["global-cpu-pressure"]?.curSum, + _time: Date.now(), + }); + }, }, - }); + rtcStatsConnection, + ); dispatch(connectionMonitorStarted({ stopIssueSubscription: issueMonitorSubscription.stop })); }); diff --git a/packages/core/src/redux/slices/rtcConnection/index.ts b/packages/core/src/redux/slices/rtcConnection/index.ts index a27a0bec4..2437256f8 100644 --- a/packages/core/src/redux/slices/rtcConnection/index.ts +++ b/packages/core/src/redux/slices/rtcConnection/index.ts @@ -6,6 +6,7 @@ import { RtcManagerDispatcher, RtcEvents, RtcManagerCreatedPayload, + RtcStatsConnection, RtcStreamAddedPayload, RtcClientConnectionStatusChangedPayload, CAMERA_STREAM_ID, @@ -37,6 +38,8 @@ import { selectSpotlights } from "../spotlights"; import { rtcEvents } from "./actions"; export { rtcEvents } from "./actions"; +const RTCSTATS_URL = "wss://rtcstats.srv.whereby.com"; + export const createWebRtcEmitter = (dispatch: AppDispatch) => { return { emit: (eventName: keyof RtcEvents, data: RtcEvents[keyof RtcEvents]) => { @@ -74,6 +77,7 @@ export interface RtcConnectionState { }; rtcManager: RtcManager | null; rtcManagerDispatcher: RtcManagerDispatcher | null; + rtcStatsConnection: RtcStatsConnection; rtcManagerInitialized: boolean; status: "inactive" | "ready" | "reconnecting"; isAcceptingStreams: boolean; @@ -85,6 +89,7 @@ export const rtcConnectionSliceInitialState: RtcConnectionState = { isCreatingDispatcher: false, reportedStreamResolutions: {}, rtcManager: null, + rtcStatsConnection: new RtcStatsConnection({ url: RTCSTATS_URL }), rtcManagerDispatcher: null, rtcManagerInitialized: false, status: "inactive", @@ -196,6 +201,7 @@ export const doConnectRtc = createAppThunk(() => (dispatch, getState) => { const state = getState(); const socket = selectSignalConnectionRaw(state).socket; const dispatcher = selectRtcConnectionRaw(state).rtcManagerDispatcher; + const rtcStatsConnection = selectRtcStatsConnection(state); const isNodeSdk = selectAppIsNodeSdk(state); if (dispatcher || !socket) { @@ -209,6 +215,7 @@ export const doConnectRtc = createAppThunk(() => (dispatch, getState) => { const rtcManagerDispatcher = new RtcManagerDispatcher({ emitter: createWebRtcEmitter(dispatch), serverSocket: socket, + rtcStats: rtcStatsConnection, webrtcProvider, features: { isNodeSdk, @@ -250,10 +257,7 @@ export const doHandleAcceptStreams = createAppThunk((payload: StreamStatusUpdate for (const { clientId, streamId, state } of payload) { const participant = remoteClients.find((p) => p.id === clientId); if (!participant) continue; - if ( - state === "to_accept" || - (state === "new_accept" && shouldAcceptNewClients) - ) { + if (state === "to_accept" || (state === "new_accept" && shouldAcceptNewClients)) { rtcManager.acceptNewStream({ streamId: streamId === CAMERA_STREAM_ID ? clientId : streamId, clientId, @@ -311,7 +315,10 @@ export const doRtcManagerInitialize = createAppThunk(() => (dispatch, getState) const isMicrophoneEnabled = selectIsMicrophoneEnabled(getState()); if (localMediaStream && rtcManager) { - rtcManager.addCameraStream(localMediaStream, { audioPaused: !isMicrophoneEnabled, videoPaused: !isCameraEnabled }); + rtcManager.addCameraStream(localMediaStream, { + audioPaused: !isMicrophoneEnabled, + videoPaused: !isCameraEnabled, + }); } dispatch(rtcManagerInitialized()); @@ -328,6 +335,7 @@ export const selectRtcDispatcherCreated = (state: RootState) => state.rtcConnect export const selectRtcIsCreatingDispatcher = (state: RootState) => state.rtcConnection.isCreatingDispatcher; export const selectRtcStatus = (state: RootState) => state.rtcConnection.status; export const selectIsAcceptingStreams = (state: RootState) => state.rtcConnection.isAcceptingStreams; +export const selectRtcStatsConnection = (state: RootState) => state.rtcConnection.rtcStatsConnection; /** * Reactors diff --git a/packages/core/src/redux/tests/store.setup.ts b/packages/core/src/redux/tests/store.setup.ts index af328191e..1ebabfc0f 100644 --- a/packages/core/src/redux/tests/store.setup.ts +++ b/packages/core/src/redux/tests/store.setup.ts @@ -1,6 +1,26 @@ -import { RtcEvents } from "@whereby.com/media"; +import { RtcEvents, RtcStatsConnection } from "@whereby.com/media"; import { RootState, createStore as createRealStore } from "../store"; +jest.mock("@whereby.com/media", () => { + return { + ...jest.requireActual("@whereby.com/media"), + __esModule: true, + RtcStatsConnection: jest.fn().mockImplementation(() => { + return {}; + }), + RtcManagerDispatcher: jest.fn(), + ServerSocket: jest.fn().mockImplementation(() => { + return mockServerSocket; + }), + setClientProvider: jest.fn(), + subscribeIssues: jest.fn().mockImplementation(() => { + return { + stop: jest.fn(), + }; + }), + }; +}); + export const mockSignalEmit = jest.fn(); export const mockServerSocket = { on: jest.fn(), @@ -104,6 +124,7 @@ export function createStore({ initialState, withSignalConnection, withRtcManager // eslint-disable-next-line @typescript-eslint/no-explicit-any rtcManager: mockRtcManager as any, isAcceptingStreams: false, + rtcStatsConnection: new RtcStatsConnection({ url: "wss://rtcstats.test" }), ...initialState.rtcConnection, }; } diff --git a/packages/core/src/redux/tests/store/cameraEffects.spec.ts b/packages/core/src/redux/tests/store/cameraEffects.spec.ts index 6a82e4fa8..a3dfe8b89 100644 --- a/packages/core/src/redux/tests/store/cameraEffects.spec.ts +++ b/packages/core/src/redux/tests/store/cameraEffects.spec.ts @@ -30,6 +30,9 @@ jest.mock("@whereby.com/media", () => ({ replaced.forEach((t) => stream.removeTrack(t)); return replaced; }), + RtcStatsConnection: jest.fn().mockImplementation(() => { + return {}; + }), })); const mockEffectStop = jest.fn(); @@ -282,9 +285,7 @@ describe("cameraEffectsSlice", () => { store.dispatch(localMediaSlice.toggleCameraEnabled({ enabled: true })); await flushMicrotasks(); - const createEffectStream = jest.mocked( - jest.requireMock("@whereby.com/camera-effects").createEffectStream, - ); + const createEffectStream = jest.mocked(jest.requireMock("@whereby.com/camera-effects").createEffectStream); const state = store.getState().cameraEffects; expect(createEffectStream).toHaveBeenCalled(); expect(state.isPaused).toBe(false); diff --git a/packages/core/src/redux/tests/store/connectionMonitor.spec.ts b/packages/core/src/redux/tests/store/connectionMonitor.spec.ts index 319161d01..ae04b1707 100644 --- a/packages/core/src/redux/tests/store/connectionMonitor.spec.ts +++ b/packages/core/src/redux/tests/store/connectionMonitor.spec.ts @@ -2,18 +2,6 @@ import { createStore } from "../store.setup"; import { doStartConnectionMonitor, doStopConnectionMonitor } from "../../slices/connectionMonitor"; import { setClientProvider, subscribeIssues } from "@whereby.com/media"; -jest.mock("@whereby.com/media", () => { - return { - __esModule: true, - setClientProvider: jest.fn(), - subscribeIssues: jest.fn().mockImplementation(() => { - return { - stop: jest.fn(), - }; - }), - }; -}); - describe("connectionMonitorSlice", () => { describe("actions", () => { it("doStartConnectionMonitor", () => { diff --git a/packages/core/src/redux/tests/store/localMedia.spec.ts b/packages/core/src/redux/tests/store/localMedia.spec.ts index 2c5cdc447..f709f9340 100644 --- a/packages/core/src/redux/tests/store/localMedia.spec.ts +++ b/packages/core/src/redux/tests/store/localMedia.spec.ts @@ -28,6 +28,7 @@ jest.mock("@whereby.com/media", () => ({ __esModule: true, getStream: jest.fn(() => Promise.resolve()), getUpdatedDevices: jest.fn(() => Promise.resolve({ addedDevices: {}, changedDevices: {} })), + RtcStatsConnection: jest.fn().mockReturnValue({}), })); const mockedGetStream = jest.mocked(MediaDevices.getStream); diff --git a/packages/core/src/redux/tests/store/rtcConnection.spec.ts b/packages/core/src/redux/tests/store/rtcConnection.spec.ts index fd31b9488..fe1cb0299 100644 --- a/packages/core/src/redux/tests/store/rtcConnection.spec.ts +++ b/packages/core/src/redux/tests/store/rtcConnection.spec.ts @@ -15,8 +15,6 @@ import { diff } from "deep-object-diff"; import { coreVersion } from "../../../version"; import { doAppStop } from "../../slices/app"; -jest.mock("@whereby.com/media"); - describe("actions", () => { it("doHandleAcceptStreams", () => { const id1 = randomString("stream1"); diff --git a/packages/core/src/redux/tests/store/signalConnection.spec.ts b/packages/core/src/redux/tests/store/signalConnection.spec.ts index 11ca2525c..c7d989eb6 100644 --- a/packages/core/src/redux/tests/store/signalConnection.spec.ts +++ b/packages/core/src/redux/tests/store/signalConnection.spec.ts @@ -4,15 +4,6 @@ import { randomDeviceCredentials } from "../../../__mocks__/appMocks"; import { diff } from "deep-object-diff"; import { ServerSocket } from "@whereby.com/media"; -jest.mock("@whereby.com/media", () => { - return { - __esModule: true, - ServerSocket: jest.fn().mockImplementation(() => { - return mockServerSocket; - }), - }; -}); - describe("signalConnectionSlice", () => { describe("actions", () => { it("doSignalConnect", () => {