Pion and aiortc

// Pixel Arena server peer with Pion: answers a browser offer over HTTP, echoes a VP8 video track
// back to the sender and answers player inputs on an unreliable data channel.
package main

import (
	"encoding/json"
	"log"
	"net/http"
	"os"
	"strconv"
	"sync/atomic"
	"time"

	"github.com/pion/ice/v4"
	"github.com/pion/interceptor"
	"github.com/pion/rtcp"
	"github.com/pion/webrtc/v4"
)

var peers atomic.Int32

const maxPeers = 8

func newAPI() (*webrtc.API, error) {
	se := webrtc.SettingEngine{}
	lo, _ := strconv.Atoi(os.Getenv("UDP_MIN"))
	hi, _ := strconv.Atoi(os.Getenv("UDP_MAX"))
	if lo > 0 && hi >= lo {
		if err := se.SetEphemeralUDPPortRange(uint16(lo), uint16(hi)); err != nil {
			return nil, err
		}
	}
	if ip := os.Getenv("PUBLIC_IP"); ip != "" { // address the browser can reach (Docker, NAT)
		se.SetNAT1To1IPs([]string{ip}, webrtc.ICECandidateTypeHost)
	}
	se.SetICEMulticastDNSMode(ice.MulticastDNSModeDisabled) // disabled: containers cannot answer .local queries

	m := &webrtc.MediaEngine{}
	if err := m.RegisterDefaultCodecs(); err != nil {
		return nil, err
	}
	ir := &interceptor.Registry{}
	if err := webrtc.RegisterDefaultInterceptors(m, ir); err != nil { // NACK, RTCP reports, TWCC
		return nil, err
	}
	return webrtc.NewAPI(webrtc.WithSettingEngine(se), webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(ir)), nil
}

func handleOffer(api *webrtc.API) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			http.Error(w, "POST only", http.StatusMethodNotAllowed)
			return
		}
		if peers.Load() >= maxPeers {
			http.Error(w, "arena full", http.StatusServiceUnavailable)
			return
		}
		var offer webrtc.SessionDescription
		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&offer); err != nil || offer.Type != webrtc.SDPTypeOffer {
			http.Error(w, "expected an SDP offer", http.StatusBadRequest)
			return
		}

		pc, err := api.NewPeerConnection(webrtc.Configuration{})
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		peers.Add(1)
		var rtpPackets, rtpBytes atomic.Uint64

		// The echo track must exist before the answer is created so that it is part of it.
		echo, _ := webrtc.NewTrackLocalStaticRTP(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, "video", "arena-echo")
		sender, err := pc.AddTrack(echo)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		go func() { // RTCP from the browser must be read, or the interceptors stall
			buf := make([]byte, 1500)
			for {
				if _, _, err := sender.Read(buf); err != nil {
					return
				}
			}
		}()

		pc.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
			if track.Kind() != webrtc.RTPCodecTypeVideo {
				return
			}
			go func() { // ask for a key frame now and then, so late joiners can start decoding
				for range time.Tick(2 * time.Second) {
					if pc.ConnectionState() >= webrtc.PeerConnectionStateClosed {
						return
					}
					_ = pc.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}})
				}
			}()
			for {
				pkt, _, err := track.ReadRTP()
				if err != nil {
					return
				}
				rtpPackets.Add(1)
				rtpBytes.Add(uint64(len(pkt.Payload)))
				_ = echo.WriteRTP(pkt) // the simplest SFU: one input, one output, no copy of the payload
			}
		})

		pc.OnDataChannel(func(dc *webrtc.DataChannel) {
			if dc.Label() != "arena" {
				_ = dc.Close()
				return
			}
			tick := 0
			dc.OnMessage(func(msg webrtc.DataChannelMessage) {
				var in struct {
					Seq int `json:"seq"`
				}
				if json.Unmarshal(msg.Data, &in) != nil {
					return
				}
				tick++
				reply, _ := json.Marshal(map[string]any{"type": "snapshot", "tick": tick, "ackSeq": in.Seq})
				_ = dc.SendText(string(reply)) // Send would deliver a binary message
				if tick%20 == 0 {
					st, _ := json.Marshal(map[string]any{"type": "rtp", "packets": rtpPackets.Load(), "bytes": rtpBytes.Load()})
					_ = dc.SendText(string(st))
				}
			})
		})

		pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
			log.Printf("peer state: %s", s)
			if s == webrtc.PeerConnectionStateFailed || s == webrtc.PeerConnectionStateClosed {
				if peers.Add(-1) >= 0 {
					_ = pc.Close()
				}
			}
		})

		if err := pc.SetRemoteDescription(offer); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			_ = pc.Close()
			return
		}
		answer, err := pc.CreateAnswer(nil)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		gathered := webrtc.GatheringCompletePromise(pc) // non-trickle: candidates go inside the answer
		if err := pc.SetLocalDescription(answer); err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		<-gathered
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(pc.LocalDescription())
	}
}

