<div align="center">
# 🎬 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.
[](https://crates.io/crates/arcly-stream)
[](https://docs.rs/arcly-stream)
[](#-license)
[](#-minimum-supported-rust-version)
[](#-guarantees)
</div>
---
## 👀 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.
```rust
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);
```
## 🔄 Coming from a standalone server (xiu, MediaMTX, Ant Media)?
Those are **daemons**: you run a binary, point a TOML/YAML file at it, and extend
it by forking. `arcly-stream` is the opposite shape — a **library** you embed in
*your* Rust service, so your application owns the process, the runtime, and the
control plane. The mapping is direct:
| Edit `config.toml`, restart | Call `EngineBuilder` methods at startup — config is your code |
| Built-in HTTP API | Wire the engine into *your* Axum/Hyper handlers; the kernel imposes no framework |
| Fork to add auth / storage / metrics | Implement `StreamAuthenticator` / `StorageBackend` / `Observer` traits |
| Add a protocol = patch the server | Implement `InboundProtocol`, register with `.protocol(...)` |
| One global process | Instantiate as many `Engine`s as you like — it's `Send + Sync`, no singletons |
If you want a turnkey server, run a daemon. If you're **building a product** that
happens to stream — and want the lock-free core without inheriting someone's
framework — embed this.
## 📦 Capability Matrix
| Zero-copy broadcast fan-out | ✅ **Core** | `StreamHandle::subscribe_resilient` |
| Instant-start GOP replay | ✅ **Core** | `AppSpec::gop_cache(n)` + `replay_buffer()` |
| Publish/play authorization | ✅ **Core** | `StreamAuthenticator` (permit-all default) |
| Live QoS (bitrate / fps) | ✅ **Core** | `StreamHandle::qos()` |
| Slow-subscriber eviction | ✅ **Core** | `Subscription::max_lag` |
| Idle-stream reaper | ✅ **Core** | `EngineBuilder::idle_timeout` |
| Coordinated graceful shutdown | ✅ **Core** | `Engine::serve_until_signal` |
| **Pluggable protocol ingestion** | ✅ **Core** | **`InboundProtocol` trait** |
| **RTMP publish / play** | ✅ `rtmp` | `protocol::rtmp::RtmpHandler` |
| **RTSP ingest (IP cameras)** | ✅ `rtsp` | `protocol::rtsp::RtspHandler` — client-pull, `OPTIONS`→`PLAY`, SDP, TCP-interleaved RTP (H.264 + H.265) |
| **SRT ingest (broadcast feeds)** | ✅ `srt` | `protocol::srt::SrtHandler` — listener handshake + MPEG-TS demux (unencrypted) |
| **WebRTC WHIP ingest / WHEP egress** | ✅ `webrtc` | `protocol::webrtc::{WhipEndpoint, WhepEndpoint}` — SDP signaling + RTP/RTCP routing; egress packetizes H.264/H.265, over a pluggable `DtlsSrtpTransport` |
| **RTP payload formats** | ✅ `rtsp`/`webrtc` | `protocol::rtp` — H.264 (RFC 6184), H.265 (RFC 7798), VP9 (draft-ietf-payload-vp9), AV1 (AOMedia) packetizers + depacketizers |
| **MPEG-TS muxing (playable)** | ✅ `mpegts` | `packager::MpegTsMuxer` — H.264 + H.265 (`+codec-h265`) video, AAC audio |
| **Fragmented-MP4 / CMAF muxing** | ✅ `fmp4` | `packager::Fmp4Muxer` — H.264 `avc1` + H.265 `hvc1` (`+codec-h265`) + AV1 `av01` (`+codec-av1`) + VVC `vvc1` (`+codec-vvc`) init, codec-generic fragments (LL-HLS) |
| **Production token auth** | ✅ `auth-token` | `auth::TokenAuthenticator` — HMAC-SHA-256, dependency-free |
| **Audit trail / structured telemetry** | ✅ **Core** | `FileAuditSink`, `StandardTelemetry` |
| H.264/265 · AV1 · VP9 · VVC · AAC parsing | 🧩 `codec*` | `codec::{h264,h265,av1,vp9,vvc,aac}` |
| HLS / LL-HLS packaging | 🧩 `hls` | `HlsSegmenter` + `HlsPlaylist` |
| Recording (VOD / DVR) | 🧩 `record` | `RecordingSink` |
| Filesystem storage | 🧩 `storage-fs` | `FsStorage` |
| Prometheus metrics | 🧩 `metrics` | `PrometheusObserver` |
| Hardware transcode / ABR | 🧭 **Seam** | implement `Transcoder` |
| Edge federation | 🧭 **Seam** | implement `ClusterRelay` |
> ✅ **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**:
```text
┌─────────────────────────────────────────────────┐
│ 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
```text
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:
```bash
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:
```rust,ignore
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.
```rust,ignore
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`](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)
| `InboundProtocol` | **Teach the engine a new wire protocol** (RTMP, RTSP, SRT, WHIP all shipped — and your own is one impl away) |
| `DtlsSrtpTransport` | Plug a WebRTC crypto backend (DTLS/SRTP/ICE) behind the native WHIP/WHEP signaling |
| `MediaSource` / `MediaSink` | Produce / consume frames (driven by `Engine::pump_source`) |
| `StorageBackend` | Object storage for segments & recordings |
| `Muxer` | Container byte-format for HLS segments (MPEG-TS shipped; fMP4/CMAF DIY) |
| `Transcoder` | Decode/scale/encode into an ABR ladder |
| `StreamAuthenticator` | Authorize publish / play (permit-all by default) |
| `Observer` / `AuditSink` / `HealthCheck` | Telemetry, audit trail, readiness |
| `ClusterRelay` | Origin/edge discovery & stream federation |
| `PublishRegistry` / `PlaybackRegistry` / `EventBus` | Swap the engine for your own bus |
## ⚙️ Cargo features
| *(none)* | Pure kernel: frame model, bus, auth/audit/health/cluster contracts |
| `codec` | H.264 NAL/SPS + AAC ADTS parsers |
| `codec-h265` / `-av1` / `-vp9` / `-vvc` / `codecs-all` | Next-gen codec parsers |
| `ingest` | Shared TCP accept loop, keyframe gate, rate limiter, NAL scan |
| `rtmp` | Working RTMP publish/play (`RtmpHandler`) |
| `rtsp` | Native RTSP client-pull ingest (`RtspHandler`) — IP cameras over TCP-interleaved RTP |
| `srt` | Native SRT listener ingest (`SrtHandler`) — handshake + MPEG-TS demux (unencrypted) |
| `webrtc` | WHIP ingest + WHEP egress signaling, RTP/RTCP routing & packetization (`WhipEndpoint`/`WhepEndpoint`) over a pluggable `DtlsSrtpTransport` |
| `mpegts` | Native MPEG-TS muxer (`MpegTsMuxer`) — playable `.ts` (H.264/H.265 + AAC) |
| `fmp4` | Native fragmented-MP4/CMAF muxer (`Fmp4Muxer`) — LL-HLS init + fragments (H.264/H.265/AV1/VVC) |
| `hls` | HLS/LL-HLS packager + segmenter + playlists |
| `record` | Segment/recording sink over a `StorageBackend` |
| `storage-fs` | Filesystem `StorageBackend` |
| `metrics` | Prometheus `Observer` |
| `auth-token` | Production HMAC-signed token `StreamAuthenticator` (dependency-free) |
| `macros` | `#[derive(MediaSink)]`, `#[protocol]` |
| `full` | Everything |
## 🧰 Examples
```bash
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](https://docs.rs/arcly-stream). See
[`BLUEPRINT.md`](BLUEPRINT.md) for the extraction analysis and the mapping from
the original 25-crate `stream-center` to this single crate.
## 📄 License
MIT.