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

//! Two publishers and three subscribers through the relay over the WebSocket fallback
//! (`draft-ietf-webtrans-http2-15` capsules, one to each message).

mod common;

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn two_publishers_three_subscribers_over_websocket() {
    tokio::time::timeout(std::time::Duration::from_secs(120), common::two_publishers_three_subscribers(common::Via::WebSocket))
        .await
        .expect("the scenario finishes in time");
}

/// A session with nothing to carry pings its peer, so a front that closes idle connections
/// keeps it; the peer's pongs come back and the session stays open.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_idle_session_pings_its_peer() {
    use futures::StreamExt;
    use joule_moq_relay::transport::mux::MuxConfig;
    use joule_moq_relay::transport::websocket::session_pinging;
    use std::time::Duration;
    use tokio_tungstenite::tungstenite::protocol::Role;
    use tokio_tungstenite::tungstenite::Message;
    use tokio_tungstenite::WebSocketStream;

    let (relay_end, peer_end) = tokio::io::duplex(1 << 16);
    let ws = WebSocketStream::from_raw_socket(relay_end, Role::Server, None).await;
    let mut peer = WebSocketStream::from_raw_socket(peer_end, Role::Client, None).await;
    let _session = session_pinging(ws, false, "peer".into(), MuxConfig::default(), Duration::from_millis(50));

    let mut pings = 0;
    let read = tokio::time::timeout(Duration::from_secs(10), async {
        while pings < 3 {
            match peer.next().await {
                Some(Ok(Message::Ping(_))) => pings += 1,
                Some(Ok(_)) => {}
                other => panic!("the session ended before its pings: {other:?}"),
            }
        }
    })
    .await;
    assert!(read.is_ok(), "three pings at a 50 ms interval within 10 s; {pings} arrived");
}