feat(VoiceReceiver): Proof of concept - Video recording

src/client/voice/receiver/FFmpegHandler.js

@@ -0,0 +1,106 @@
+'use strict';
+
+const { spawn } = require('child_process');
+const { createSocket } = require('dgram');
+const { EventEmitter } = require('events');
+const { Buffer } = require('node:buffer');
+const { Writable } = require('stream');
+const Util = require('../../../util/Util');
+const { StreamOutput } = require('../util/Socket');
+
+/**
+ * Represents a FFmpeg handler
+ * @extends {EventEmitter}
+ */
+class FFmpegHandler extends EventEmitter {
+  constructor(codec, portUdp, output) {
+    super();
+
+    /**
+     * The codec of the stream
+     * @type {VideoCodec}
+     */
+    this.codec = codec;
+
+    /**
+     * The UDP port to listen to
+     * @type {number}
+     */
+    this.portUdp = portUdp;
+
+    const isStream = output instanceof Writable;
+    if (isStream) {
+      this.outputStream = StreamOutput(output);
+    }
+
+    /**
+     * The output of the stream
+     * @type {string|Readable}
+     */
+    this.output = output;
+
+    const sdpData = Util.getSDPCodecName(codec, portUdp);
+    /**
+     * The FFmpeg process is ready or not
+     * @type {boolean}
+     */
+    this.ready = false;
+
+    /**
+     * The FFmpeg process
+     * @type {ChildProcessWithoutNullStreams}
+     */
+    this.stream = spawn('ffmpeg', [
+      '-reorder_queue_size',
+      '50',
+      '-err_detect',
+      'ignore_err',
+      '-flags2',
+      '+export_mvs',
+      '-fflags',
+      '+genpts',
+      '-fflags',
+      '+discardcorrupt',
+      '-use_wallclock_as_timestamps',
+      '1',
+      '-protocol_whitelist',
+      'file,udp,rtp,pipe,fd',
+      '-i',
+      '-', // Read from stdin
+      '-buffer_size',
+      '1000000',
+      '-max_delay',
+      '500000',
+      '-y',
+      '-f', // Specify the format
+      'matroska', // MKV format
+      isStream ? this.outputStream.url : output,
+    ]);
+    this.stream.stdin.write(sdpData);
+    this.stream.stdin.end();
+    this.stream.stderr.once('data', data => {
+      this.emit('debug', `stderr: ${data}`);
+      this.ready = true;
+      this.emit('ready');
+    });
+    this.socket = createSocket('udp4');
+  }
+  /**
+   * Send a payload to FFmpeg via UDP
+   * @param {Buffer} payload The payload
+   * @param {*} callback Callback
+   */
+  sendPayloadToFFmpeg(
+    payload,
+    callback = e => {
+      if (e) {
+        console.error('Error sending packet:', e);
+      }
+    },
+  ) {
+    const message = Buffer.from(payload);
+    this.socket.send(message, 0, message.length, this.portUdp, '127.0.0.1', callback);
+  }
+}
+
+module.exports = FFmpegHandler;

src/client/voice/receiver/PacketHandler.js

@@ -4,6 +4,7 @@ const EventEmitter = require('events');
 const { Buffer } = require('node:buffer');
 const crypto = require('node:crypto');
 const { setTimeout } = require('node:timers');
+const FFmpegHandler = require('./FFmpegHandler');
 const Speaking = require('../../../util/Speaking');
 const secretbox = require('../util/Secretbox');
 const { SILENCE_FRAME } = require('../util/Silence');
@@ -25,6 +26,7 @@ class PacketHandler extends EventEmitter {
     super();
     this.receiver = receiver;
     this.streams = new Map();
+    this.videoStreams = new Map();
     this.speakingTimeouts = new Map();
   }
 
@@ -54,6 +56,15 @@ class PacketHandler extends EventEmitter {
     return stream;
   }
 
