[IMP] discuss: better bus types

Author: ThanhDodeurOdoo

src/client.ts

@@ -25,6 +25,7 @@ import type {
     AvailableFeatures,
     StartupData
 } from "#src/shared/types";
+import type { RequestMessage } from "#src/shared/bus-types";
 import type { TransportConfig, SessionId, SessionInfo } from "#src/models/session";
 
 interface Consumers {
@@ -273,24 +274,24 @@ export class SfuClient extends EventTarget {
         if (this.state !== SfuClientState.CONNECTED) {
             throw new Error("Cannot start recording when not connected");
         }
-        return this._bus?.request(
+        return this._bus!.request(
             {
                 name: CLIENT_REQUEST.START_RECORDING
             },
             { batch: true }
-        ) as Promise<boolean>;
+        );
     }
 
     async stopRecording(): Promise<boolean> {
         if (this.state !== SfuClientState.CONNECTED) {
             throw new Error("Cannot stop recording when not connected");
         }
-        return this._bus?.request(
+        return this._bus!.request(
             {
                 name: CLIENT_REQUEST.STOP_RECORDING
             },
             { batch: true }
-        ) as Promise<boolean>;
+        );
     }
 
     /**
@@ -531,10 +532,10 @@ export class SfuClient extends EventTarget {
         });
         transport.on("produce", async ({ kind, rtpParameters, appData }, callback, errback) => {
             try {
-                const result = (await this._bus!.request({
+                const result = await this._bus!.request({
                     name: CLIENT_REQUEST.INIT_PRODUCER,
                     payload: { type: appData.type as StreamType, kind, rtpParameters }
-                })) as { id: string };
+                });
                 callback({ id: result.id });
             } catch (error) {
                 errback(error as Error);
@@ -626,7 +627,10 @@ export class SfuClient extends EventTarget {
         }
     }
 
-    private async _handleRequest({ name, payload }: BusMessage): Promise<JSONSerializable | void> {
+    private async _handleRequest({
+        name,
+        payload
+    }: RequestMessage): Promise<JSONSerializable | void> {
         switch (name) {
             case SERVER_REQUEST.INIT_CONSUMER: {
                 const { id, kind, producerId, rtpParameters, sessionId, type, active } = payload;

src/models/session.ts

@@ -21,6 +21,7 @@ import {
     STREAM_TYPE
 } from "#src/shared/enums.ts";
 import type { BusMessage, JSONSerializable, StartupData, StreamType } from "#src/shared/types";
+import type { RequestMessage } from "#src/shared/bus-types";
 import type { Bus } from "#src/shared/bus.ts";
 import type { Channel } from "#src/models/channel.ts";
 import { RECORDER_STATE } from "#src/models/recorder.ts";
@@ -361,15 +362,15 @@ export class Session extends EventEmitter {
                 this._ctsTransport?.close();
                 this._stcTransport?.close();
             });
-            this._clientCapabilities = (await this.bus!.request({
+            this._clientCapabilities = await this.bus!.request({
                 name: SERVER_REQUEST.INIT_TRANSPORTS,
                 payload: {
                     capabilities: this._channel.router!.rtpCapabilities,
-                    stcConfig: this._createTransportConfig(this._stcTransport),
-                    ctsConfig: this._createTransportConfig(this._ctsTransport),
+                    stcConfig: this._createTransportConfig(this._stcTransport!),
+                    ctsConfig: this._createTransportConfig(this._ctsTransport!),
                     producerOptionsByKind: config.rtc.producerOptionsByKind
                 }
-            })) as RtpCapabilities;
+            });
             await Promise.all([
                 this._ctsTransport.setMaxIncomingBitrate(config.MAX_BITRATE_IN),
                 this._stcTransport.setMaxOutgoingBitrate(config.MAX_BITRATE_OUT)
@@ -628,7 +629,10 @@ export class Session extends EventEmitter {
         }
     }
 
-    private async _handleRequest({ name, payload }: BusMessage): Promise<JSONSerializable | void> {
+    private async _handleRequest({
+        name,
+        payload
+    }: RequestMessage): Promise<JSONSerializable | void> {
         switch (name) {
             case CLIENT_REQUEST.CONNECT_STC_TRANSPORT: {
                 const { dtlsParameters } = payload;

src/shared/bus-types.ts

@@ -0,0 +1,21 @@
+// eslint-disable-next-line node/no-unpublished-import
+import type { RtpCapabilities } from "mediasoup-client/lib/types";
+import type { CLIENT_REQUEST, SERVER_REQUEST } from "./enums";
+import type { BusMessage } from "./types";
+
+export interface RequestMap {
+    [CLIENT_REQUEST.CONNECT_CTS_TRANSPORT]: void;
+    [CLIENT_REQUEST.CONNECT_STC_TRANSPORT]: void;
+    [CLIENT_REQUEST.INIT_PRODUCER]: { id: string };
+    [CLIENT_REQUEST.START_RECORDING]: boolean;
+    [CLIENT_REQUEST.STOP_RECORDING]: boolean;
+    [SERVER_REQUEST.INIT_CONSUMER]: void;
+    [SERVER_REQUEST.INIT_TRANSPORTS]: RtpCapabilities;
+    [SERVER_REQUEST.PING]: void;
+}
+
+export type RequestName = keyof RequestMap;
+
+export type RequestMessage<T extends RequestName = RequestName> = Extract<BusMessage, { name: T }>;
+
+export type ResponseFrom<T extends RequestName> = RequestMap[T];

src/shared/bus.ts

@@ -1,9 +1,11 @@
 import type { WebSocket as NodeWebSocket } from "ws";
 
 import type { JSONSerializable, BusMessage } from "./types";
+import type { RequestMessage, RequestName, ResponseFrom } from "./bus-types";
+
 export interface Payload {
     /** The actual message content */
-    message: BusMessage;
+    message: BusMessage | JSONSerializable;
     /** Request ID if this message expects a response */
     needResponse?: string;
     /** Response ID if this message is responding to a request */
@@ -46,11 +48,9 @@ export class Bus {
     /** Global ID counter for Bus instances */
     private static _idCount = 0;
     /** Message handler for incoming messages */
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
     public onMessage?: (message: BusMessage) => void;
     /** Request handler for incoming requests */
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    public onRequest?: (request: BusMessage) => Promise<any | void>;
+    public onRequest?: (request: RequestMessage) => Promise<JSONSerializable | void>;
     /** Unique bus instance identifier */
     public readonly id: number = Bus._idCount++;
     /** Request counter for generating unique request IDs */
@@ -96,16 +96,22 @@ export class Bus {
     /**
      * Sends a request and waits for a response
      */
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    request(message: BusMessage, options: RequestOptions = {}): Promise<JSONSerializable> {
+    request<T extends RequestName>(
+        message: RequestMessage<T>,
+        options: RequestOptions = {}
+    ): Promise<ResponseFrom<T>> {
         const { timeout = 5000, batch } = options;
         const requestId = this._getNextRequestId();
         return new Promise((resolve, reject) => {
             const timeoutId = setTimeout(() => {
                 reject(new Error("bus request timed out"));
                 this._pendingRequests.delete(requestId);
             }, timeout);
-            this._pendingRequests.set(requestId, { resolve, reject, timeout: timeoutId });
+            this._pendingRequests.set(requestId, {
+                resolve,
+                reject,
+                timeout: timeoutId
+            });
             this._sendPayload(message, { needResponse: requestId, batch });
         });
     }
@@ -138,8 +144,7 @@ export class Bus {
     }
 
     private _sendPayload(
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        message: BusMessage,
+        message: BusMessage | JSONSerializable,
         options: {
             needResponse?: string;
             responseTo?: string;
@@ -212,11 +217,11 @@ export class Bus {
             }
         } else if (needResponse) {
             // This is a request that expects a response
-            const response = await this.onRequest?.(message);
-            this._sendPayload(response!, { responseTo: needResponse });
+            const response = await this.onRequest?.(message as RequestMessage);
+            this._sendPayload(response ?? {}, { responseTo: needResponse });
         } else {
             // This is a plain message
-            this.onMessage?.(message);
+            this.onMessage?.(message as BusMessage);
         }
     }
 }

tests/bus.test.ts

@@ -4,6 +4,7 @@ import { expect, describe, jest } from "@jest/globals";
 
 import { Bus } from "#src/shared/bus";
 import type { JSONSerializable, BusMessage } from "#src/shared/types";
+import { RequestMessage } from "#src/shared/bus-types.ts";
 
 class MockTargetWebSocket extends EventTarget {
     send(message: JSONSerializable) {
@@ -74,14 +75,14 @@ describe("Bus API", () => {
                 return "pong";
             }
         };
-        const response = await aliceBus.request("ping" as unknown as BusMessage);
+        const response = await aliceBus.request("ping" as unknown as RequestMessage);
         expect(response).toBe("pong");
     });
     test("promises are rejected when the bus is closed", async () => {
         const { aliceSocket } = mockSocketPair();
         const aliceBus = new Bus(aliceSocket as unknown as WebSocket);
         let rejected = false;
-        const promise = aliceBus.request("ping" as unknown as BusMessage);
+        const promise = aliceBus.request("ping" as unknown as RequestMessage);
         aliceBus.close();
         try {
             await promise;
@@ -96,7 +97,7 @@ describe("Bus API", () => {
         const { aliceSocket } = mockSocketPair();
         const aliceBus = new Bus(aliceSocket as unknown as WebSocket);
         const timeout = 500;
-        const promise = aliceBus.request("hello" as unknown as BusMessage, { timeout });
+        const promise = aliceBus.request("hello" as unknown as RequestMessage, { timeout });
         jest.advanceTimersByTime(timeout);
         await expect(promise).rejects.toThrow();
         jest.useRealTimers();