Test node-sctp and mediasoup

all-tests.sh

#!/usr/bin/env bash

set -e

./node-sctp-mediasoup-test.js 1000

./node-sctp-mediasoup-test.js 1300

./node-sctp-mediasoup-test.js 5000

HIGH_WATER_MARK=20000 ./node-sctp-mediasoup-test.js 20000

NUM_MSG=2 HIGH_WATER_MARK=50000 ./node-sctp-mediasoup-test.js 50000

PMTU=32000 NUM_MSG=3 HIGH_WATER_MARK=50000 ./node-sctp-mediasoup-test.js 50000

HIGH_WATER_MARK=150000 TIMEOUT=2 ./node-sctp-mediasoup-test.js 150000

# This fails. It seems (via tshark) that mediasoup does not send much bytes
# to the node-sctp receiving Socket.
PMTU=32000 HIGH_WATER_MARK=300000 TIMEOUT=3 ./node-sctp-mediasoup-test.js 300000

node-sctp-and-mediasoup-test.js

#!/usr/bin/env node

const dgram = require('dgram');
const mediasoup = require('mediasoup');
const sctp = require('sctp');

if ([ '--help', '-h' ].includes(process.argv[2])) {
  help();
  process.exit(0);
}

const MSG_SIZE = Number(process.argv[2]);
const NUM_MSG = Number(process.env.NUM_MSG) || 1;
const PMTU = Number(process.env.PMTU);
const RWND = Number(process.env.RWND);
const HIGH_WATER_MARK = Number(process.env.HIGH_WATER_MARK) || 16000;
const TIMEOUT = Number(process.env.TIMEOUT) || 1;
const SRC_IP = process.env.SCR_IP || '127.0.0.1';
const DST_IP = process.env.DST_IP || '127.0.0.1';

if (!MSG_SIZE) {
  console.error('[ERROR]: missing MSG_SIZE command line argument');

  help();
  process.exit(1);
}

function help() {
  console.log('');
  console.log('USAGE: [NUM_MSG=X] [PMTU=X] [RWND=X] [HIGH_WATER_MARK=X] [TIMEOUT=X] [SRC_IP=X] [DST_IP=X] ./node-sctp-mediasoup-test.js MSG_SIZE');
  console.log('');
  console.log('  Command line arguments:');
  console.log('  - MSG_SIZE : Size in bytes of the SCTP message to be sent (mandatory)');
  console.log('');
  console.log('  Environment variables:');
  console.log('  - NUM_MSG         : Number of messages to send all together (default: 1)');
  console.log('  - PMTU            : PMTU for node-sctp');
  console.log('  - RWND            : RWND for node-sctp');
  console.log('  - HIGH_WATER_MARK : highWaterMark value for node-sctp Socket (default: 16000)');
  console.log('  - TIMEOUT         : Time in seconds to wait for SCTP messages delivery (default: 1)');
  console.log('  - SRC_IP          : IP of the sending mediasoup transport (default: "127.0.0.1")');
  console.log('  - DST_IP          : IP of the receiving mediasoup transport (default: "127.0.0.1")');
  console.log('');
}

run()
  .then(() => {
    console.log('[INFO] test succeeds :)');

    process.exit(0);
   })
  .catch((error) => {
    console.error('[ERROR]: test failed: %o', error);

    process.exit(1);
  });

