mediasoup

mkdir huddle-sfu && cd huddle-sfu
npm install mediasoup@3.31.0 mediasoup-client@3.25.3 ws@8.22.0

{
  "name": "huddle-sfu",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "dependencies": {
    "mediasoup": "3.31.0",
    "mediasoup-client": "3.25.3",
    "ws": "8.22.0"
  }
}

// Huddle SFU server: WebSocket signalling + mediasoup (one worker, one router per room).
import http from 'node:http';
import { readFile } from 'node:fs/promises';
import { WebSocketServer } from 'ws';
import * as mediasoup from 'mediasoup';

const HTTP_PORT = Number(process.env.HTTP_PORT ?? 41200);
const RTC_PORT = Number(process.env.RTC_PORT ?? 41201);
const ANNOUNCED = process.env.ANNOUNCED_ADDRESS ?? '127.0.0.1';

const mediaCodecs = [
  { kind: 'audio', mimeType: 'audio/opus', clockRate: 48000, channels: 2 },
  { kind: 'video', mimeType: 'video/VP8', clockRate: 90000 },
  { kind: 'video', mimeType: 'video/H264', clockRate: 90000,
    parameters: { 'packetization-mode': 1, 'profile-level-id': '42e01f', 'level-asymmetry-allowed': 1 } },
];

const worker = await mediasoup.createWorker({ logLevel: 'warn' });
worker.on('died', (err) => { console.error('worker died', err); process.exit(1); });

const webRtcServer = await worker.createWebRtcServer({
  listenInfos: [
    { protocol: 'udp', ip: '0.0.0.0', announcedAddress: ANNOUNCED, port: RTC_PORT },
    { protocol: 'tcp', ip: '0.0.0.0', announcedAddress: ANNOUNCED, port: RTC_PORT },
  ],
});

const rooms = new Map(); // name -> { router, peers: Map, speakerObserver }

async function getRoom(name) {
  let room = rooms.get(name);
  if (!room) {
    const router = await worker.createRouter({ mediaCodecs });
    const speakerObserver = await router.createActiveSpeakerObserver({ interval: 300 });
    room = { router, peers: new Map(), speakerObserver };
    speakerObserver.on('dominantspeaker', ({ producer }) => {
      broadcast(room, null, 'activeSpeaker', { peerId: producer.appData.peerId });
    });
    rooms.set(name, room);
  }
  return room;
}

function broadcast(room, exceptId, notify, data) {
  for (const p of room.peers.values()) {
    if (p.id !== exceptId) p.ws.send(JSON.stringify({ notify, data }));
  }
}

const server = http.createServer(async (req, res) => {
  const file = req.url === '/' ? '/index.html' : req.url.split('?')[0];
  try {
    const body = await readFile(new URL(`./public${file}`, import.meta.url));
    res.writeHead(200, { 'content-type': file.endsWith('.js') ? 'text/javascript' : 'text/html' });
    res.end(body);
  } catch { res.writeHead(404).end(); }
});
const wss = new WebSocketServer({ server, path: '/ws' });