+  makeVideoStream(user, portUdp, codec = 'H264', output) {
+    if (this.videoStreams.has(user)) return this.videoStreams.get(user);
+    const stream = new FFmpegHandler(codec, portUdp, output);
+    stream.on('ready', () => {
+      this.videoStreams.set(user, stream);
+    });
+    return stream;
+  }
+
   parseBuffer(buffer, shouldReturnTuple = false) {
     const { secret_key, mode } = this.receiver.connection.authentication;
     // Open packet
@@ -180,7 +191,7 @@ class PacketHandler extends EventEmitter {
     const userStat = this.connection.ssrcMap.get(ssrc - 1); // Video_ssrc
 
     if (!userStat) return;
-
+    const streamInfo = this.videoStreams.get(userStat.userId);
     // If the user is in video, we need to check if the packet is just silence
     if (userStat.hasVideo) {
       const packet = this.parseBuffer(buffer, true);
@@ -193,6 +204,10 @@ class PacketHandler extends EventEmitter {
         return;
       }
       this.receiver.emit('videoData', ssrc, userStat, header, videoPacket);
+
+      if (streamInfo) {
+        streamInfo.sendPayloadToFFmpeg(Buffer.concat(packet));
+      }
     }
   }
 

src/client/voice/receiver/Receiver.js

@@ -54,6 +54,29 @@ class VoiceReceiver extends EventEmitter {
     return stream;
   }
 
+  /**
+   * Options passed to `VoiceReceiver#createVideoStream`.
+   * @typedef {Object} ReceiveVideoStreamOptions
+   * @property {number} [portUdp] The UDP port to use for the video stream (local stream).
+   * @property {string} [codec='H264'] The codec to use for encoding the video. Default is 'H264'.
+   * @property {any} [output] Additional output options, as required.
+   */
+
+  /**
+   * Creates a new video receiving stream. If a stream already exists for a user, then that stream will be returned
+   * rather than generating a new one.
+   * <info>Proof of concept - Requires a very good internet connection</info>
+   * @param {UserResolvable} user The user to start listening to.
+   * @param {ReceiveVideoStreamOptions} options Options.
+   * @returns {FFmpegHandler} The video stream for the specified user.
+   */
+  createVideoStream(user, { portUdp, codec = 'H264', output } = {}) {
+    user = this.connection.client.users.resolve(user);
+    if (!user) throw new Error('VOICE_USER_MISSING');
+    const stream = this.packets.makeVideoStream(user.id, portUdp, codec, output);
+    return stream;
+  }
+
   /**
    * Emitted whenever there is a video data (Raw)
    * @event VoiceReceiver#videoData
@@ -61,49 +84,6 @@ class VoiceReceiver extends EventEmitter {
    * @param {{ userId: Snowflake, hasVideo: boolean }} ssrcData SSRC Data
    * @param {Buffer} header The unencrypted RTP header contains 12 bytes, Buffer<0xbe, 0xde> and the extension size
    * @param {Buffer} packetDecrypt Decrypted contains the extension, if any, the video packet
-   * @example
-   * // Send packet to VLC
-   * const dgram = require('dgram');
-   * // Replace these with your actual values
-   * const PORT = 5004; // The port VLC is listening on
-   * const HOST = '127.0.0.1'; // Your localhost or the IP address of the machine running VLC
-   * // Create a UDP socket
-   * const socket = dgram.createSocket('udp4');
-   * function sendRTPPacket(payload) {
-   *   const message = Buffer.from(payload);
-   *   socket.send(message, 0, message.length, PORT, HOST, err => {
-   *     if (err) {
-   *       console.error('Error sending packet:', err);
-   *     } else {
-   *       console.log(message);
-   *     }
-   *   });
-   * }
-   * const connection = await client.voice.joinChannel(channel, {
-   *     selfMute: true,
-   *     selfDeaf: true,
-   *     selfVideo: false,
-   *   });
-   * connection.receiver.on('videoData', (ssrc, ssrcData, header, packetDecrypt) => {
-   *  if (ssrcData.hasVideo) {
-   *      header[0] &= 0xef; // Remove the marker bit
-   *     // Strip decrypted RTP Header Extension if present
-   *     if (header.slice(12, 14).compare(Buffer.from([0xbe, 0xde])) === 0) {
-   *       const headerExtensionLength = header.slice(14).readUInt16BE();
-   *       packetDecrypt = packetDecrypt.subarray(4 * headerExtensionLength);
-   *     }
-   *      sendRTPPacket(Buffer.concat([header.slice(0, 12), packetDecrypt]));
-   *  }
-   * });
-   * // VLC SDP file (You can have it with FFmpeg)
-   * // ! Very buggy
-   * // o=- 0 0 IN IP4 <HOST>
-   * // s=No Name
-   * // c=IN IP4 <HOST>
-   * // t=0 0
-   * // a=tool:libavformat 61.1.100
-   * // m=video <PORT> RTP/AVP <RTP Dynamic Payload Type>
-   * // a=rtpmap:<RTP Dynamic Payload Type> <VP8|VP9|H264|H265>/90000
    */
 }
 

