arcly-stream 0.8.3

//! Runnable demo of the in-process [`ClusterRelay`] (feature `cluster`).
//!
//! Two engines in one process play the part of an **origin** and an **edge**
//! node. The origin publishes a stream and announces it; the edge — which has no
//! local copy — `locate`s the origin via the shared directory, `pull`s the
//! stream, and a viewer on the *edge* receives the mirrored frames.
//!
//! ```text
//! cargo run -p arcly-stream --example cluster_relay --features cluster
//! ```

use std::sync::Arc;
use std::time::Duration;

use arcly_stream::bus::PublishRegistry;
use arcly_stream::cluster::{ClusterDirectory, ClusterRelay, InProcessRelay};
use arcly_stream::{AppSpec, CodecId, Engine, MediaFrame, PlaybackRegistry, StreamKey};

#[tokio::main]
async fn main() {
    let directory = Arc::new(ClusterDirectory::new());
    let key = StreamKey::new("live", "cam");

    // ── Origin node: publishes the stream + announces it to the directory. ──
    let origin = Engine::builder()
        .application(AppSpec::new("live").gop_cache(30))
        .build();
    let origin_relay = InProcessRelay::new("origin-1", origin.clone(), directory.clone());
    let src = origin.start_publish(&key).await.unwrap();
    origin_relay.announce(&key).await.unwrap();
    println!("[origin-1] publishing + announced {}", key);

    // Pump frames from the origin at ~25 fps in the background.
    let pump = {
        let src = src.clone();
        tokio::spawn(async move {
            let mut pts = 0i64;
            for n in 0..50 {
                let key = n % 25 == 0; // a keyframe every second
                let f = MediaFrame::new_video(
                    pts,
                    pts,
                    bytes::Bytes::from_static(b"frame"),
                    CodecId::H264,
                    key,
                );
                let _ = src.publish_frame(f);
                pts += 40;
                tokio::time::sleep(Duration::from_millis(40)).await;
            }
        })
    };

    // ── Edge node: no local copy. Locate the origin, then pull it. ──────────
    let edge = Engine::builder()
        .application(AppSpec::new("live").gop_cache(30))
        .build();
    let edge_relay = InProcessRelay::new("edge-7", edge.clone(), directory.clone())
        .with_peer("origin-1", origin.clone() as Arc<dyn PlaybackRegistry>);

    assert!(edge.get_stream(&key).is_err());
    println!("[edge-7]  no local stream; locating origin…");

    let origin_addr = edge_relay
        .locate(&key)
        .await
        .unwrap()
        .expect("directory knows the origin");
    println!("[edge-7]  located at '{}'; pulling…", origin_addr.0);
    edge_relay.pull(&key, &origin_addr).await.unwrap();

    // ── Viewer on the EDGE: subscribes to the mirrored local stream. ────────
    let mirror = edge
        .get_stream(&key)
        .expect("edge mirror is now publishing");
    let mut viewer = mirror.subscribe_resilient();
    let counter = tokio::spawn(async move {
        let mut frames = 0u32;
        let mut keys = 0u32;
        while let Some(f) = viewer.recv().await {
            frames += 1;
            if f.is_keyframe() {
                keys += 1;
            }
        }
        (frames, keys)
    });

    pump.await.unwrap();
    // Let the mirror drain, then end the origin publish (closes the edge mirror).
    tokio::time::sleep(Duration::from_millis(200)).await;
    origin.end_publish(&key).await.unwrap();

    let (frames, keys) = counter.await.unwrap();
    println!("[edge-7]  viewer received {frames} mirrored frames ({keys} keyframes) from origin-1");
    assert!(
        frames > 0,
        "the edge viewer must have received mirrored frames"
    );
    println!("\nOK: ClusterRelay mirrored the origin stream to an edge viewer.");
}