doloro/simple-instant-stream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d4b77bd71714971269691995fc59beb2afdee868
simple-instant-stream/AGENTS.md
T

25 KiB
Raw Blame History

AGENTS.md

This file provides comprehensive guidance for AI agents working with this repository. It is the canonical reference; CLAUDE.md may be a subset of this.

Project Summary

RTMP-to-WHIP/WHEP bridge. Accepts RTMP video+audio publish streams and re-streams to browsers via WebRTC (WHEP). Also has a stub for WHIP ingest.

Binary name: rtmp-to-whip Workspace: Rust 2024 edition, 3 crates (server, entity, migration) No automated tests exist (except one ignored FLV replay test).

Commands

# Build (all crates)
cargo build

# Run (server binary)
cargo run

# Build (Nix)
nix build

# Dev shell (provides clang + mold linker + sea-orm-cli + opus + just)
nix develop

Workspace Layout

crates/
  server/     — main binary (RTMP, HTTP, WebRTC, codecs, audio)
  entity/     — SeaORM entity definitions (users, stream_key, stream_session, auth_session)
  migration/  — SeaORM migrations (4 files)
target/       — build output (gitignored)
db/           — SQLite database at runtime (db/db.sqlite, gitignored)
index.html    — WHEP test page (browser-based viewer)

Ports & Networking

Port Protocol Purpose
1935 TCP RTMP ingest (OBS/encoder)
3000 TCP HTTP API (axum)
6969 UDP WebRTC media (configurable via RTC_PORT env)

Environment Variables

Variable Default Purpose
RTC_PORT 6969 UDP port for WebRTC media traffic
PUBLIC_DOMAIN — Domain for ICE candidates (DNS-resolved); falls back to STUN discovery if unset
SIGNUP_CODE — Required token for /api/user signup; empty = disabled
RUST_LOG — Tracing filter (e.g. info,warn)

Entry Point — crates/server/src/main.rs

Initializes all components, spawns 4 workers in a JoinSet:

workers.spawn(http.run());
workers.spawn(proxy.run());
workers.spawn(webrtc.run());
workers.spawn(rtmp.run());

Shuts down on Ctrl-C or any worker exit (workers.abort_all()).

AppState — Arc<Mutex<AppState>> wrapping DashMap<i32, StreamSession>. Keyed by stream_key.id.

StreamSession fields:

  • stream_key_id: i32
  • stream_key_label: String
  • frame_channel: async_broadcast::Sender<Arc<VideoFrame>>
  • audio_channel: async_broadcast::Sender<Arc<OpusAudioFrame>>
  • codec: Option<StreamCodec> — H264 | H265 | AV1
  • started_at: DateTime<Utc>
  • viewers: AtomicU32

Shared channels:

  • offer_tx/offer_rx: mpsc::channel<(i32, i32, String)>(64) — HTTP → WebRTC (request_id, stream_key_id, sdp_offer)
  • answer_tx/answer_rx: broadcast::<(i32, Option<String>)>(64) — WebRTC → HTTP (request_id, answer_or_none)

DB: sqlite://./db/db.sqlite?mode=rwc. Migrations run at startup via Migrator::up. stream_session::Model::clean_unended_streams() repairs crash-residual sessions.

Architecture — Signal Flow

OBS/encoder  →  RTMP (port 1935)  →  H264Parser / H265Parser / Av1Parser  →  async_broadcast frame channel
                                                                                     ↓
Browser  ←  WebRTC/UDP  ←  str0m Rtc  ←  WHEP HTTP POST /api/stream/{slug}
                                  ↑
                         WebrtcProxy (UDP port from RTC_PORT env, default 6969)

WHIP ingest:  Browser/encoder  →  WHIP POST /api/whip  →  (stub: parses SDP, logs media)

RTMP Ingestion — crates/server/src/rtmp.rs

pub struct Rtmp { listener, stream_sessions, db }