typings/index.d.ts

@@ -56,7 +56,7 @@ import {
   APIChannel,
   TeamMemberRole,
 } from 'discord-api-types/v9';
-import { ChildProcess } from 'node:child_process';
+import { ChildProcess, ChildProcessWithoutNullStreams } from 'node:child_process';
 import { EventEmitter } from 'node:events';
 import { AgentOptions } from 'node:https';
 import { Response } from 'node-fetch';
@@ -168,6 +168,7 @@ import {
   RawWidgetData,
   RawWidgetMemberData,
 } from './rawDataTypes';
+import { Socket } from 'node:dgram';
 
 //#region Classes
 
@@ -1043,6 +1044,7 @@ export class VoiceConnection extends EventEmitter {
   public voiceManager: ClientVoiceManager;
   public videoCodec: VideoCodec;
   public streamConnection: StreamConnection | null;
+  public streamWatchConnection: Collection<Snowflake, StreamConnectionReadonly>;
   public disconnect(): void;
   public playAudio(input: Readable | string, options?: StreamOptions): AudioDispatcher;
   public playVideo(input: Readable | string, options?: VideoOptions): VideoDispatcher;
@@ -1065,12 +1067,13 @@ export class VoiceConnection extends EventEmitter {
   public once(event: string, listener: (...args: any[]) => void): this;
 
   public createStreamConnection(): Promise<StreamConnection>;
+  public joinStreamConnection(user: UserResolvable): Promise<StreamConnectionReadonly>;
 }
 
 export class StreamConnection extends VoiceConnection {
   public createStreamConnection(): Promise<this>;
   public readonly voiceConnection: VoiceConnection;
-  public serverId: string;
+  public serverId: Snowflake;
   public isPaused: boolean;
   public streamConnection: this;
   public sendSignalScreenshare(): void;
@@ -1079,9 +1082,40 @@ export class StreamConnection extends VoiceConnection {
   public readonly streamKey: string;
 }
 
+export class StreamConnectionReadonly extends VoiceConnection {
+  public joinStreamConnection(): Promise<this>;
+  public readonly voiceConnection: VoiceConnection;
+  public serverId: Snowflake;
+  public userId: Snowflake;
+  public streamConnection: null;
+  public sendSignalScreenshare(): void;
+  private sendStopScreenshare(): void;
+  public readonly streamKey: string;
+  /** @deprecated removed */
+  public override playAudio(): AudioDispatcher;
+  /** @deprecated removed */
+  public override playVideo(): VideoDispatcher;
+}
+
+export class FFmpegHandler extends EventEmitter {
+  public codec: VideoCodec | 'H265' | 'VP9' | 'AV1';
+  public portUdp: number;
+  public ready: boolean;
+  public stream: ChildProcessWithoutNullStreams;
+  public socket: Socket;
+  public output: Writable | string;
+  public sendPayloadToFFmpeg(payload: Buffer): void;
+  public on(event: 'ready', listener: () => void): this;
+  public once(event: 'ready', listener: () => void): this;
+}
+
 export class VoiceReceiver extends EventEmitter {
   constructor(connection: VoiceConnection);
   public createStream(user: UserResolvable, options?: { mode?: 'opus' | 'pcm'; end?: 'silence' | 'manual' }): Readable;
+  public createVideoStream(
+    user: UserResolvable,
+    options?: { portUdp: number; codec: VideoCodec | 'H265' | 'VP9' | 'AV1'; output: Writable | string },
+  ): FFmpegHandler;
 
   public on(event: 'debug', listener: (error: Error | string) => void): this;
   public on(