let nextPeer = 1;
wss.on('connection', (ws) => {
  const peer = { id: `p${nextPeer++}`, ws, transports: new Map(), producers: new Map(), consumers: new Map(), room: null };

  const handlers = {
    async join({ room: roomName, name }) {
      const room = await getRoom(roomName);
      peer.room = room; peer.name = name;
      room.peers.set(peer.id, peer);
      const existing = [];
      for (const other of room.peers.values()) {
        for (const pr of other.producers.values()) existing.push({ producerId: pr.id, peerId: other.id, kind: pr.kind });
      }
      return { peerId: peer.id, routerRtpCapabilities: room.router.rtpCapabilities, existing };
    },
    async createTransport() {
      const transport = await peer.room.router.createWebRtcTransport({
        webRtcServer, enableUdp: true, enableTcp: true, preferUdp: true,
      });
      peer.transports.set(transport.id, transport);
      return {
        id: transport.id,
        iceParameters: transport.iceParameters,
        iceCandidates: transport.iceCandidates,
        dtlsParameters: transport.dtlsParameters,
      };
    },
    async connectTransport({ transportId, dtlsParameters }) {
      await peer.transports.get(transportId).connect({ dtlsParameters });
    },
    async produce({ transportId, kind, rtpParameters }) {
      const producer = await peer.transports.get(transportId)
        .produce({ kind, rtpParameters, appData: { peerId: peer.id } });
      peer.producers.set(producer.id, producer);
      if (kind === 'audio') await peer.room.speakerObserver.addProducer({ producerId: producer.id });
      broadcast(peer.room, peer.id, 'newProducer', { producerId: producer.id, peerId: peer.id, kind });
      return { id: producer.id };
    },
    async consume({ transportId, producerId, rtpCapabilities }) {
      if (!peer.room.router.canConsume({ producerId, rtpCapabilities })) throw new Error('cannot consume');
      const consumer = await peer.transports.get(transportId)
        .consume({ producerId, rtpCapabilities, paused: true }); // resume after the client is ready
      peer.consumers.set(consumer.id, consumer);
      consumer.on('producerclose', () => {
        peer.consumers.delete(consumer.id);
        peer.ws.send(JSON.stringify({ notify: 'consumerClosed', data: { consumerId: consumer.id } }));
      });
      return { id: consumer.id, producerId, kind: consumer.kind, rtpParameters: consumer.rtpParameters, type: consumer.type };
    },
    async resumeConsumer({ consumerId }) { await peer.consumers.get(consumerId).resume(); },
    async setPreferredLayers({ consumerId, spatialLayer, temporalLayer }) {
      await peer.consumers.get(consumerId).setPreferredLayers({ spatialLayer, temporalLayer });
    },
    async getStats({ consumerId }) {
      const stats = await peer.consumers.get(consumerId).getStats();
      return stats.map((s) => ({ type: s.type, byteCount: s.byteCount, bitrate: s.bitrate, score: s.score }));
    },
  };

  ws.on('message', async (raw) => {
    const msg = JSON.parse(raw.toString());
    try {
      const data = await handlers[msg.method](msg.data ?? {});
      ws.send(JSON.stringify({ id: msg.id, data }));
    } catch (e) {
      ws.send(JSON.stringify({ id: msg.id, error: e.message }));
    }
  });
  ws.on('close', () => {
    for (const t of peer.transports.values()) t.close(); // closes producers and consumers too
    if (peer.room) {
      peer.room.peers.delete(peer.id);
      broadcast(peer.room, peer.id, 'peerLeft', { peerId: peer.id });
    }
  });
});

server.listen(HTTP_PORT, () => console.log(`huddle-sfu http://localhost:${HTTP_PORT} rtc ${RTC_PORT}`));
process.on('SIGTERM', () => { worker.close(); process.exit(0); });

// Huddle SFU client built on mediasoup-client.
import { Device } from 'mediasoup-client';

export async function startHuddle({ room, name }) {
  const ws = new WebSocket(`ws://${location.host}/ws`);
  await new Promise((r) => (ws.onopen = r));

  const pending = new Map();
  const listeners = {};
  let seq = 0;
  ws.onmessage = (ev) => {
    const m = JSON.parse(ev.data);
    if (m.notify) return listeners[m.notify]?.(m.data);
    const p = pending.get(m.id); pending.delete(m.id);
    m.error ? p.reject(new Error(m.error)) : p.resolve(m.data);
  };
  const request = (method, data) => new Promise((resolve, reject) => {
    const id = ++seq; pending.set(id, { resolve, reject });
    ws.send(JSON.stringify({ id, method, data }));
  });

  const joined = await request('join', { room, name });
  const device = await Device.factory();
  await device.load({ routerRtpCapabilities: joined.routerRtpCapabilities });

  async function makeTransport(direction) {
    const params = await request('createTransport');
    const t = direction === 'send' ? device.createSendTransport(params) : device.createRecvTransport(params);
    t.on('connect', ({ dtlsParameters }, ok, fail) =>
      request('connectTransport', { transportId: t.id, dtlsParameters }).then(ok, fail));
    if (direction === 'send') {
      t.on('produce', ({ kind, rtpParameters }, ok, fail) =>
        request('produce', { transportId: t.id, kind, rtpParameters }).then(({ id }) => ok({ id }), fail));
    }
    t.on('connectionstatechange', (s) => (window.states ??= []).push(`${direction}:${s}`));
    return t;
  }

  const sendTransport = await makeTransport('send');
  const recvTransport = await makeTransport('recv');

  const stream = await navigator.mediaDevices.getUserMedia({ audio: true, video: { width: 1280, height: 720 } });
  const audio = await sendTransport.produce({ track: stream.getAudioTracks()[0] });
  const video = await sendTransport.produce({
    track: stream.getVideoTracks()[0],
    encodings: [
      { maxBitrate: 150_000, scaleResolutionDownBy: 4, scalabilityMode: 'L1T3' },
      { maxBitrate: 500_000, scaleResolutionDownBy: 2, scalabilityMode: 'L1T3' },
      { maxBitrate: 1_500_000, scalabilityMode: 'L1T3' },
    ],
  });

  const consumers = new Map();
  async function consumeProducer({ producerId, peerId }) {
    const info = await request('consume', { transportId: recvTransport.id, producerId, rtpCapabilities: device.recvRtpCapabilities });
    const consumer = await recvTransport.consume({ id: info.id, producerId, kind: info.kind, rtpParameters: info.rtpParameters });
    await request('resumeConsumer', { consumerId: consumer.id });
    consumers.set(consumer.id, { consumer, peerId });
    return consumer;
  }
  listeners.newProducer = consumeProducer;
  listeners.consumerClosed = ({ consumerId }) => { consumers.get(consumerId)?.consumer.close(); consumers.delete(consumerId); };
  listeners.activeSpeaker = ({ peerId }) => { window.activeSpeaker = peerId; };

  for (const e of joined.existing) await consumeProducer(e);

  return {
    peerId: joined.peerId, audio, video, consumers, request,
    async inboundBytes() {
      let total = 0;
      for (const { consumer } of consumers.values()) {
        for (const s of (await consumer.getStats()).values()) if (s.type === 'inbound-rtp') total += s.bytesReceived;
      }
      return total;
    },
  };
}