async fn run(self) — accepts TCP connections on port 1935, spawns one task per connection:

  1. Handshake — rml_rtmp::Handshake (C0+C1 → S0+S1+S2 → C2), 30s timeout.
  2. Session setup — rml_rtmp::ServerSession.
  3. Event loop — 4096-byte reads, 15s idle timeout:
    • ConnectionRequested → accepted unconditionally.
    • PublishStreamRequested → DB lookup of stream key; rejects if not found or already live. Inserts StreamSession into AppState + DB row.
    • VideoDataReceived → dispatched to CodecParser::parse (H.264, H.265, or AV1). Frame broadcast on frame_channel.
    • AudioDataReceived → AACParser decodes AAC → PCM, AudioProcesser resamples + encodes Opus. OpusAudioFrame broadcast on audio_channel.
    • PublishStreamFinished → removes StreamSession from AppState, sets ended_at in DB.

Codec detection (parse_video_codec):

  • Enhanced RTMP (byte 0 bit 7 = 1): bytes 1–4 = FourCC (hvc1→H.265, avc1→H.264, av01→AV1).
  • Legacy RTMP (byte 0 bit 7 = 0): nibble & 0x0F == 7 → H.264.
  • StreamSession.codec is stamped on first VideoDataReceived and never changed.

HTTP API — crates/server/src/http.rs

pub struct HttpServer fields: offer_tx, accept_rx, appstate, request_count: AtomicI32, db, config.

Serves on 0.0.0.0:3000 with CORS for http://localhost:5173 and https://stream.h.doloro.co.uk.

Routes

Method Path Handler Auth? Notes
GET /api/catalog catalog_handler No Returns { active_streams: [...] } from DB + AppState
POST /api/user create_user_handler No Signup; requires SIGNUP_CODE header
POST /api/login login_handler No Verifies password, creates auth_session, sets cookie
GET /api/stream-key get_all_stream_keys Yes Returns user's stream keys
POST /api/stream-key create_stream_key_handler Yes Creates key with UUID value (stream-key-<uuid>)
PATCH /api/stream-key edit_stream_key Yes Edits label (validated by KEY_RE regex)
POST /api/whip handle_whip_injest No Stub — parses SDP, logs media lines
POST /api/stream/{slug} stream_handler No WHEP offer; forwards to WebRTC, waits for answer
GET /api/meow meow_handler No Returns "meow"

Auth

  • Session token sent as session cookie or header.
  • AuthUser extractor: looks up auth_session by token value, then joins to users.
  • Protected routes require AuthUser.
  • Cookie flags: HttpOnly, Path=/, SameSite=None (dev) or Lax (prod).
  • DevFlag extractor: reads ?dev=1 query param.

Validation

  • Username: max 32 chars, non-empty.
  • Stream key label: 1–64 chars, regex ^[A-Za-z0-9 _-]{1,67}$, must contain at least one letter.
  • User has stream_key_limit (default 3) enforced at creation time.

WebRTC Proxy — crates/server/src/webrtc_proxy.rs

pub struct WebrtcProxy — all fields Arc-wrapped, Clone-able.

  • clients_ufrag: DashMap<String, mpsc::Sender<(Bytes, SocketAddr)>> — pending ICE ufrag → per-client channel
  • clients_addr: DashMap<SocketAddr, mpsc::Sender<(Bytes, SocketAddr)>> — established addr → per-client channel
  • socket: Arc<UdpSocket> — shared UDP socket bound to 0.0.0.0:{RTC_PORT}
  • public_addr: SocketAddr — from PUBLIC_DOMAIN DNS or STUN discovery

async fn run(self) — UDP receive loop:

  1. Receives datagrams on shared socket.
  2. If source addr in clients_addr, forwards to that client.
  3. Otherwise parses STUN binding request to extract ufrag (username attribute, split on :), looks up clients_ufrag, promotes to clients_addr, forwards.

fn add_client(ufrag) — called by Webrtc when setting up peer connection. Returns (Arc<UdpSocket>, Receiver<(Bytes, SocketAddr)>).

Public IP discovery:

  • If PUBLIC_DOMAIN env set: DNS-resolve for IPv4.
  • Else if cfg!(debug_assertions): returns 127.0.0.1 (for testing).
  • Else: sends STUN Binding Request to stun.l.google.com:19302, parses XOR-MAPPED-ADDRESS.

WebRTC Negotiation — crates/server/src/webrtc.rs

pub struct Webrtc { offer_rx, accept_tx, sessions_ref, db, proxy }

