🎬 arcly-stream

A high-performance, open-extensible live-media streaming kernel for Rust

Lock-free zero-copy fan-out · instant-start GOP replay · a pluggable multi-protocol ingestion layer · a feature-gated pure-Rust media plane — with zero baked-in config, metrics singleton, or HTTP runtime.

👀 At a Glance

Every streaming server re-solves the same hard core: fan one publisher's frames out to thousands of late-joining subscribers — without copying payloads, without locks on the hot path, and without silently dropping the slow ones. Everything else — which wire protocol you ingest, how you package egress, where you store, how you authorize and observe — is deployment-specific.

arcly-stream is exactly that core, extracted and hardened into an embeddable kernel. You implement a few small traits for the parts unique to your system; the kernel owns the lock-free pub/sub bus, instant-start replay, back-pressure, lifecycle, and QoS. It ships no opinion about your runtime: no global registry, no TOML schema, no HTTP server.

use arcly_stream::prelude::*;

let engine = Engine::builder()
    .max_publishers(10_000)
    .application(AppSpec::new("live").gop_cache(120)) // instant-start replay
    .build();
assert_eq!(engine.list_apps().len(), 1);

📦 Capability Matrix

✅ Core/feature = shipped, native-Rust, unsafe-free. 🧭 Seam = a documented public trait you implement in your own crate; the kernel never drags in a native transport or codec you didn't ask for.

WebRTC scope note: the webrtc feature ships the WHIP/WHEP signaling (SDP offer/answer), RTP/RTCP routing, and PLI/FIR feedback natively. The DTLS/SRTP/ICE crypto transport is injected by the host through the DtlsSrtpTransport trait (back it with str0m/webrtc-rs) — so the kernel stays crypto-free and #![forbid(unsafe_code)]. SRT ships unencrypted (handshake + MPEG-TS demux); AES-encrypted SRT is out of scope.

🏛 Architecture

Three tiers — an always-on Open Kernel, an Injected Multi-Protocol Layer, and a Feature-Gated Media Plane:

                       ┌─────────────────────────────────────────────────┐
                       │                  OPEN KERNEL                     │
                       │                 (always compiled)                │
   publisher ────────▶ │  Engine · StreamHandle                          │ ─────▶ subscribers
 (any protocol)        │   • broadcast fan-out (zero-copy Arc<Frame>)    │   (packager / SFU /
                       │   • GOP cache → instant-start replay            │    recorder / edge)
                       │   • live QoS (bitrate / fps)                    │
                       │  PublishRegistry · PlaybackRegistry · EventBus  │
                       │  Observer · StreamAuthenticator · AuditSink     │
                       └─────────────────────────────────────────────────┘
                          ▲ implements                       ▲ feature-gated, pure Rust
        ┌─────────────────┴───────────────────┐  ┌───────────┴──────────────────────────┐
        │   INJECTED MULTI-PROTOCOL LAYER      │  │           MEDIA PLANE                 │
        │        (trait: InboundProtocol)      │  │          (opt-in features)            │
        │  RtmpHandler   reference impl  ✅    │  │  MpegTsMuxer   playable .ts     ✅    │
        │  IngestContext · PublishSession      │  │  codec  H.264/265 · AV1/VP9/VVC · AAC │
        │  ── your crate ──────────────────    │  │  hls    Packager + segmenter          │
        │  MyRtspHandler · MyWhipHandler …     │  │  record RecordingSink (VOD/DVR)       │
        └──────────────────────────────────────┘  └───────────────────────────────────────┘

A new wire protocol is one InboundProtocol impl in your own crate, registered on the builder — no kernel fork. RtmpHandler is just the first reference implementation of that public trait.

Workspace layout

arcly-stream/             the published library crate (this README documents it)
arcly-stream-macros/      optional proc-macros (the `macros` feature)
examples/
  minimal_ingest/         MediaSource + Observer, drive the bus
  custom_protocol/        legacy ProtocolHandler + Engine::serve
  custom_protocol_plugin/ custom InboundProtocol registered alongside RTMP
  enterprise_gateway/     RTMP + token auth + FS recording + metrics + audit
  multi_protocol_gateway/ RTMP + RTSP pull + SRT + WHIP under one engine
  codec_inspect/          per-codec random-access + HLS codec strings
  rtmp_to_hls/            real RTMP ingest → MPEG-TS HLS on disk

Design invariants

