arcly_stream/

lib.rs

//! **arcly-stream** — an open-extensible, high-performance live-media streaming
//! kernel.
//!
//! The pub/sub core of a streaming server: lock-free, zero-copy frame fan-out to
//! an arbitrary number of late-joining subscribers, with instant-start GOP
//! replay, publish/play authorization, live QoS, a pluggable **multi-protocol
//! ingestion layer** ([`InboundProtocol`]), and a feature-gated pure-Rust media
//! plane — all free of any config schema, metrics singleton, or HTTP runtime.
//! You bring the wire protocols and telemetry by implementing small traits; the
//! kernel owns the hard, hot-path part.
//!
//! ```
//! use arcly_stream::prelude::*;
//!
//! let engine = Engine::builder()
//!     .max_publishers(1000)
//!     .application(AppSpec::new("live").gop_cache(120)) // instant-start
//!     .build();
//! assert_eq!(engine.list_apps().len(), 1);
//! ```
//!
//! # Architecture
//!
//! The crate is layered as a small always-on **kernel**, a feature-gated
//! **media plane**, and **deferred wire protocols** that exist only as traits:
//!
//! ```text
//!   ┌─────────────────────── Core kernel (always on) ───────────────────────┐
//!   │  Engine · StreamHandle (broadcast fan-out, GOP cache, live QoS)        │
//!   │  PublishRegistry / PlaybackRegistry / EventBus · Observer · auth ·     │
//!   │  audit · health · MediaFrame / CodecId · StreamError                   │
//!   └───────────────────────────────────────────────────────────────────────┘
//!         ▲ implements                              ▲ feature-gated, pure Rust
//!         │                                         │
//!   ┌─────┴── Wire protocols & muxers ──┐   ┌───────┴── Media plane ──────────┐
//!   │ rtmp   RTMP publish/play (ready)  │   │ codec  H.264/H.265/AV1/VP9/VVC  │
//!   │ rtsp   RTSP pull ingest (ready)   │   │ hls    Packager + segmenter     │
//!   │ srt    SRT/MPEG-TS ingest (ready) │   │ mpegts MPEG-TS muxer            │
//!   │ webrtc WHIP/WHEP signaling+RTP    │   │ fmp4   CMAF muxer               │
//!   │ Transcoder/ClusterRelay (traits)  │   │ record RecordingSink · ingest   │
//!   └───────────────────────────────────┘   └─────────────────────────────────┘
//! ```
//!
//! The wire plane ships four operational protocols, most **bidirectional** —
//! [`rtmp`](crate::protocol::rtmp) (publish + play), [`rtsp`](crate::protocol::rtsp)
//! (client-pull from IP cameras, plus a server that both serves players and
//! ingests `ANNOUNCE`/`RECORD` publishers over UDP or TCP-interleaved RTP),
//! [`srt`](crate::protocol::srt) (a multi-connection server that ingests
//! `m=publish` MPEG-TS feeds and serves `m=request` callers back), and
//! [`webrtc`](crate::protocol::webrtc) (WHIP/WHEP signaling and RTP/RTCP routing
//! over a host-supplied [`DtlsSrtpTransport`](crate::protocol::webrtc::DtlsSrtpTransport))
//! — alongside the [`MpegTsMuxer`](crate::packager::MpegTsMuxer) and
//! [`Fmp4Muxer`](crate::packager::Fmp4Muxer) egress muxers (the latter feeding
//! HLS/LL-HLS and **LL-DASH chunked-CMAF** packaging). Server-side egress is
//! gated by an injected [`EgressGate`](crate::auth::EgressGate); ABR/cluster
//! backends remain trait contracts a host implements.
//!
//! ## Design invariants
//!
//! - **Zero-copy fan-out** — frames are `bytes::Bytes`; one publish clones an
//!   `Arc<MediaFrame>` pointer to every subscriber, never the payload.
//! - **No locks on the hot path** — the frame bus is a `tokio::broadcast`
//!   channel; cached CONFIG frames and the GOP head use `ArcSwap` / atomics.
//! - **No global state** — telemetry ([`Observer`]), authorization
//!   ([`StreamAuthenticator`]), and audit ([`AuditSink`]) are *injected* traits
//!   with no-op/permit-all defaults, never singletons. The engine is
//!   `Send + Sync` and can be instantiated many times per process.
//! - **Contracts are traits** — protocols depend on [`PublishRegistry`] /
//!   [`PlaybackRegistry`], not on the concrete [`Engine`], so a host can swap in
//!   its own bus.
//! - **Pluggable ingest** — a new wire protocol is one [`InboundProtocol`]
//!   ([`inbound`]) implementation, registered with
//!   [`EngineBuilder::protocol`](crate::EngineBuilder::protocol). The engine runs
//!   any worker that satisfies the trait; the bundled `RtmpHandler` is just the
//!   first reference implementation.
//! - **`#![forbid(unsafe_code)]`** — the kernel contains no `unsafe`, enforced
//!   by the compiler. Documentation completeness (`#![deny(missing_docs)]`) and
//!   intra-doc link integrity are compiler-enforced too.
//!
//! ## Module layout
//!
//! | Module        | Responsibility                                                |
//! |---------------|---------------------------------------------------------------|
//! | [`frame`]     | `MediaFrame`, `CodecId`, `FrameType`, `FrameFlags`            |
//! | [`identity`]  | `AppName`, `StreamId`, `StreamKey` (cheap `Arc<str>` ids)     |
//! | [`error`]     | `StreamError` and the crate [`Result`] alias                 |
//! | [`traits`]    | `MediaSource`, `MediaSink`, `ProtocolHandler`, `StorageBackend`, `HwAccelBackend` |
//! | [`bus`]       | `StreamHandle` fan-out (GOP cache, live QoS) + registry contracts |
//! | [`inbound`]   | `InboundProtocol` trait + `IngestContext` / `PublishSession` — the multi-protocol ingestion seam |
//! | [`engine`]    | `Engine`, `EngineBuilder`, `AppSpec` — composition, run loop, source pump, idle reaper |
//! | [`auth`]      | `StreamAuthenticator` publish/play authorization (permit-all default) |
//! | [`observe`]   | `Observer` telemetry hook + `NoopObserver`                   |
//! | [`audit`]     | `AuditSink` + `AuditPipeline` lifecycle audit trail          |
//! | [`health`]    | `HealthCheck` / `HealthRegistry` readiness aggregation       |
//! | [`transcode`] | `Transcoder` / `RenditionSpec` ABR contracts                |
//! | [`cluster`]   | `ClusterRelay` origin/edge federation contracts             |
//! | [`testing`]   | In-memory helpers for downstream tests                       |
//!
//! ## Cargo features
//!
//! | Feature | Effect |
//! |---|---|
//! | *(none)* | Pure engine: frame model, bus, traits, auth/audit/health/cluster contracts |
//! | `codec` | Baseline H.264 NAL/SPS + AAC ADTS parsers ([`codec`]) |
//! | `codec-h265` / `codec-av1` / `codec-vp9` / `codec-vvc` | Next-gen codec parsers (opt-in, pure Rust) |
//! | `codecs-all` | Every codec parser at once |
//! | `storage-fs` | Filesystem [`StorageBackend`] adapter |
//! | `ingest` | Shared TCP accept loop, keyframe gate, rate limiter, NAL scan |
//! | `hls` | HLS/LL-HLS packager: `Muxer`/`Packager`, segmenter, playlists ([`packager`]) |
//! | `mpegts` | Native MPEG-TS `Muxer` (`MpegTsMuxer`) — playable `.ts` segments |
//! | `fmp4` | Native fragmented-MP4/CMAF `Muxer` ([`Fmp4Muxer`](packager::Fmp4Muxer)) — `ftyp`+`moov` init & `moof`+`mdat` fragments for LL-HLS |
//! | `rtmp` | Working RTMP publish/play [`ProtocolHandler`](protocol::rtmp::RtmpHandler) |
//! | `rtsp` | Native RTSP client-pull ingest ([`RtspHandler`](protocol::rtsp::RtspHandler)) — `OPTIONS`/`DESCRIBE`/`SETUP`/`PLAY`/`TEARDOWN`, SDP, TCP-interleaved RTP |
//! | `srt` | Native SRT listener ingest ([`SrtHandler`](protocol::srt::SrtHandler)) — handshake + MPEG-TS demux (unencrypted) |
//! | `webrtc` | WHIP/WHEP signaling + RTP/RTCP routing ([`WhipEndpoint`](protocol::webrtc::WhipEndpoint)) over a pluggable `DtlsSrtpTransport` crypto seam — incl. RTCP feedback ([`rtcp`](protocol::webrtc::rtcp)), per-viewer adaptive bitrate, viewer-driven keyframe recovery, conference rooms ([`Room`](protocol::webrtc::Room)), active-speaker, and `peer_stats` |
//! | `record` | Segment/recording sink over a [`StorageBackend`] ([`record`]) |
//! | `hls` / `record` | Shared keyframe-boundary [`segment`] timing |
//! | `metrics` | Prometheus [`Observer`] implementation |
//! | `cluster` | In-process reference [`ClusterRelay`] + `ClusterDirectory` (origin/edge federation) |
//! | `auth-token` | Production signed-token [`StreamAuthenticator`] ([`TokenAuthenticator`]) — HMAC-SHA-256, dependency-free |
//! | `macros` | `#[derive(MediaSink)]`, `#[protocol]` ergonomics |
//! | `full` | everything |
//!
//! Ready-to-use reference implementations ship for the operational traits too:
//! [`FileAuditSink`] ([`AuditSink`] → file),
//! [`StandardTelemetry`] (structured-logging [`Observer`]), and the
//! `metrics`-gated [`PrometheusObserver`](observability::PrometheusObserver).