async fn run(mut self) — receives (request_id, stream_id, sdp_body) from offer_rx:

  1. Looks up stream_key in DB; rejects if not found.
  2. Gets codec from StreamSession.codec (determines which str0m codec config to enable).
  3. Configures str0m Rtc::builder:
    • Opus always enabled.
    • H.265: cc.enable_h265(true).
    • AV1: cc.enable_av1(true).
    • H.264 (default): adds 3 profiles — Constrained Baseline (0x42e01f), Main (0x4d001f), High (0x64001f). Firefox only offers Constrained Baseline — without it, no codec matches.
  4. Accepts SDP offer → produces SDP answer.
  5. Validates answer has non-empty video PT list (str0m returns empty list if no codec matched).
  6. Extracts ICE ufrag from answer, registers with proxy.
  7. Broadcasts answer on accept_tx.
  8. Spawns detach_connection per peer.

detach_connection per-peer loop:

  • Drains rtc.poll_output() — transmits via proxy UDP, handles events.
  • Event::MediaAdded — selects best video PT by profile_level_id (H.264), h265_profile_tier_level (H.265), or level_idx (AV1).
  • Event::Connected — starts subscribing to frame/audio channels from AppState.
  • tokio::select! on: str0m deadline, incoming UDP from proxy, video frame, audio frame.
  • Keyframe gating: drops non-keyframe frames until first keyframe is seen.
  • Overflow handling: on ring buffer overflow, resets saw_keyframe to wait for next keyframe.
  • Video RTP timestamp: MediaTime::from_90khz(ts * 90) where ts is VideoFrame timestamp in ms.
  • Audio RTP timestamp: MediaTime::new(ts * 48, Frequency::FORTY_EIGHT_KHZ).
  • Drains up to 7 pending frames per tick to catch up.

Codecs — crates/server/src/codec/

Trait — CodecParser

pub trait CodecParser: Send {
    fn parse(&mut self, data: &[u8], timestamp_ms: u32) -> Option<VideoFrame>;
}

VideoFrame

pub struct VideoFrame {
    pub data: Bytes,       // Annex-B elementary stream
    pub is_keyframe: bool,
    pub timestamp_ms: u32, // Presentation timestamp (DTS + CTS for B-frame streams)
}

H.264 — h264.rs

Handles both legacy and enhanced RTMP.

Legacy RTMP: byte 0 = (frame_type << 4) | codec_id, byte 1 = packet type, bytes 2–4 = CTS. Enhanced RTMP: byte 0 = 0x80 | (frame_type << 4) | packet_type, bytes 1–4 = FourCC.

  • Packet type 0: parses AVCDecoderConfigurationRecord, caches SPS+PPS.
  • Packet type 1: AVCC → Annex-B conversion, prepends SPS+PPS before keyframes. Uses PTS = DTS + CTS to avoid B-frame stuttering.
  • Packet type 3: same as type 1 but no CTS field.

H.265 — h265.rs

Enhanced RTMP only (FourCC hvc1).

  • Packet type 0: bytes[5..] = HEVCDecoderConfigurationRecord.
  • Packet type 1: bytes 5–7 = CTS, bytes 8+ = HVCC NALUs. Uses PTS = DTS + CTS.
  • Packet type 3: bytes 5+ = HVCC NALUs, no CTS.

HEVCDecoderConfigurationRecord parsing: skips first 22 bytes (profile/level/tier), parses arrays at byte 22, caches VPS (NAL type 32), SPS (33), PPS (34).

AV1 — av1.rs

Enhanced RTMP only (FourCC av01).

  • Packet type 0: bytes[5..] = AV1CodecConfigurationRecord, extracts configOBUs from payload[4..].
  • Packet type 1/3: bytes[5..] = OBU stream.

Keyframe detection:

  1. RTMP FrameType=1.
  2. Scan OBUs for OBU_SEQUENCE_HEADER (obu_type=1).
  3. First coded frame after config (bootstrap fallback — OBS may never set FrameType=1 for AV1).

Config OBUs prepended to keyframe payloads so str0m's Av1Packetizer sees a Sequence Header OBU.

FLV Replay Test — flv_replay_test.rs