• Zero-copy — one publish clones an Arc<MediaFrame> pointer per subscriber, never the payload (bytes::Bytes).
• Lock-free hot path — tokio::broadcast fan-out; ArcSwap + atomics for cached config, GOP head, and QoS.
• No global state — telemetry, auth, and audit are injected traits with safe defaults; instantiate as many engines per process as you like.
• Contracts are traits — protocols depend on PublishRegistry, not the concrete Engine, so a host can swap the bus.
• #![forbid(unsafe_code)] + #![deny(missing_docs)] — no unsafe, and every public item documented; both compiler-enforced.

🚀 Quick Start

A real RTMP-in → HLS-out origin

The rtmp handler ingests from OBS or FFmpeg; the mpegts muxer writes playable HLS. Run the bundled, fully-wired version:

cargo run -p rtmp-to-hls
# ffmpeg -re -i input.mp4 -c:v libx264 -c:a aac -f flv rtmp://localhost:1935/live/cam
# → playable HLS under ./hls/live/cam/index.m3u8

The origin itself is a few lines — register the handler on the builder and serve:

use arcly_stream::prelude::*;
use arcly_stream::protocol::rtmp::RtmpHandler;

#[tokio::main]
async fn main() -> arcly_stream::Result<()> {
    let engine = Engine::builder()
        .application(AppSpec::new("live").gop_cache(120)) // instant-start
        .protocol(RtmpHandler::new("0.0.0.0:1935".parse().unwrap()))
        .build();

    // Publish to rtmp://host/live/<stream>; point an HlsSegmenter + MpegTsMuxer
    // at the same stream for HLS egress.
    engine.serve_registered_until_signal().await
}

Plug in a custom third-party protocol

Teach the engine a brand-new wire protocol from your own crate — implement the public InboundProtocol trait and register it on the builder alongside the native RTMP handler. Both run concurrently under one coordinated shutdown.

use arcly_stream::prelude::*;
use arcly_stream::inbound::{InboundProtocol, IngestContext};
use arcly_stream::protocol::rtmp::RtmpHandler;

struct MyProtocol { /* listener config, handshake state … */ }

#[async_trait]
impl InboundProtocol for MyProtocol {
    fn name(&self) -> &'static str { "my-proto" }

    async fn serve(&self, ctx: IngestContext, shutdown: CancellationToken)
        -> arcly_stream::Result<()>
    {
        // 1. bind your listener · 2. accept + handshake · 3. resolve a StreamKey
        let session = ctx.open_publish(StreamKey::new("live", "cam")).await?;
        // 4. bridge decoded access units → MediaFrame and publish:
        //    session.publish_frame(frame)?;   // → fan-out · GOP cache · QoS
        shutdown.cancelled().await;            // 5. wind down on shutdown
        session.finish().await                 //    (Drop releases it otherwise)
    }
}

let engine = Engine::builder()
    .application(AppSpec::new("live").gop_cache(120))
    .protocol(RtmpHandler::new("0.0.0.0:1935".parse().unwrap())) // native
    .protocol(MyProtocol { /* … */ })                            // yours
    .build();

A full runnable version lives in examples/custom_protocol_plugin.

Runtime contract: workers are Send + Sync + 'static, must observe shutdown and return promptly, and run on a Tokio runtime. Returning Err trips the engine's coordinated teardown, winding sibling workers down too. Existing ProtocolHandler impls keep working — a blanket bridge makes every one an InboundProtocol automatically.

🧩 Extension points (the traits you implement)

⚙️ Cargo features

🧰 Examples

cargo run -p minimal-ingest          # synthetic publisher + resilient subscriber
cargo run -p custom-protocol         # ProtocolHandler driven by Engine::serve
cargo run -p custom-protocol-plugin  # custom InboundProtocol registered with RTMP
cargo run -p enterprise-gateway      # RTMP + token auth + FS recording + metrics + audit
cargo run -p multi-protocol-gateway  # RTMP + RTSP pull + SRT + WHIP under one engine
cargo run -p codec-inspect           # codec::dispatch over H.264/265/AV1/VP9/VVC
cargo run -p rtmp-to-hls             # real RTMP origin → playable HLS
cargo bench  -p arcly-stream         # broadcast fan-out throughput (criterion)

🛡 Guarantees

• #![forbid(unsafe_code)] — no unsafe in the kernel, compiler-enforced.
• #![deny(missing_docs)] — every public item is documented.
• Zero-copy payloads (bytes::Bytes), lock-free fan-out.
• No singletons — many engines per process; ideal for tests.

📐 Minimum Supported Rust Version

arcly-stream supports Rust 1.85 and later. An MSRV bump is a minor-version change.

📚 Documentation

Full API docs are on docs.rs/arcly-stream. See BLUEPRINT.md for the extraction analysis and the mapping from the original 25-crate stream-center to this single crate.

📄 License

MIT.