async function run() {
  console.log();
  console.log(
    '[INFO] running test with MSG_SIZE:%s, NUM_MSG:%d, PMTU:%s, RWND:%s, HIGH_WATER_MARK:%s, TIMEOUT:%d, SRC_IP:%s, DST_IP:%s',
    MSG_SIZE,
    NUM_MSG,
    PMTU,
    RWND,
    HIGH_WATER_MARK,
    TIMEOUT,
    SRC_IP,
    DST_IP);

  let outboundSctpStreamTotalSentBytes = 0;
  let outboundSctpStreamTotalSentMessages = 0;
  let inboundSctpStreamTotalReceivedBytes = 0;
  let inboundSctpStreamTotalReceivedMessages = 0;

  // Set node-sctp global defaults.
  // if (PMTU)
  //   sctp.defaults({ PMTU: PMTU });
  // if (RWND)
  //   sctp.defaults({ RWND: RWND });

  // Create a mediasoup Worker.
  const worker = await mediasoup.createWorker({
    logLevel: 'debug',
    logTags: [ 'sctp' ]
  });

  // Create a mediasoup Router.
  const router = await worker.createRouter();

  // Create a mediasoup PlainRtpTransport for connecting the sending node-sctp
  // Socket.
  const sendTransport = await router.createPlainRtpTransport({
    listenIp: { ip: SRC_IP },
    enableSctp: true,
    numSctpStreams: { OS: 512, MIS: 512 },
    maxSctpMessageSize: 4000000 // 4 MB
  });

  // Node UDP socket for the sending SCTP.
  const sendUdpSocket = dgram.createSocket({ type: 'udp4' });

  await new Promise(resolve => sendUdpSocket.bind(11111, SRC_IP, resolve));

  const localSendUdpPort = sendUdpSocket.address().port;

  // Connect the mediasoup send PlainRtpTransport to the UDP socket.
  await sendTransport.connect({ ip: SRC_IP, port: localSendUdpPort });

  // Create a node-sctp sending Socket.
  const sendSctpSocket = sctp.connect({
    localPort: 5000, // Required for SCTP over plain UDP in mediasoup.
    port: 5000,      // Required for SCTP over plain UDP in mediasoup.
    OS: sendTransport.sctpParameters.OS,
    MIS: sendTransport.sctpParameters.MIS,
    unordered: false,
    udpTransport: sendUdpSocket,
    udpPeer: {
      address: sendTransport.tuple.localIp,
      port: sendTransport.tuple.localPort,
    },
    highWaterMark: HIGH_WATER_MARK
  });

  sendSctpSocket.on('error', (error) => {
    console.error('[ERROR] node-sctp sending Socket "error" event: %o', error);

    process.exit(2);
  });

  console.log('[INFO] waiting for the sending SCTP association to be open');

  // Wait for the SCTP association to be open.
  await Promise.race([
    new Promise((resolve, reject) => {
      setTimeout(() => reject(new Error('SCTP connection timeout')), 2000)
    }),
    new Promise(resolve => sendSctpSocket.on('connect', resolve)),
  ]);

  console.log('[INFO] creating a node-sctp outbound Stream [streamId:1]');

  // Create a node-sctp outbound Stream with id 1 (don't use the implicit Stream in the
  // node-sctp Socket).
  const outboundSctpStream = sendSctpSocket.createStream(1);

  // Create a mediasoup DataProducer representing the node-sctp outbound Stream.
  console.log(
    '[INFO] creating a mediasoup DataProducer associated to the node-sctp outbound Stream');

  const dataProducer = await sendTransport.produceData({
    sctpStreamParameters: {
      streamId: 1,
      ordered: true,
    }
  });

  // Create a mediasoup PlainRtpTransport for consuming SCTP data from the
  // node-sctp outbound Stream.
  const recvTransport = await router.createPlainRtpTransport({
    listenIp: { ip: DST_IP },
    enableSctp: true,
    numSctpStreams: { OS: 512, MIS: 512 },
    maxSctpMessageSize: 4000000 // 4 MB
  });

  // Node UDP socket for the receiving SCTP.
  const recvUdpSocket = dgram.createSocket({ type: 'udp4' });

  await new Promise(resolve => recvUdpSocket.bind(22222, DST_IP, resolve));

  const localRecvUdpPort = recvUdpSocket.address().port;

  // Connect the mediasoup receiving PlainRtpTransport to the UDP socket.
  await recvTransport.connect({ ip: DST_IP, port: localRecvUdpPort });

  // Create a node-sctp receiving Socket.
  const recvSctpSocket = sctp.connect({
    localPort: 5000, // Required for SCTP over plain UDP in mediasoup.
    port: 5000,      // Required for SCTP over plain UDP in mediasoup.
    OS: recvTransport.sctpParameters.OS,
    MIS: recvTransport.sctpParameters.MIS,
    unordered: false,
    udpTransport: recvUdpSocket,
    udpPeer: {
      address: recvTransport.tuple.localIp,
      port: recvTransport.tuple.localPort,
    },
    highWaterMark: HIGH_WATER_MARK
  });

  recvSctpSocket.on('error', (error) => {
    console.error('[ERROR] node-sctp receiving Socket "error" event: %o', error);

    process.exit(2);
  });

  console.log('[INFO] waiting for the receiving SCTP association to be open');

  await Promise.race([
    new Promise((resolve, reject) => {
      setTimeout(() => reject(new Error('SCTP connection timeout')), 2000)
    }),
    new Promise(resolve => recvSctpSocket.on('connect', resolve)),
  ]);

  let inboundSctpStreamCreated = false;

  // Handle SCTP messages received in the node-sctp receiving Socket.
  recvSctpSocket.on('stream', async (stream, streamId) => {
    console.log(
      '[INFO] node-sctp inbound Stream created in the node-sctp receiving Socket [streamId:%d]',
      streamId);

    if (inboundSctpStreamCreated) {
      console.error('[ERROR] just a single node-sctp inbound Stream should be generated');

      process.exit(2);
    }

    inboundSctpStreamCreated = true;

    stream.on('data', (buffer) => {
      // Ensure it's a WebRTC DataChannel string.
      if (buffer.ppid !== sctp.PPID.WEBRTC_STRING) {
        console.error(
          '[ERROR] non WebRTC string data received in the node-sctp inbound Stream');

        process.exit(2);
      }

      console.log(
        '[INFO] SCTP message received in the node-sctp inbound Stream [size:%d]',
        buffer.byteLength);

      inboundSctpStreamTotalReceivedBytes += buffer.byteLength;
      inboundSctpStreamTotalReceivedMessages++;
    });
  });

  // Create a mediasoup DataConsumer representing the node-sctp inbound Stream.
  console.log(
    '[INFO] creating a mediasoup DataConsumer associated to the node-sctp inbound Stream');

  dataConsumer = await recvTransport.consumeData({
    dataProducerId: dataProducer.id
  });

  for (let i = 0; i < NUM_MSG; ++i) {
    sendData(outboundSctpStream, dataProducer, MSG_SIZE);

    outboundSctpStreamTotalSentBytes += MSG_SIZE;
    outboundSctpStreamTotalSentMessages++;
  }

  // Wait a bit for all bytes to be delivered.
  await new Promise((resolve) => setTimeout(resolve, TIMEOUT * 1000));

  // Check sent and received bytes and SCTP messages.
  const dataProducerStats = await dataProducer.getStats();
  const dataProducerTotalReceivedBytes = dataProducerStats[0].bytesReceived;
  const dataProducerTotalReceivedMessages = dataProducerStats[0].messagesReceived;
  const dataConsumerStats = await dataConsumer.getStats();
  const dataConsumerTotalSentBytes = dataConsumerStats[0].bytesSent;
  const dataConsumerTotalSentMessages = dataConsumerStats[0].messagesSent;

  console.log();
  console.log('[INFO] test results:');
  console.log(
    '- node-sctp outbound Stream sent %d bytes in %d SCTP messages to mediasoup DataProducer',
    outboundSctpStreamTotalSentBytes,
    outboundSctpStreamTotalSentMessages);
  console.log(
    '- mediasoup DataProducer received %d bytes in %d SCTP messages from node-sctp outbound Stream',
    dataProducerTotalReceivedBytes,
    dataProducerTotalReceivedMessages);
  console.log(
    '- mediasoup DataConsumer sent %d bytes in %d SCTP messages to node-sctp inbound Stream',
    dataConsumerTotalSentBytes,
    dataConsumerTotalSentMessages);
  console.log(
    '- node-sctp inbound Streamm received %d bytes in %d SCTP messages from mediasoup DataConsumer',
    inboundSctpStreamTotalReceivedBytes,
    inboundSctpStreamTotalReceivedMessages);
  console.log();

  if (
    outboundSctpStreamTotalSentBytes !== dataProducerTotalReceivedBytes ||
    dataProducerTotalReceivedBytes !== dataConsumerTotalSentBytes ||
    dataConsumerTotalSentBytes !== inboundSctpStreamTotalReceivedBytes ||
    outboundSctpStreamTotalSentMessages !== dataProducerTotalReceivedMessages ||
    dataProducerTotalReceivedMessages !== dataConsumerTotalSentMessages ||
    dataConsumerTotalSentMessages !== inboundSctpStreamTotalReceivedMessages
  ) {
    throw new Error('SCTP sent and received bytes and/or messages do not match!');
  }
}