func main() {
	api, err := newAPI()
	if err != nil {
		log.Fatal(err)
	}
	mux := http.NewServeMux()
	mux.Handle("/", http.FileServer(http.Dir("web")))
	mux.HandleFunc("/offer", handleOffer(api))
	addr := os.Getenv("ADDR")
	if addr == "" {
		addr = ":8080"
	}
	log.Printf("listening on %s", addr)
	log.Fatal(http.ListenAndServe(addr, mux))
}

FROM golang:1.25 AS build
WORKDIR /src
COPY go.mod ./
COPY main.go ./
RUN go mod tidy && CGO_ENABLED=0 go build -o /arena .

FROM gcr.io/distroless/static-debian12
COPY --from=build /arena /arena
COPY web /web
WORKDIR /
USER nonroot
ENTRYPOINT ["/arena"]

"""Huddle recording bot with aiortc: answers a browser offer, records the call and echoes inputs."""
import asyncio
import json
import os
import random
import re
from pathlib import Path

import av
from aiohttp import web
from aiortc import RTCPeerConnection, RTCSessionDescription
from aiortc.contrib.media import MediaRecorder

RECORDINGS = Path(os.environ.get("RECORDINGS", "/recordings"))
PUBLIC_IP = os.environ.get("PUBLIC_IP")          # address the browser can reach (Docker, NAT)
PORT_RANGE = os.environ.get("UDP_PORTS")         # for example "41520-41529"; aiortc has no such option

if PORT_RANGE:  # pin the ICE sockets to a published range: a small, test-only patch of the event loop
    lo, hi = (int(p) for p in PORT_RANGE.split("-"))
    _create = asyncio.BaseEventLoop.create_datagram_endpoint

    async def _pinned(self, factory, local_addr=None, **kw):
        if local_addr and local_addr[1] == 0:
            ports = list(range(lo, hi + 1))
            random.shuffle(ports)
            for port in ports:
                try:
                    return await _create(self, factory, local_addr=(local_addr[0], port), **kw)
                except OSError:
                    continue
        return await _create(self, factory, local_addr=local_addr, **kw)

    asyncio.BaseEventLoop.create_datagram_endpoint = _pinned

peers: dict[str, dict] = {}


def advertise(sdp: str) -> str:
    """Replace the container address in host candidates by the address the browser can reach."""
    if not PUBLIC_IP:
        return sdp
    return re.sub(r"(a=candidate:\S+ \d+ udp \d+ )\S+( \d+ typ host)", rf"\g<1>{PUBLIC_IP}\g<2>", sdp)


async def offer(request: web.Request) -> web.Response:
    body = await request.json()
    if body.get("type") != "offer" or len(body.get("sdp", "")) > 64_000:
        raise web.HTTPBadRequest(text="expected an SDP offer")
    if len(peers) >= 8:
        raise web.HTTPServiceUnavailable(text="arena full")

    pc = RTCPeerConnection()
    pid = os.urandom(4).hex()
    recorder = MediaRecorder(str(RECORDINGS / f"{pid}.mp4"))
    peers[pid] = {"pc": pc, "recorder": recorder}

    @pc.on("track")
    def on_track(track):
        recorder.addTrack(track)          # video and audio are muxed into one MP4

    @pc.on("datachannel")
    def on_datachannel(channel):
        if channel.label != "arena":
            channel.close()
            return
        tick = 0

        @channel.on("message")
        def on_message(message):
            nonlocal tick
            inp = json.loads(message)
            tick += 1
            channel.send(json.dumps({"type": "snapshot", "tick": tick, "ackSeq": inp["seq"]}))

    @pc.on("connectionstatechange")
    async def on_state():
        if pc.connectionState == "connected":
            await recorder.start()
        elif pc.connectionState in ("failed", "closed"):
            peers.pop(pid, None)

    await pc.setRemoteDescription(RTCSessionDescription(sdp=body["sdp"], type="offer"))
    await pc.setLocalDescription(await pc.createAnswer())   # aiortc gathers candidates here
    return web.json_response({"type": "answer", "sdp": advertise(pc.localDescription.sdp), "id": pid})


