//! A PUBLISH_NAMESPACE withdrawn after its FIN (`draft-ietf-moq-transport-22` §6.4.2.3: "An
//! endpoint that has already sent a FIN on its sending direction and subsequently wishes to
//! cancel sends STOP_SENDING on the receiving direction"). Over native QUIC and the WebSocket
//! binding, a publisher's namespace stands after the publisher's FIN, as a FIN is no
//! withdrawal (§4.1); its STOP_SENDING then withdraws it, and a namespace subscriber is told
//! NAMESPACE_DONE. The relay used to wait for the session's end alone once the FIN came.

mod common;

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

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

async fn run() {
    let server = relay(|_| {}).await;
    for (i, via) in [Via::Quic, Via::WebSocket].into_iter().enumerate() {
        let top = format!("withdrawn{i}");
        let prefix = ns(&[top.as_str(), "x"]);
        let (p, _pe) = connect(&server, via, ClientOptions::default()).await;
        let (s, _se) = connect(&server, via, ClientOptions::default()).await;
        let (ptx, mut prx) = p.publish_namespace(prefix.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE");
        let (_stx, mut srx) = s.subscribe_namespace(ns(&[top.as_str()])).await.expect("SUBSCRIBE_NAMESPACE");
        let seen = timeout(STEP, srx.next()).await.expect("NAMESPACE in time").expect("it reads");
        assert!(matches!(seen, Some(ControlMessage::Namespace(_))), "{via:?}: {seen:?}");
        // The publisher's FIN: no withdrawal.
        ptx.fin().await;
        let quiet = timeout(Duration::from_millis(500), srx.next()).await;
        assert!(quiet.is_err(), "{via:?}: a FIN withdrew the namespace: {quiet:?}");
        // Its STOP_SENDING on the direction it receives: a withdrawal.
        prx.reader.stop(0x1);
        let done = timeout(STEP, srx.next()).await.unwrap_or_else(|_| panic!("{via:?}: no NAMESPACE_DONE after the publisher's STOP_SENDING")).expect("it reads");
        assert!(matches!(done, Some(ControlMessage::NamespaceDone(_))), "{via:?}: {done:?}");
        p.close(0, "done");
        s.close(0, "done");
    }
    server.shutdown().await;
}