Stream Management

StreamManager is the stream management core of Monibuca V6, responsible for the full lifecycle management of stream creation, publishing, subscription, and destruction.

Stream Lifecycle

flowchart LR
  C["Create"] --> P["Publish"]
  P --> S["Subscribe"]
  S --> D["Destroy"]
  P -.-> InitTracks["Init Tracks<br/>Set Codec<br/>Write Frames"]
  S -.-> ReadFrames["Read Frames<br/>Wait / Unsubscribe"]
  D -.-> Dispose["Dispose<br/>Clean Up<br/>Remove from Registry"]

Detailed State Transitions

stateDiagram-v2
  [*] --> Init
  Init --> TrackAdded: init_video_track / init_audio_track
  TrackAdded --> Subscribed: add_subscriber
  Subscribed --> WaitingSubscriber: all subscribers leave
  WaitingSubscriber --> Subscribed: new subscriber joins
  WaitingSubscriber --> WaitingReconnect: disconnect with reconnect-on-disconnect
  Subscribed --> WaitingReconnect: disconnect with reconnect-on-disconnect
  WaitingReconnect --> Subscribed: republish
  WaitingReconnect --> Disposed: timeout
  Disposed --> [*]

  note right of Subscribed
    Special state Paused: data timeout detection disabled
    (VOD / playback scenarios)
  end note

StreamManager Responsibilities

StreamManager is the unified entry point for stream management, composed of three sub-components:

flowchart TB
  SM["StreamManager"]
  SM --> Reg["StreamRegistry<br/>DashMap high-concurrency storage"]
  SM --> Life["StreamLifecycle<br/>create / destroy / reconnect-on-disconnect"]
  SM --> WQ["WaitingQueue<br/>subscribers waiting for unpublished streams"]

Core API

impl StreamManager {
    // Create a stream (called by publisher)
    fn create_publisher(stream_path) -> Publisher

    // Get a stream (called by subscriber)
    fn get_publisher(stream_path) -> Option<Publisher>

    // Create with Dispatcher
    fn create_publisher_with_dispatcher(stream_path)
        -> (Publisher, Dispatcher)

    // Event subscription
    fn subscribe_events() -> Receiver<StreamEvent>

    // Stream info queries
    fn stream_exists(stream_path) -> bool
    fn stream_count() -> usize
    fn list_streams() -> Vec<StreamInfo>
}

Publisher

The Publisher manages the ingest side of a stream:

pub struct Publisher {
    stream_path: String,       // Stream path (e.g., "live/camera1")
    plugin_name: String,       // Name of the plugin that created this stream
    stream_type: String,       // Stream type (e.g., "rtmp", "rtsp")
    video_track: Option<Arc<VideoTrack>>,  // Video track
    audio_track: Option<Arc<AudioTrack>>,  // Audio track
    subscribers: DashMap<u64, SubscriberInfo>,  // Subscriber list
    frame_notify: watch::Sender<u64>,  // Frame notification channel
    config: PublisherConfig,   // Configuration
    task: Option<Arc<Task>>,   // Associated task (hierarchical cancellation)
}

Publisher Configuration

PublisherConfig {
    video_buffer_capacity: 1024,       // Video buffer size
    audio_buffer_capacity: 64,         // Audio buffer size
    default_buffer_time: 2s,           // Default buffer time
    max_buffer_time: 10s,              // Maximum buffer time
    wait_timeout: 30s,                 // No-subscriber wait timeout
    max_fps: 0,                        // Max frame rate (0 = unlimited)
    data_timeout: 60s,                 // Data timeout (0 = no timeout)
    continue_push_timeout: 30s,        // Reconnect-on-disconnect timeout
}

Publishing Flow

// 1. Create Publisher
let publisher = stream_manager.create_publisher("live/stream1")?;

// 2. Initialize tracks
let mut pub_guard = publisher.write();
pub_guard.init_video_track();
pub_guard.init_audio_track();

