Injecting audio/video stream into mediasoup using ffmpeg/gstreamer

ffmpeg.js

// Class to handle child process used for running FFmpeg

const childProcess = require('child_process');
const Streamer = require('./streamer');

module.exports = class FFmpeg extends Streamer {
  constructor(options) {
    super(options);

    this.time = options.time || '00:00:00.0'; // https://ffmpeg.org/ffmpeg-utils.html#time-duration-syntax
    this.createProcess();
    this.initListeners();
  }

  createProcess() {
    this.process = childProcess.spawn('ffmpeg', this.getArgs(this.kind, this.port, this.filename, this.time));
  }

  getArgs(kind, port, filename, time) {
    const map = (kind === 'video') ? '0:v:0' : '0:a:0';
    return [
      '-loglevel',
      'debug',
      '-re',
      '-v',
      'info',
      '-ss',
      time,
      '-i',
      filename,
      '-map',
      map,
      '-f',
      'tee',
      '-acodec',
      'libopus',
      '-ab',
      '128k',
      '-ac',
      '2',
      '-ar',
      '48000',
      '-pix_fmt',
      'yuv420p',
      '-c:v',
      'libvpx',
      '-b:v',
      '1000k',
      '-deadline',
      'realtime',
      '-cpu-used', // https://www.webmproject.org/docs/encoder-parameters/
      '2',
      // `[select=v:f=rtp:ssrc=22222222:payload_type=102]rtp://127.0.0.1:${port}`,
      `[select=a:f=rtp:ssrc=11111111:payload_type=101]rtp://127.0.0.1:${port}`,
    ];
  }
};

gstreamer.js

// Class to handle child process used for running GStreamer

const childProcess = require('child_process');
const Streamer = require('./streamer');

module.exports = class GStreamer extends Streamer {
  constructor(options) {
    super(options);

    this.createProcess();
    this.initListeners();
  }

  createProcess() {
    const args = (this.kind === 'audio') ?
      this.getAudioArgs(this.kind, this.port, this.filename):
      this.getVideoArgs(this.kind, this.port, this.filename);
    console.log(args);
    this.process = childProcess.spawn('gst-launch-1.0', args);
  }

  getVideoArgs(kind, port, filename) {
    // const map = (kind === 'video') ? '0:v:0' : '0:a:0';
    const VIDEO_SSRC = 22222222;
    const VIDEO_PAYLOAD_TYPE = 102;
    return [
      // '-e',
      // 'rtpbin', 'name=rtpbin', 'rtp-profile=avpf',
      // 'rtpbin', 'name=r', 'do-retransmission=1',
      'rtpbin', 'name=r',
      'filesrc', `location="${filename}"`, '!', 'decodebin',
      '!', 'queue',
      // '!', 'videorate', '!', 'video/x-raw,framerate=30/1',
      // '!', 'videoconvert', '!', 'video/x-raw,format=I420,framerate=30/1',
      '!', 'videoconvert',
      // '!', 'vp8enc', 'deadline=1',
      // '!', 'rtpvp8pay', 'pt=102', 'ssrc=22222222', 'picture-id-mode=1',
      // '!', 'rtpvp8pay', '!', 'capssetter', 'caps="application/x-rtp,payload=(int)102,clock-rate=(int)90000,ssrc=(uint)22222222,rtcp-fb-nack-pli=(int)1"',
      '!', 'x264enc', 'tune=zerolatency',
      // '!', 'x264enc', 'tune=zerolatency', 'speed-preset=1', 'dct8x8=true', 'quantizer=23', 'pass=qual',
      // '!', 'x264enc', '!', 'video/x-h264,profile=constrained-baseline,level=(string)3.1',
      // '!', 'rtph264pay', `ssrc=${VIDEO_SSRC}`, `pt=${VIDEO_PAYLOAD_TYPE}`,
      '!', 'rtph264pay', '!', 'capssetter', 'caps="application/x-rtp,payload=(int)102,clock-rate=(int)90000,ssrc=(uint)22222222,rtcp-fb-nack-pli=(int)1"',
      // '!', 'rtprtxqueue', 'max-size-time=2000', 'max-size-packets=0',
      '!', 'r.send_rtp_sink_0',
      'r.send_rtp_src_0', '!', 'udpsink', 'host=127.0.0.1', `port=${this.port}`,
      'r.send_rtcp_src_0', '!', 'udpsink', 'host=127.0.0.1', `port=${this.port}`, 'sync=false', 'async=false', 'udpsrc',
      '!', 'r.recv_rtcp_sink_0',
    ];
  }

  getAudioArgs(kind, port, filename) {
    // const map = (kind === 'video') ? '0:v:0' : '0:a:0';
    const VIDEO_SSRC = 11111111;
    const VIDEO_PAYLOAD_TYPE = 101;
    return [
      'rtpbin', 'name=r',
      'filesrc', `location="${filename}"`, '!', 'decodebin',
      '!', 'queue',
      '!', 'audioconvert',
      // '!', 'opusenc', 'bandwidth=superwideband bitrate-type=vbr',
      '!', 'opusenc',
      '!', 'rtpopuspay', 'pt=101', 'ssrc=11111111',
      // '!', 'rtpopuspay', '!', 'capssetter', 'caps="application/x-rtp,payload=(int)101,clock-rate=(int)48000,ssrc=(uint)11111111"',
      '!', 'rtprtxqueue',
      '!', 'r.send_rtp_sink_0',
      'r.send_rtp_src_0', '!', 'udpsink', 'host=127.0.0.1', `port=${this.port}`,
      'r.send_rtcp_src_0', '!', 'udpsink', 'host=127.0.0.1', `port=${this.port}`, 'sync=false', 'async=false', 'udpsrc',
      '!', 'r.recv_rtcp_sink_0',
    ];
  }
};

