open-standards/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/whep_play.rs

//! A WHEP player (str0m as the client) watches a publisher's tracks through the gateway:
//! it gets the same frames the publisher sent, from the start of the group it joined in,
//! each verified against its receipt at the gateway before it was forwarded. Before the
//! publisher's catalog exists the player is told to come back (409, Retry-After).

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq_ingest::frame::Codec;
use std::time::Duration;
use str0m::media::{Direction, MediaKind};
use tokio::time::timeout;

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_whep_player_gets_the_frames_the_publisher_sent() {
    timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
}

async fn run() {
    let w = World::start(|_| {}).await;
    let viewer = SigningKey::from_bytes(&[0x61; 32]);
    let va = w.auth(&viewer, 1);

    // No publisher yet: 409, with Retry-After (WHEP §4.3.9).
    let o = offer(&[Codec::Avc, Codec::Opus], Direction::RecvOnly, true, true).await;
    let r = va.send("POST", &w.whep_url(), &[("content-type", "application/sdp")], o.offer.as_bytes()).await;
    assert_eq!(r.status, 409, "{}", r.text());
    assert!(r.header("retry-after").is_some());

    let auth = w.auth(&w.publisher, 0);
    let p = publish(&w, &auth).await;
    let mut sent = Sent::default();
    send(&p, &mut sent, 1, 15).await;

    // The player, once the publisher's catalog is there.
    let deadline = std::time::Instant::now() + STEP;
    let (o, resp) = loop {
        let o = offer(&[Codec::Avc, Codec::Opus], Direction::RecvOnly, true, true).await;
        let r = va.send("POST", &w.whep_url(), &[("content-type", "application/sdp")], o.offer.as_bytes()).await;
        if r.status == 201 {
            break (o, r);
        }
        assert_eq!(r.status, 409, "{}", r.text());
        assert!(std::time::Instant::now() < deadline, "the catalog arrives");
        tokio::time::sleep(Duration::from_millis(200)).await;
    };
    let player = o.start(&resp);
    player.connected().await;

    send(&p, &mut sent, 2, 15).await;
    // A last keyframe closes the group before it, so every frame before it is verified
    // and forwarded while the publisher is still live.
    // And, after a pause longer than an audio group, a last audio packet closes the audio
    // group before it the same way.
    let last_key = sent.video.len();
    let last_frame = sent.video.last().cloned();
    tokio::time::sleep(Duration::from_millis(300)).await;
    p.write(MediaKind::Video, Codec::Avc, avc(last_key as u64, true), 1_000 + last_key as u64 * 3000 + 27_000);
    let a = sent.audio.len() as u64;
    p.write(MediaKind::Audio, Codec::Opus, opus(a), 5_000 + a * 960 + 14_400);
    assert!(
        until(STEP, || {
            let got = player.media(MediaKind::Video);
            !got.is_empty() && got.last() == last_frame.as_ref()
        })
        .await,
        "the player gets every frame up to the last closed group"
    );
    let gv = player.media(MediaKind::Video);
    let j = run_of(&sent.video, &gv);
    assert_eq!(j % 15, 0, "the player joins at a group's first frame, a keyframe");
    assert_eq!(j + gv.len(), last_key, "every frame of every group from the join");
    let last_audio = sent.audio.last().cloned();
    assert!(until(STEP, || player.media(MediaKind::Audio).last() == last_audio.as_ref()).await, "the player gets every audio packet up to the last closed group");
    let ga = player.media(MediaKind::Audio);
    let ja = run_of(&sent.audio, &ga);
    assert_eq!(ja + ga.len(), sent.audio.len(), "every audio packet of every group from the join");

    // The publisher leaves: the WHEP session ends with the tracks.
    assert_eq!(delete(&w, &p.location, &auth).await.status, 200);
    assert!(until(STEP, || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "whep")).await, "the WHEP session ends with the tracks");
    let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "whep").unwrap();
    rec.verify().unwrap();
    assert!(rec.record.ended.starts_with("track_ended"), "{}", rec.record.ended);
    let refused: u64 = rec.record.tracks.iter().map(|t| t.dropped_refused).sum();
    assert_eq!(refused, 0, "nothing failed its receipt");
    let out: u64 = rec.record.tracks.iter().filter(|t| t.dir == 2).map(|t| t.objects).sum();
    assert!(out as usize >= gv.len() + ga.len());
    assert_eq!(rec.record.sessions, 1);

    p.close();
    player.close();
    w.stop().await;
}