Ignored test harness for codec validation:

FLV_IN=test.flv ES_OUT=output.es cargo test -p server flv_replay -- --ignored --nocapture

Parses FLV video tags, runs through codec parser, dumps Annex-B elementary stream with temporal delimiter OBUs for ffprobe validation. Checks for duplicate PTS values.

Audio — crates/server/src/audio.rs

AACParser

Parses RTMP audio payloads:

  • Byte 0: >> 4 == 10 → AAC codec.
  • Byte 1: 0 = AudioSpecificConfig → initializes Symphonia AAC decoder with extra_data.
  • Byte 1: 1 = raw AAC frame → decodes via Symphonia → interleaved f32 PCM.

AudioFrame

pub struct AudioFrame {
    pub data: Bytes,       // interleaved f32 PCM, little-endian
    pub timestamp_ms: u32,
    pub sample_rate: u32,  // typically 44100 Hz from AAC-LC
}

AudioProcesser

AAC → Opus transcoding pipeline:

  • Resampler: rubato::Fft from 44100 Hz → 48000 Hz (if needed).
  • Encoder: opus::Encoder — 48kHz stereo, LowDelay mode.
  • Frame size: 960 samples per channel (20ms), 1920 interleaved.
  • Timestamp: monotonic 48kHz counter: timestamp_ms = samples_emitted / 48 (independent of RTMP timestamps).
  • Buffer: accumulates resampled PCM until full Opus frame, then encodes and emits OpusAudioFrame.

OpusAudioFrame

pub struct OpusAudioFrame {
    pub data: Bytes,
    pub timestamp_ms: u32,
}

Password Hashing — crates/server/src/hash.rs

  • hash_password(password) — Argon2id with random OsRng salt, returns PHC-format string.
  • verify_password(password, hash) — parses PHC string, verifies with Argon2. Panics if hash is not valid PHC format (uses .unwrap() on PasswordHash::new).

Error Handling — crates/server/src/http_error.rs

#[derive(Error, Debug)] pub enum HttpError { ... }

Maps to HTTP status codes:

  • DbErr/Hash/Internal → 500
  • NotFound → 404
  • Unauthorized → 401
  • Forbidden → 403
  • Conflict → 409
  • BadRequest → 400
  • Unprocessable → 422
  • NotAcceptable → 406

4xx errors return message body; 5xx returns empty body (no internal details leaked).

WebRTC Ingest (WHIP) — crates/server/src/webrtc_ingest.rs

STUB — not implemented. Currently:

  • Accepts POST /api/whip with SDP body.
  • Parses SDP via str0m::change::SdpOffer.
  • Logs media lines.

Database (SeaORM + SQLite)

DB file: ./db/db.sqlite Migrations: crates/migration/ (4 migrations)

Tables

users

Column Type Constraints
id INTEGER PK, autoincrement
username TEXT NOT NULL, UNIQUE
hashed_password TEXT NOT NULL
stream_key_limit INTEGER NOT NULL, default 3

Relations: has_many → stream_key, auth_session

Custom methods:

  • Entity::create(db, username, password_hash) — inserts user
  • Entity::find_by_username(db, username) — lookup
  • Entity::find_by_auth_session(db, token) — join auth_session → user
  • ActiveModel::update_username, update_password, change_stream_key_limit

stream_key

Column Type Constraints
id INTEGER PK, autoincrement
key_value TEXT NOT NULL, UNIQUE
user_id INTEGER NOT NULL, FK → users
label TEXT NOT NULL
is_active BOOLEAN NOT NULL, default true
is_unlisted BOOLEAN NOT NULL, default true
created_at DATETIME NOT NULL

Relations: belongs_to → users, has_many → stream_session

Custom methods:

  • Entity::create(db, user_id, key_value, label, is_unlisted) — generates stream-key-<uuid> as key_value
  • Entity::find_by_key(db, key_value) — lookup by raw key string
  • Entity::find_by_user(db, user_id) — list user's keys
  • ActiveModel::change_label_value

stream_session

Column Type Constraints
id INTEGER PK, autoincrement
stream_key_id INTEGER NOT NULL, FK → stream_key
started_at DATETIME NOT NULL
ended_at DATETIME nullable