#![cfg_attr(docsrs, feature(doc_cfg))]
// The engine contains no `unsafe` — make that a compiler-enforced guarantee so
// a future change can't quietly introduce it.
#![forbid(unsafe_code)]
// Documentation completeness is enforced, not aspirational: every public item
// carries docs, and intra-doc links must resolve.
#![deny(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]

pub mod audit;
pub mod auth;
pub mod bus;
pub mod cluster;
pub mod engine;
pub mod error;
pub mod frame;
pub mod health;
pub mod identity;
pub mod inbound;
pub mod observe;
pub mod testing;
pub mod traits;
pub mod transcode;

// Canonical Annex-B start-code scanner, shared by the codec NAL utilities and
// the ingest packetizers. Internal; compiled whenever either layer is present.
#[cfg(any(feature = "_nal", feature = "ingest"))]
mod bytescan;

// Dependency-free SHA-256 / HMAC-SHA-256 backing the signed-token authenticator.
#[cfg(feature = "auth-token")]
mod crypto;

#[cfg(feature = "_codec")]
#[cfg_attr(docsrs, doc(cfg(feature = "codec")))]
pub mod codec;

#[cfg(feature = "storage-fs")]
#[cfg_attr(docsrs, doc(cfg(feature = "storage-fs")))]
pub mod storage;