async def finish(request: web.Request) -> web.Response:
    """Stop recording, close the peer and report what the MP4 contains."""
    pid = request.match_info["pid"]
    peer = peers.pop(pid, None)
    if peer is None:
        raise web.HTTPNotFound()
    await peer["recorder"].stop()
    await peer["pc"].close()
    path = RECORDINGS / f"{pid}.mp4"
    with av.open(str(path)) as container:
        kinds = {s.type: s.frames for s in container.streams}
        duration = float(container.duration or 0) / av.time_base
    return web.json_response({"file": path.name, "bytes": path.stat().st_size, "streams": kinds, "seconds": round(duration, 2)})


async def close_all(app: web.Application) -> None:
    for peer in list(peers.values()):
        await peer["pc"].close()


def make_app() -> web.Application:
    RECORDINGS.mkdir(parents=True, exist_ok=True)
    app = web.Application(client_max_size=64 * 1024)
    app.router.add_post("/offer", offer)
    app.router.add_post("/finish/{pid}", finish)
    app.router.add_static("/", Path(__file__).parent / "web")
    app.on_shutdown.append(close_all)
    return app


if __name__ == "__main__":
    web.run_app(make_app(), host="0.0.0.0", port=int(os.environ.get("PORT", "8080")))

aiortc==1.15.0
aiohttp==3.14.5

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY bot.py .
COPY web web
RUN useradd -m bot && mkdir /recordings && chown bot /recordings
USER bot
CMD ["python", "bot.py"]

// Pixel Arena server peer with node-datachannel's native API (libdatachannel), non-trickle HTTP signalling.
import http from "node:http";
import { readFile } from "node:fs/promises";
import nodeDataChannel from "node-datachannel";

const { PeerConnection } = nodeDataChannel;
const PORT = Number(process.env.PORT ?? 8080);
const peers = new Set();

function answerOffer(offer) {
  return new Promise((resolve, reject) => {
    const pc = new PeerConnection(`arena-${peers.size}`, {
      iceServers: [],
      portRangeBegin: Number(process.env.UDP_MIN ?? 0),
      portRangeEnd: Number(process.env.UDP_MAX ?? 0),
    });
    peers.add(pc);
    const timer = setTimeout(() => reject(new Error("ICE gathering timed out")), 5000);

    pc.onStateChange((state) => {
      if (state === "failed" || state === "closed") { peers.delete(pc); pc.close(); }
    });
    pc.onGatheringStateChange((state) => {
      if (state !== "complete") return;
      clearTimeout(timer);
      resolve(pc.localDescription());                // {type: "answer", sdp: "..."} with the candidates inside
    });
    pc.onDataChannel((dc) => {
      if (dc.getLabel() !== "arena") return dc.close();
      let tick = 0;
      dc.onMessage((msg) => {
        const input = JSON.parse(msg.toString());
        dc.sendMessage(JSON.stringify({ type: "snapshot", tick: ++tick, ackSeq: input.seq }));
      });
    });
    pc.setRemoteDescription(offer.sdp, offer.type);   // libdatachannel then generates the answer by itself
  });
}

const server = http.createServer(async (req, res) => {
  if (req.method === "POST" && req.url === "/offer") {
    let body = "";
    for await (const chunk of req) {
      body += chunk;
      if (body.length > 64 * 1024) return res.writeHead(413).end();
    }
    try {
      const answer = await answerOffer(JSON.parse(body));
      res.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify(answer));
    } catch (err) {
      res.writeHead(400).end(String(err.message));
    }
    return;
  }
  if (req.method === "GET" && req.url.startsWith("/arena.html")) {
    return res.writeHead(200, { "Content-Type": "text/html" }).end(await readFile(new URL("./web/arena.html", import.meta.url)));
  }
  res.writeHead(404).end();
});
server.listen(PORT, "127.0.0.1", () => console.log(`listening on ${PORT}`));
process.on("SIGINT", () => { for (const pc of peers) pc.close(); nodeDataChannel.cleanup(); process.exit(0); });

{ "type": "module",
  "dependencies": { "node-datachannel": "0.33.4" } }