Relations: belongs_to → stream_key

Custom methods:

  • Model::create_stream_session(db, stream_key_id, started_at) — creates open session
  • Model::get_stream_session(db, id) — get by ID
  • Model::get_all_active_sessions(db) — all sessions where ended_at IS NULL
  • Model::get_active_by_stream_key_id(db, key_id) — single active session
  • Model::clean_unended_streams(db) — sets ended_at = started_at for crash recovery
  • ActiveModel::finish_stream_session(db, ended_at) — marks session ended

auth_session

Column Type Constraints
id INTEGER PK, autoincrement
id_user INTEGER NOT NULL, FK → users
value TEXT UUID v4 string

Relations: belongs_to → users

Custom methods:

  • Entity::create(db, user_id) — generates UUID token, creates session
  • Entity::find_by_user_id(db, user_id) — lookup by user

Conventions

  • Query methods on Entity (e.g. Entity::find_by_x).
  • Mutation helpers on ActiveModel.
  • For destructive schema changes: expand-contract pattern (add → backfill → switch code → drop old).

Dependencies

Server (crates/server/Cargo.toml)

Crate Version Purpose
tokio 1 Async runtime (full)
str0m 0.20.0 WebRTC
rml_rtmp 0.8.0 RTMP server
axum 0.8 HTTP framework
axum-extra 0.12.6 Cookie extraction
sea-orm 1 ORM (SQLite, tokio-rustls)
symphonia 0.5 Audio decode (AAC)
opus 0.3.1 Audio encode (Opus)
rubato 3.0.0 Audio resampling
argon2 0.5.3 Password hashing
uuid 1.23.3 UUID v4
dashmap 6.2.1 Concurrent map
async-broadcast 0.7.2 Ring buffer channels
tower-http 0.6 CORS
regex 1 Stream key label validation
thiserror 2.0.18 Error types
tracing 0.1 Structured logging
chrono 0.4.45 Date/time
bytes 1 Byte buffers
futures 0.3.32 async block_on
serde 1.0.228 Serialization

Entity (crates/entity/Cargo.toml)

Crate Version Purpose
sea-orm 1 ORM macros
serde 1 Serialization
chrono 0.4 DateTime
rand 0.10.1 Random salts
argon2 0.5.3 Password hash
uuid 1.23.3 UUID v4

Migration (crates/migration/Cargo.toml)

Crate Version Purpose
sea-orm-migration 1 Migration CLI
tokio 1 Async runtime

Build & Release

Profiles (root Cargo.toml)

  • dev: debug = true
  • release: lto = true, codegen-units = 1, panic = "abort"
  • flamegraph: inherits release + debug + force-frame-pointers

Cross-compilation

  • x86_64: clang linker + mold (via rustflags).
  • aarch64: cross-toolchain (aarch64-linux-gnu-gcc), no mold.

Docker

  • x86_64: multi-stage, strips binary, copies only needed .so deps.
  • aarch64: QEMU cross-build from x86_64 builder, copies arm64 .so deps.
  • Both use scratch final stage.
  • Exposes: 1935/tcp, 3000/tcp, 6969/udp.

Nix

  • Uses crane for cargo builds.
  • Dev shell: clang, mold, sea-orm-cli, cmake, opus, pkgconf, just.
  • ADMIN_REF_CODE set in dev shell: meowmeowpurrrmeow.

WHEP Test Page — index.html

  • Connects to http://localhost:5000/whep/test (note: hardcoded to port 5000, not 3000).
  • Creates RTCPeerConnection with video recvonly transceiver.
  • Sends WHEP offer POST, receives SDP answer.
  • Streams video to <video> element.
  • Polls pc.getStats() every 500ms, displays:
    • Server-to-client latency (network + jitter + decode)
    • Network one-way RTT (from ICE candidate pair)
    • Jitter buffer delay
    • Decode latency
    • FPS
    • Packet loss ratio
    • Jitter