main.js

const publishVideoStream = async (data) => {
  const streamTransport = await router.createPlainRtpTransport({
    listenIp: '127.0.0.1',
    rtcpMux: true,
    comedia: true,
  });

  const producer = await streamTransport.produce({
    kind: 'video',
    rtpParameters: {
      codecs: [
        {
          // mimeType: 'video/vp8',
          mimeType: 'video/H264',
          payloadType: 102,
          clockRate: 90000,
          parameters: {
            'level-asymmetry-allowed': 1,
            'packetization-mode': 1,
            'profile-level-id': '42e01f',
          },
          rtcpFeedback: [
            { type: 'nack' },
            { type: 'nack', parameter: 'pli' },
            { type: 'ccm', parameter: 'fir' },
            { type: 'goog-remb' },
          ],
        },
        // {
        //   mimeType: 'video/rtx',
        //   payloadType: 103,
        //   clockRate: 90000,
        //   parameters: { apt: 102 },
        // },
      ],
      encodings: [{ ssrc: 22222222 }],
    },
    appData: {},
  });

  // producer.enableTraceEvent(['keyframe', 'pli', 'nack']);
  // producer.on('trace', (trace) => {
  //   log.debug('trace', trace);
  // });

  // const file = this.session.files.get(data.id);
  // new FFmpeg({
  new GStreamer({
    kind: 'video',
    port: streamTransport.tuple.localPort,
    filename: 'video.mp4',
  });

  return producer;
};

const publishAudioStream = async (data) => {
  const streamTransport = await router.createPlainRtpTransport({
    listenIp: '127.0.0.1',
    rtcpMux: true,
    comedia: true,
  });

  const producer = await streamTransport.produce({
    kind: 'audio',
    rtpParameters: {
      codecs: [{
        mimeType: 'audio/opus',
        clockRate: 48000,
        payloadType: 101,
        channels: 2,
        parameters: { 'sprop-stereo': 1 },
        rtcpFeedback: [
          { type: 'transport-cc' },
        ],
      }],
      encodings: [{ ssrc: 11111111 }],
    },
    appData: {},
  });

  // new GStreamer({
  new FFmpeg({
    kind: 'audio',
    port: streamTransport.tuple.localPort,
    filename: 'video.mp4',
  });

  return producer;
};

streamer.js

const { EventEmitter } = require('events');

/**
 * Streams audio/video file
 */
class Streamer {
  constructor(options = {}) {
    this.kind = options.kind;
    this.port = options.port;
    this.rtpPort = options.rtpPort;
    this.rtcpPort = options.rtcpPort;
    this.filename = options.filename;

    this.process = null;
    this.observer = new EventEmitter();
  }

  initListeners() {
    if (this.process.stderr) {
      this.process.stderr.setEncoding('utf-8');
      this.process.stderr.on('data', this.onData.bind(this));
    }

    if (this.process.stdout) {
      this.process.stdout.setEncoding('utf-8');
      this.process.stdout.on('data', this.onData.bind(this));
    }

    this.process.on('message', message => {
      console.log('process::message', message)
    });

    this.process.on('error', error => {
      console.error('process::error', error)
    });

    this.process.once('close', () => {
      console.log('process::close');
      this.observer.emit('close');
    });
  }

  onData(data) {
    // TODO: parse and fetch the time
    // this.observer.emit('time', time);
    console.log('process::data', data);
  }

  /**
   * Stops streaming
   */
  stop() {
    console.log('process::stop [pid:%d]', this.process.pid);
    this.process.kill('SIGINT');
  }
}

module.exports = Streamer;