function sendData(outboundSctpStream, dataProducer, size) {
  console.log(
    '[INFO] sending a message of %d bytes from the node-sctp outbound Stream',
    size);

  const buffer = Buffer.alloc(size, 'X');

  // Set ppid of type WebRTC DataChannel string.
  buffer.ppid = sctp.PPID.WEBRTC_STRING;

  outboundSctpStream.write(buffer);
}

Cool, I've added a new all-tests.sh above that runs the test with different arguments. The only failing test is the following:

$ PMTU=32000 HIGH_WATER_MARK=300000 TIMEOUT=3 ./node-sctp-mediasoup-test.js 300000

[INFO] running test with MSG_SIZE:300000, NUM_MSG:1, PMTU:32000, RWND:NaN, HIGH_WATER_MARK:300000, TIMEOUT:3, SRC_IP:127.0.0.1, DST_IP:127.0.0.1
[INFO] waiting for the sending SCTP association to be open
[INFO] creating a node-sctp outbound Stream [streamId:1]
[INFO] creating a mediasoup DataProducer associated to the node-sctp outbound Stream
[INFO] waiting for the receiving SCTP association to be open
[INFO] creating a mediasoup DataConsumer associated to the node-sctp inbound Stream
[INFO] sending a message of 300000 bytes from the node-sctp outbound Stream

