media: Replace rtcStats singleton with factory function

.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.

.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.

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 }));
 });

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

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,
         };
     }

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);

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", () => {

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);

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");

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", () => {

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));