[IMP] wip/poc

Author: ThanhDodeurOdoo

src/client.ts

@@ -260,6 +260,23 @@ export class SfuClient extends EventTarget {
         await Promise.all(proms);
         return stats;
     }
+    async startRecording() {
+        return this._bus?.request(
+            {
+                name: CLIENT_REQUEST.START_RECORDING,
+            },
+            { batch: true }
+        );
+    }
+
+    async stopRecording() {
+        return this._bus?.request(
+            {
+                name: CLIENT_REQUEST.STOP_RECORDING,
+            },
+            { batch: true }
+        );
+    }
 
     /**
      * Updates the server with the info of the session (isTalking, isCameraOn,...) so that it can broadcast it to the

src/config.ts

@@ -67,7 +67,7 @@ export const PORT: number = Number(process.env.PORT) || 8070;
 /**
  * Whether the recording feature is enabled, true by default.
  */
-export const RECORDING: boolean = !FALSY_INPUT.has(process.env.LOG_TIMESTAMP!);
+export const RECORDING: boolean = !FALSY_INPUT.has(process.env.RECORDING!);
 
 /**
  * The number of workers to spawn (up to core limits) to manage RTC servers.

src/models/channel.ts

@@ -187,7 +187,6 @@ export class Channel extends EventEmitter {
         this.createDate = now.toISOString();
         this.remoteAddress = remoteAddress;
         this.recorder = config.recording.enabled && options.useRecording ? new Recorder(this) : undefined;
-        this.recorder?.todo();
         this.key = key ? Buffer.from(key, "base64") : undefined;
         this.uuid = crypto.randomUUID();
         this.name = `${remoteAddress}*${this.uuid.slice(-5)}`;

src/models/recorder.ts

@@ -6,13 +6,25 @@ const logger = new Logger("RECORDER");
 
 export class Recorder extends EventEmitter {
     channel: Channel;
+    state: "started" | "stopped" = "stopped";
+    ffmpeg = null;
 
     constructor(channel: Channel) {
         super();
         this.channel = channel;
     }
 
-    todo() {
-        logger.warn("TODO: Everything");
+    async start() {
+        this.state = "started";
+        logger.trace("TO IMPLEMENT");
+        // TODO ffmpeg instance creation, start
+        return { state: this.state };
+    }
+
+    async stop() {
+        this.state = "stopped";
+        logger.trace("TO IMPLEMENT");
+        // TODO ffmpeg instance stop, cleanup, save,...
+        return { state: this.state };
     }
 }

src/models/session.ts

@@ -110,6 +110,7 @@ const logger = new Logger("SESSION");
  *
  * @fires Session#stateChange - Emitted when session state changes
  * @fires Session#close - Emitted when session is closed
+ * @fires Session#producer - Emitted when a new producer is created
  */
 export class Session extends EventEmitter {
     /** Communication bus for WebSocket messaging */
@@ -138,9 +139,9 @@ export class Session extends EventEmitter {
         camera: null,
         screen: null
     };
-    public permissions: SessionPermissions = {
+    public readonly permissions: SessionPermissions = Object.seal({
         recording: false
-    };
+    });
     /** Parent channel containing this session */
     private readonly _channel: Channel;
     /** Recovery timeouts for failed consumers */
@@ -184,9 +185,26 @@ export class Session extends EventEmitter {
 
     set state(state: SESSION_STATE) {
         this._state = state;
+        /**
+         * @event Session#stateChange
+         * @type {{ state: SESSION_STATE }}
+         */
         this.emit("stateChange", state);
     }
 
+    updatePermissions(permissions: SessionPermissions | undefined): void {
+        if (!permissions) {
+            return;
+        }
+        for (const key of Object.keys(this.permissions) as (keyof SessionPermissions)[]) {
+            const newVal = permissions[key];
+            if (newVal === undefined) {
+                continue;
+            }
+            this.permissions[key] = Boolean(permissions[key]);
+        }
+    }
+
     async getProducerBitRates(): Promise<ProducerBitRates> {
         const bitRates: ProducerBitRates = {};
         const proms: Promise<void>[] = [];
@@ -651,8 +669,25 @@ export class Session extends EventEmitter {
                 logger.debug(`[${this.name}] producing ${type}: ${codec?.mimeType}`);
                 this._updateRemoteConsumers();
                 this._broadcastInfo();
+                /**
+                 * @event Session#producer
+                 * @type {{ type: StreamType, producer: Producer }}
+                 */
+                this.emit("producer", { type, producer });
                 return { id: producer.id };
             }
+            case CLIENT_REQUEST.START_RECORDING: {
+                if (this.permissions.recording && this._channel.recorder) {
+                    return this._channel.recorder.start();
+                }
+                return;
+            }
+            case CLIENT_REQUEST.STOP_RECORDING: {
+                if (this.permissions.recording && this._channel.recorder) {
+                    return this._channel.recorder.stop();
+                }
+                return;
+            }
             default:
                 logger.warn(`[${this.name}] Unknown request type: ${name}`);
                 throw new Error(`Unknown request type: ${name}`);

src/services/ws.ts

@@ -130,9 +130,7 @@ function connect(webSocket: WebSocket, credentials: Credentials): Session {
     }
     const bus = new Bus(webSocket, { batchDelay: config.timeouts.busBatch });
     const { session } = Channel.join(channel.uuid, session_id);
-    if (permissions) {
-        Object.assign(session.permissions, permissions);
-    }
+    session.updatePermissions(permissions);
     webSocket.send(JSON.stringify(session.availableFeatures)); // client can start using ws after this message.
     session.once("close", ({ code }: { code: string }) => {
         let wsCloseCode = WS_CLOSE_CODE.CLEAN;

src/shared/enums.ts

@@ -33,7 +33,11 @@ export enum CLIENT_REQUEST {
     /** Requests the server to connect the server-to-client transport */
     CONNECT_STC_TRANSPORT = "CONNECT_STC_TRANSPORT",
     /** Requests the creation of a consumer that is used to upload a track to the server */
-    INIT_PRODUCER = "INIT_PRODUCER"
+    INIT_PRODUCER = "INIT_PRODUCER",
+    /** Requests to start recording of the call */
+    START_RECORDING = "START_RECORDING",
+    /** Requests to stop recording of the call */
+    STOP_RECORDING = "STOP_RECORDING"
 }
 
 export enum CLIENT_MESSAGE {

src/shared/types.ts

@@ -54,6 +54,8 @@ export type BusMessage =
           name: typeof CLIENT_REQUEST.INIT_PRODUCER;
           payload: { type: StreamType; kind: MediaKind; rtpParameters: RtpParameters };
       }
+    | { name: typeof CLIENT_REQUEST.START_RECORDING; payload?: never }
+    | { name: typeof CLIENT_REQUEST.STOP_RECORDING; payload?: never }
     | {
           name: typeof SERVER_MESSAGE.BROADCAST;
           payload: { senderId: SessionId; message: JSONSerializable };

src/utils/utils.ts

@@ -12,6 +12,7 @@ const ASCII = {
         green: "\x1b[32m",
         yellow: "\x1b[33m",
         white: "\x1b[37m",
+        cyan: "\x1b[36m",
         default: "\x1b[0m"
     }
 } as const;
@@ -48,6 +49,19 @@ export interface ParseBodyOptions {
     json?: boolean;
 }
 
+function getCallChain(depth: number = 8): string {
+    const stack = new Error().stack?.split("\n").slice(2, depth + 2) ?? [];
+    return stack
+        .map(line => {
+            const match = line.trim().match(/^at\s+(.*?)\s+\(/);
+            return match ? match[1] : null;
+        })
+        .slice(1, depth + 1)
+        .filter(Boolean)
+        .reverse()
+        .join(" > ");
+}
+
 export class Logger {
     private readonly _name: string;
     private readonly _colorize: (text: string, color?: string) => string;
@@ -83,6 +97,9 @@ export class Logger {
     verbose(text: string): void {
         this._log(console.log, ":VERBOSE:", text, ASCII.color.white);
     }
+    trace(message: string, { depth = 8 }: { depth?: number } = {}): void {
+        this._log(console.log, ":TRACE:", `${getCallChain(depth)} ${message}`, ASCII.color.cyan);
+    }
     private _generateTimeStamp(): string {
         const now = new Date();
         return now.toISOString() + " ";

tests/network.test.ts

@@ -289,4 +289,16 @@ describe("Full network", () => {
         expect(event1.detail.payload.message).toBe(message);
         expect(event2.detail.payload.message).toBe(message);
     });
+    test("POC RECORDING", async () => {
+        const channelUUID = await network.getChannelUUID();
+        const user1 = await network.connect(channelUUID, 1);
+        await once(user1.session, "stateChange");
+        const sender = await network.connect(channelUUID, 3);
+        await once(sender.session, "stateChange");
+        sender.session.updatePermissions({ recording: true });
+        const startResult = await sender.sfuClient.startRecording() as { state: string };
+        expect(startResult.state).toBe("started");
+        const stopResult = await sender.sfuClient.stopRecording()  as { state: string };
+        expect(stopResult.state).toBe("stopped");
+    });
 });