// 3. Set codec information
pub_guard.set_video_codec(VideoCodec::new(H264, 1920, 1080));
pub_guard.set_audio_codec(AudioCodec::new(AAC, 44100, 2));

// 4. Write frame data (in the publishing loop)
pub_guard.write_video(IDR, timestamp, dts, data)?;
pub_guard.write_audio(timestamp, data)?;

Subscriber

Subscribers consume frame data from the Publisher’s RingBuffer:

Three Subscription Modes

Mode	Description
RealTime	Start playback from the latest IDR frame (default)
Buffer	Start from an IDR frame before the specified buffer time
WaitKeyframe	Wait for the next keyframe

Subscription Configuration

SubscriberConfig {
    mode: SubscribeMode::RealTime,
    buffer_time: Duration::from_secs(2),
    receive_video: true,
    receive_audio: true,
    keyframe_timeout: Some(Duration::from_secs(30)),
}

Subscription Flow

// 1. Create Subscriber
let mut subscriber = Subscriber::new("live/stream1");

// 2. Subscribe to Publisher
let publisher = stream_manager.get_publisher("live/stream1")?;
subscriber.subscribe(&publisher)?;

// 3. Read frame data (in the playback loop)
loop {
    subscriber.wait_for_frames().await?;

    while let Ok(Some(frame)) = subscriber.read_video() {
        // Process video frame
    }

    while let Some(frame) = subscriber.read_audio() {
        // Process audio frame
    }
}

// 4. Unsubscribe
subscriber.unsubscribe(&publisher);

Automatic Opus→AAC Transcoding

When the source stream audio is Opus-encoded (e.g., WebRTC ingest) and the subscriber needs AAC (e.g., RTMP/FLV playback), the Subscriber automatically performs transcoding:

flowchart LR
  WebRTC["WebRTC ingest Opus"] --> Pub["Publisher"] --> Sub["Subscriber"]
  Sub -->|default| AAC["AAC auto-transcode"] --> RTMP["RTMP playback"]
  Sub -->|skip_opus_transcode=true| Direct["WebRTC subscriber consumes Opus directly"]

Audio/Video Track Management

VideoTrack

flowchart TB
  VT["VideoTrack"]
  VT --> VRB["RingBuffer 1024 slots"]
  VT --> VC["VideoCodec H.264/H.265/AV1"]
  VT --> VSeq["SequenceGenerator"]
  VT --> VStats["Stats BPS/FPS"]
  VT --> VIDR["IDR Tracking"]

VideoTrack supports two frame write formats:

• AVCC format: From RTMP/FLV (length-prefixed NAL units)
• Raw NAL format: From RTSP/RTP (raw NAL unit list)

AudioTrack

flowchart TB
  AT["AudioTrack"]
  AT --> ARB["RingBuffer 64 slots"]
  AT --> AC["AudioCodec AAC/Opus/G.711/…"]
  AT --> ASeq["SequenceGenerator"]
  AT --> AStats["Stats BPS"]

Sequence Header

The Publisher maintains video and audio sequence headers:

• Video sequence header: AVC/HEVC Decoder Configuration Record
• Audio sequence header: AAC AudioSpecificConfig

When a new subscriber connects, the Dispatcher sends the sequence headers first to ensure proper decoder initialization.

Pull/Push Proxy

Pull Proxy

Pulls a stream from a remote source and publishes it locally:

flowchart LR
  Remote["Remote RTMP server"] -->|pull| Pub["Monibuca Publisher"] --> Local["Local subscribers"]

Supported pull protocols: RTMP, RTSP, SRT, HLS, HTTP-FLV, MP4

Push Proxy

Pushes a local stream to a remote server:

flowchart LR
  Local["Local Publisher"] -->|push| Remote["Remote RTMP / RTSP / SRT server"]

Proxy configuration supports automatic retry, reconnection intervals, and maximum retry count.

Recording

The recording module writes stream data to files:

flowchart TB
  Pub["Publisher"] --> RecSub["Subscriber recording"] --> FW["FileWriter"]
  FW --> FLV["FLV"] & MP4["MP4"] & HLS["HLS"]
  MP4 --> Frag["fMP4 / Raw"]

Supported recording formats:

Format	Description
FLV	FLV file recording
MP4	Standard MP4 file
fMP4	Fragmented MP4
HLS	HLS TS segments
Raw	Raw frame data

Recording supports three priority levels: Normal (can be auto-deleted), High (protected), and Event (event-triggered).

Transformer

The Transformer subscribes to a source stream, processes it, and publishes it as a new stream:

flowchart TB
  Src["Source stream live/camera1"] --> X["Transformer<br/>subscribe → process → publish"] --> Dst["Target stream live/camera1_720p"]

Use cases:

• Video transcoding (resolution/codec conversion)
• Watermark/overlay addition
• SEI data injection
• Format conversion

Wait Queue (WaitQueue)

When a subscriber requests a stream that has not yet been published, instead of failing immediately, it enters the wait queue:

flowchart TB
  Req["Subscriber requests live/camera1"]
  Req --> Exists{Stream exists?}
  Exists -->|Yes| Now["Subscribe immediately"]
  Exists -->|No| WQ["Join WaitQueue"]
  WQ --> Wait["Waiting for stream to be published…"]
  Wait --> Pub{Stream published?}
  Pub -->|Yes| Auto["Auto-subscribe"]
  Pub -->|Timeout| Err["Return error"]

WaitQueue decouples stream publishing from subscription — subscribers can start before the publisher. This is particularly useful in the following scenarios:

• On-demand pulling: Subscription triggers a Pull Proxy to pull from a remote source
• Device reconnection: Subscribers wait for a device to reconnect after disconnection
• Live scheduling: Viewers enter a live room in advance and wait for the broadcast to start

Reconnect on Disconnect

When a publisher disconnects, Monibuca does not immediately destroy the stream but enters the WaitingReconnect state:

flowchart TB
  Disc["Publisher disconnects"] --> WR["Publisher → WaitingReconnect"]
  WR --> Choice{Reconnects within continue_push_timeout?}
  Choice -->|Yes| TakeOver["take_over<br/>inherit timestamp · transfer seq headers<br/>restore subscribers · seamless"]
  Choice -->|No| Disp["Publisher → Disposed<br/>Notify all subscribers with EOS"]

Key design: Reconnect on disconnect ensures that viewers don’t need to reconnect during brief publisher interruptions, providing a seamless viewing experience.

The default reconnect-on-disconnect timeout is 30 seconds, configurable via continue_push_timeout. Set to 0 to disable this feature.

Data Timeout Detection

The Publisher detects whether the ingest side has not sent data for an extended period:

// Configure data timeout (default 60 seconds)
data_timeout: Duration::from_secs(60)

• last_data_time is updated on each frame write
• The Publisher is automatically disposed after timeout
• Timeout detection is disabled in the Paused state (for VOD/playback scenarios)
• Set to 0 to disable timeout detection

Event System

StreamManager publishes stream events through a broadcast channel:

let mut events = stream_manager.subscribe_events();

loop {
    match events.recv().await {
        Ok(StreamEvent::Created { path, plugin }) => {
            println!("Stream created: {} (plugin: {})", path, plugin);
        }
        Ok(StreamEvent::Disposed { path, reason }) => {
            println!("Stream disposed: {} ({})", path, reason);
        }
        Ok(StreamEvent::SubscriberAdded { stream, subscriber_id, protocol }) => {
            println!("New subscriber: {} #{} ({})", stream, subscriber_id, protocol);
        }
        Ok(StreamEvent::SubscriberRemoved { stream, subscriber_id }) => {
            println!("Subscriber left: {} #{}", stream, subscriber_id);
        }
        _ => {}
    }
}

Plugins can listen to stream events to trigger corresponding business logic (e.g., on-demand recording, automatic transcoding, etc.).