#[cfg(feature = "ingest")]
#[cfg_attr(docsrs, doc(cfg(feature = "ingest")))]
pub mod protocol;

#[cfg(feature = "onvif")]
#[cfg_attr(docsrs, doc(cfg(feature = "onvif")))]
pub mod onvif;

#[cfg(feature = "hls")]
#[cfg_attr(docsrs, doc(cfg(feature = "hls")))]
pub mod packager;

#[cfg(feature = "record")]
#[cfg_attr(docsrs, doc(cfg(feature = "record")))]
pub mod record;

#[cfg(any(feature = "hls", feature = "record"))]
#[cfg_attr(docsrs, doc(cfg(any(feature = "hls", feature = "record"))))]
pub mod segment;

#[cfg(feature = "metrics")]
#[cfg_attr(docsrs, doc(cfg(feature = "metrics")))]
pub mod observability;

// ── Flat re-exports (match sc-core's surface so migration is mechanical) ────
pub use audit::{AuditAction, AuditPipeline, AuditRecord, AuditSink, FileAuditSink};
#[cfg(feature = "auth-token")]
#[cfg_attr(docsrs, doc(cfg(feature = "auth-token")))]
pub use auth::TokenAuthenticator;
pub use auth::{AllowAll, Credentials, StreamAuthenticator};
pub use bus::{
    EventBus, PlaybackRegistry, PublishRegistry, Qos, StreamEvent, StreamEventKind, StreamHandle,
    StreamMetadata, StreamState, Subscription,
};
pub use cluster::{mirror, ClusterRelay, FrameSource, NodeAddr};
pub use engine::{AppSpec, Engine, EngineBuilder, EngineConfig};
pub use error::{ProtocolErrorKind, StreamError};
pub use frame::{AudioFrame, CodecId, FrameFlags, FrameType, MediaFrame, VideoFrame};
pub use health::{HealthCheck, HealthRegistry, HealthReport, HealthStatus};
pub use identity::{AppName, StreamId, StreamKey};
pub use inbound::{InboundProtocol, IngestContext, PublishSession};
pub use observe::{NoopObserver, Observer, StandardTelemetry};
pub use traits::{
    HwAccelBackend, MediaSink, MediaSource, ProtocolHandler, SinkDriver, StorageBackend,
};