<!doctype html><meta charset="utf-8"><title>Huddle SFU</title>
<script type="module">
  import { startHuddle } from './huddle.js';
  window.startHuddle = startHuddle;
</script>

// build.mjs: bundle client.js and mediasoup-client into public/huddle.js
import { build } from 'esbuild';
await build({ stdin: { contents: "export { startHuddle } from './client.js'", resolveDir: '.' },
  bundle: true, format: 'esm', outfile: 'public/huddle.js', target: 'es2022' });

import puppeteer from 'puppeteer-core';
const browser = await puppeteer.launch({
  executablePath: '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',
  headless: 'new',
  args: ['--use-fake-device-for-media-stream', '--use-fake-ui-for-media-stream', '--no-first-run'],
});
const open = async (name) => {
  const page = await browser.newPage();
  page.on('console', (m) => console.log(`[${name}]`, m.text()));
  page.on('pageerror', (e) => console.log(`[${name}] pageerror`, e.message));
  await page.goto('http://localhost:41200/');
  await page.evaluate((n) => startHuddle({ room: 'standup', name: n }).then((h) => (window.h = h)), name);
  return page;
};
const a = await open('alice');
const b = await open('bob');
await new Promise((r) => setTimeout(r, 4000));
for (const [n, p] of [['alice', a], ['bob', b]]) {
  const r = await p.evaluate(async () => {
    const first = await h.inboundBytes();
    await new Promise((r) => setTimeout(r, 2000));
    const second = await h.inboundBytes();
    const kinds = [...h.consumers.values()].map((c) => c.consumer.kind);
    return { peerId: h.peerId, consumers: kinds, first, second, states: window.states, speaker: window.activeSpeaker };
  });
  console.log(n, JSON.stringify(r));
}
// layer selection and server-side stats on alice's video consumer
const r2 = await a.evaluate(async () => {
  const [id] = [...h.consumers].filter(([, v]) => v.consumer.kind === 'video').map(([k]) => k);
  await h.request('setPreferredLayers', { consumerId: id, spatialLayer: 0, temporalLayer: 2 });
  await new Promise((r) => setTimeout(r, 2500));
  return h.request('getStats', { consumerId: id });
});
console.log('consumer stats', JSON.stringify(r2));
await b.close();
await new Promise((r) => setTimeout(r, 500));
console.log('after bob left alice consumers:', await a.evaluate(() => h.consumers.size));
await browser.close();

import os from 'node:os';
import * as mediasoup from 'mediasoup';

const workers = [];
for (let i = 0; i < os.availableParallelism(); i++) {
  const worker = await mediasoup.createWorker({ logLevel: 'warn' });
  const webRtcServer = await worker.createWebRtcServer({
    listenInfos: [
      { protocol: 'udp', ip: '0.0.0.0', announcedAddress: process.env.ANNOUNCED_ADDRESS, port: 41201 + i },
      { protocol: 'tcp', ip: '0.0.0.0', announcedAddress: process.env.ANNOUNCED_ADDRESS, port: 41201 + i },
    ],
  });
  worker.on('died', () => process.exit(1)); // let the supervisor restart the whole process
  workers.push({ worker, webRtcServer, rooms: 0 });
}

export function pickWorker() {
  const best = workers.reduce((a, b) => (a.rooms <= b.rooms ? a : b));
  best.rooms++;
  return best;
}