Key Implementation Details & Gotchas

  1. H.264 PT negotiation: str0m uses profile_level_id to match H.264 profiles. Firefox only offers Constrained Baseline (0x42e01f). All three profiles must be configured or Firefox sees no codec.

  2. PTS vs DTS: For both H.264 and H.265, presentation timestamp = DTS + CTS. B-frame streams stutter if DTS is used.

  3. AV1 keyframe detection: OBS may never set RTMP FrameType=1 for AV1. The parser uses triple fallback: RTMP flag, OBU scan, first-frame bootstrap.

  4. H.265 support marked "very shity" (sic): See StreamCodec enum doc comment in main.rs:56.

  5. No auth on WHEP endpoint: /api/stream/{slug} is publicly accessible — any client who knows the stream key slug can get a WHEP answer.

  6. Request ID tracking: AtomicI32::fetch_add(1, Relaxed) for WHEP request IDs. Monotonic, wraps at i32::MAX.

  7. Channel sizes: broadcast channels (frame/audio) = 32, offer channel = 64, answer channel = 64, proxy client channel = 256.

  8. RTMP 4096-byte read buffer: This is a fixed-size buffer. Large RTMP messages are split across multiple reads — rml_rtmp handles reassembly internally.

  9. Audio timestamp independence: Opus RTP timestamps use a monotonic 48kHz counter, not RTMP timestamps. This ensures stable playback even if source timestamps are irregular.

  10. Stream session cleanup: On any disconnect (error, timeout, finish), the cleanup closure removes from AppState and sets ended_at in DB. The clean_unended_streams at startup repairs crash-residual sessions.

README.md Todo List

Completed:

  • axum cookie jar support
  • H.265 codec support
  • AV1 codec support
  • Custom error types (thiserror + IntoResponse)

Incomplete:

  • Stream status via RTC data channel (frame drops, codec errors)
  • WHEP → WHIP support
  • Replace async_broadcast with circular buffer for frame → RTC management

QoL:

  • Stream key renaming (partially implemented via edit_stream_key)
  • Admin panel (username lookup, disable stream keys, change limits, CPU usage)

File Index

Server source (crates/server/src/)

  • main.rs — entry point, AppState, StreamSession, worker spawning
  • http.rs — HTTP server, all route handlers, auth extractors
  • http_error.rs — HttpError enum + IntoResponse impl
  • rtmp.rs — RTMP listener, handshake, codec dispatch, frame broadcasting
  • webrtc.rs — WebRTC negotiation, str0m Rtc, RTP sending, keyframe gating
  • webrtc_proxy.rs — UDP proxy, STUN parsing, public IP discovery
  • webrtc_ingest.rs — WHIP stub handler
  • audio.rs — AACParser, AudioProcesser (resample + Opus encode)
  • hash.rs — Argon2id password hashing
  • codec/mod.rs — CodecParser trait, VideoFrame struct
  • codec/h264.rs — H.264 AVCC/legacy → Annex-B
  • codec/h265.rs — H.265 HVCC → Annex-B (enhanced RTMP only)
  • codec/av1.rs — AV1 OBU parsing (enhanced RTMP only)
  • codec/flv_replay_test.rs — ignored test for codec validation

Entity source (crates/entity/src/)

  • lib.rs — module exports
  • prelude.rs — entity type aliases
  • users.rs — User entity + methods
  • stream_key.rs — StreamKey entity + methods
  • stream_session.rs — StreamSession entity + methods
  • auth_session.rs — AuthSession entity + methods

Migration source (crates/migration/src/)

  • lib.rs — Migrator trait, migration registry
  • main.rs — migration binary entry
  • m20260616_000001_create_users.rs — users table
  • m20260616_000002_create_stream_key.rs — stream_key table
  • m20260616_000003_create_stream_session.rs — stream_session table
  • m20260616_000004_create_auth_session.rs — auth_session table

Config

  • Cargo.toml (root) — workspace, profiles
  • crates/server/Cargo.toml — server dependencies
  • crates/entity/Cargo.toml — entity dependencies
  • crates/migration/Cargo.toml — migration dependencies
  • Dockerfile — x86_64 multi-stage build
  • Dockerfile.aarch64 — arm64 cross-build
  • compose.yaml — docker-compose deployment
  • flake.nix — Nix build + dev shell
  • Justfile — Docker build/push shortcuts
  • .gitignore — target, db, devenv files
Reference in New Issue View Git Blame Copy Permalink