// Re-export `bytes` so downstream crates (and doctests) can name `Bytes` — the
// payload type of every `MediaFrame` — without pinning a separate version.
pub use async_trait;
pub use bytes;
pub use transcode::{RenditionSpec, Transcoder};

/// The crate-wide fallible result. Mirrors `sc-core::Result`.
pub type Result<T> = std::result::Result<T, StreamError>;

/// Re-exports the derive macros expand against. Stable surface for codegen so
/// internal files can move without touching the macro crate.
#[cfg(feature = "macros")]
#[doc(hidden)]
pub mod __macro_support {
    pub use crate::bus::PublishRegistry;
    pub use crate::traits::{MediaSink, ProtocolHandler};
    pub use crate::{MediaFrame, Result, StreamError, StreamKey};
    pub use async_trait::async_trait;
    pub use std::sync::Arc;
    pub use tokio_util::sync::CancellationToken;
}

/// `use arcly_stream::prelude::*;` — everything a downstream user needs.
pub mod prelude {
    pub use crate::audit::{AuditPipeline, AuditRecord, AuditSink, FileAuditSink};
    #[cfg(feature = "auth-token")]
    pub use crate::auth::TokenAuthenticator;
    pub use crate::auth::{AllowAll, Credentials, StreamAuthenticator};
    pub use crate::bus::{
        EventBus, PlaybackRegistry, PublishRegistry, Qos, StreamEvent, StreamEventKind,
        StreamHandle, StreamState, Subscription,
    };
    pub use crate::cluster::{ClusterRelay, NodeAddr};
    pub use crate::engine::{AppSpec, Engine, EngineBuilder, EngineConfig};
    pub use crate::error::{ProtocolErrorKind, StreamError};
    pub use crate::frame::{CodecId, FrameFlags, FrameType, MediaFrame};
    pub use crate::health::{HealthCheck, HealthRegistry, HealthStatus};
    pub use crate::identity::{AppName, StreamId, StreamKey};
    pub use crate::inbound::{InboundProtocol, IngestContext, PublishSession};
    pub use crate::observe::{NoopObserver, Observer, StandardTelemetry};
    #[cfg(feature = "rtsp")]
    pub use crate::protocol::rtsp::{RtspHandler, RtspSource};
    #[cfg(feature = "srt")]
    pub use crate::protocol::srt::SrtHandler;
    #[cfg(feature = "webrtc")]
    pub use crate::protocol::webrtc::{
        DtlsSrtpTransport, PeerStats, Room, WhepEndpoint, WhepResource, WhipEndpoint, WhipResource,
    };
    pub use crate::traits::{
        HwAccelBackend, MediaSink, MediaSource, ProtocolHandler, SinkDriver, StorageBackend,
    };
    pub use crate::transcode::{drive_abr_ladder, RenditionSpec, Transcoder, RENDITION_TRACK_BASE};
    pub use crate::Result;
    pub use async_trait::async_trait;
    pub use tokio_util::sync::CancellationToken;

    #[cfg(feature = "macros")]
    pub use arcly_stream_macros::{protocol, MediaSink as MediaSinkDerive};
}