[INFO] test results:
- node-sctp outbound Stream sent 300000 bytes in 1 SCTP messages to mediasoup DataProducer
- mediasoup DataProducer received 300000 bytes in 1 SCTP messages from node-sctp outbound Stream
- mediasoup DataConsumer sent 300000 bytes in 1 SCTP messages to node-sctp inbound Stream
- node-sctp inbound Streamm received 0 bytes in 0 SCTP messages from mediasoup DataConsumer

[ERROR]: test failed: Error: SCTP sent and received bytes and/or messages do not match!
    at run (/Users/ibc/tmp/node-sctp-issue/node-sctp-mediasoup-test.js:295:11)
    at processTicksAndRejections (internal/process/task_queues.js:93:5) {
  [stack]: 'Error: SCTP sent and received bytes and/or messages do not match!\n' +
    '    at run (/Users/ibc/tmp/node-sctp-issue/node-sctp-mediasoup-test.js:295:11)\n' +
    '    at processTicksAndRejections (internal/process/task_queues.js:93:5)',
  [message]: 'SCTP sent and received bytes and/or messages do not match!'
}

$ sudo tshark -i any -n -s0 -Y 'ip.dst==127.0.0.1 and udp.port==22222'
Capturing on 'any'


  688   1.542536    127.0.0.1 → 127.0.0.1    UDP 128 48280 → 22222 Len=100
  689   1.542540    127.0.0.1 → 127.0.0.1    UDP 128 48280 → 22222 Len=100
  702   1.543201    127.0.0.1 → 127.0.0.1    UDP 60 22222 → 48280 Len=32
  703   1.543205    127.0.0.1 → 127.0.0.1    UDP 60 22222 → 48280 Len=32
  704   1.543235    127.0.0.1 → 127.0.0.1    UDP 372 48280 → 22222 Len=344
  705   1.543239    127.0.0.1 → 127.0.0.1    UDP 372 48280 → 22222 Len=344
  708   1.543668    127.0.0.1 → 127.0.0.1    UDP 284 22222 → 48280 Len=256
  709   1.543672    127.0.0.1 → 127.0.0.1    UDP 284 22222 → 48280 Len=256
  710   1.543722    127.0.0.1 → 127.0.0.1    UDP 44 48280 → 22222 Len=16
  711   1.543725    127.0.0.1 → 127.0.0.1    UDP 44 48280 → 22222 Len=16

which does not seem related to node-sctp.