open-standards/joule-code/joule-code-rs/crates/joule-moq-relay/tests/subscribe_ended_at_once.rs

//! An upstream subscription answered SUBSCRIBE_OK and at once PUBLISH_DONE. The relay has its
//! Established upstream subscription before it answers the downstream SUBSCRIBE (`draft-ietf-
//! moq-transport-22` §7.4); when that subscription ends before the downstream one is made of it,
//! the subscriber is told what happened, SUBSCRIBE_OK and then PUBLISH_DONE with the publisher's
//! status and no stream (§9.9), not REQUEST_ERROR DOES_NOT_EXIST "the publisher left", which
//! the interop run saw at -23 and at -16. A publisher of a namespace answers each of the relay's
//! SUBSCRIBEs so, many times over native QUIC and the WebSocket binding, each subscriber held to
//! that answer.

mod common;

use common::*;
use joule_moq::draft22::{ControlMessage, Parameters, PublishDone, PublishDoneCode, SubscribeOk};
use joule_moq_relay::client::ClientOptions;
use std::time::Duration;
use tokio::time::timeout;

const ROUNDS: usize = 24;

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

async fn run() {
    let server = relay(|_| {}).await;
    for via in [Via::Quic, Via::WebSocket] {
        let prefix = ns(&["ended-at-once", if via == Via::Quic { "q" } else { "w" }]);
        let (p, _pe) = connect(&server, via, ClientOptions::default()).await;
        let (s, _se) = connect(&server, via, ClientOptions::default()).await;
        let _ns = p.publish_namespace(prefix.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE");
        let mut refused = Vec::new();
        for i in 0..ROUNDS {
            let name = name(&format!("t{i}"));
            let (pfx, nm) = (prefix.clone(), name.clone());
            let s2 = &s;
            let asked = async move { s2.subscribe(pfx, nm, Parameters::new()).await };
            let answered = async {
                let (tx, _rx, first) = p.next_request(STEP).await.expect("the relay's SUBSCRIBE in time");
                assert!(matches!(first, ControlMessage::Subscribe(_)), "the relay sent {}", first.name());
                tx.send(&ControlMessage::SubscribeOk(SubscribeOk { track_alias: 1000 + i as u64, params: Parameters::new(), track_properties: track_props(64, None) })).await.unwrap();
                tx.send(&ControlMessage::PublishDone(PublishDone { status: PublishDoneCode::TRACK_ENDED, stream_count: 0, reason: Vec::new() })).await.unwrap();
                tx.fin().await;
            };
            let (got, ()) = tokio::join!(asked, answered);
            match got {
                Ok((_ok, _tx, mut rx)) => {
                    let done = timeout(STEP, rx.next()).await.expect("PUBLISH_DONE in time").expect("it reads");
                    match done {
                        Some(ControlMessage::PublishDone(d)) => assert_eq!((d.status, d.stream_count), (PublishDoneCode::TRACK_ENDED, 0), "{via:?} round {i}"),
                        other => panic!("{via:?} round {i}: after SUBSCRIBE_OK, {other:?}"),
                    }
                }
                Err(why) => refused.push(format!("round {i}: {why}")),
            }
        }
        assert!(refused.is_empty(), "{via:?}: {} of {ROUNDS} subscribers refused where the upstream answered: {:?}", refused.len(), &refused[..refused.len().min(3)]);
        p.close(0, "done");
        s.close(0, "done");
    }
    server.shutdown().await;
}