f7c6346794
A Request ID of the server's parity, or one used twice, closes the session with INVALID_REQUEST_ID; with a limit of two requests a third waits for WT_MAX_STREAMS credit, which returns when a request is cancelled. Co-Authored-By: Claude Opus 5.5 <[email protected]>
62 lines
3 KiB
Rust
62 lines
3 KiB
Rust
//! Request IDs and requests in flight (`draft-ietf-moq-transport-22` §6.4.2): a Request ID
|
|
//! of the server's parity, or one seen before, closes the session with INVALID_REQUEST_ID;
|
|
//! and the requests a client may have open at once are bounded by the relay's stream limit
|
|
//! (here WT_MAX_STREAMS on the WebSocket binding), credit returning as requests end.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq::draft22::{ControlMessage, Parameters, SessionError, StreamResetCode, TrackRequest};
|
|
use joule_moq_relay::client::ClientOptions;
|
|
use joule_moq_relay::transport::TransportError;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn request_ids_and_requests_in_flight() {
|
|
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
fn track_status(id: u64) -> ControlMessage {
|
|
ControlMessage::TrackStatus(TrackRequest { request_id: id, namespace: ns(&["x"]), name: name("y"), params: Parameters::new() })
|
|
}
|
|
|
|
async fn closed_with(c: &joule_moq_relay::client::Client) -> u64 {
|
|
match timeout(STEP, c.conn.closed()).await.expect("closed in time") {
|
|
TransportError::Closed { code, .. } => code,
|
|
other => panic!("{other:?}"),
|
|
}
|
|
}
|
|
|
|
async fn run() {
|
|
let server = relay(|c| c.limits.max_requests = 2).await;
|
|
|
|
// A Request ID with the server's parity (odd).
|
|
let (a, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _r = a.request(&track_status(1)).await.unwrap();
|
|
assert_eq!(closed_with(&a).await, SessionError::INVALID_REQUEST_ID.0);
|
|
|
|
// A Request ID used twice.
|
|
let (b, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let (_t1, mut r1) = b.request(&track_status(0)).await.unwrap();
|
|
let _ = joule_moq_relay::client::Client::response(&mut r1).await;
|
|
let _r2 = b.request(&track_status(0)).await.unwrap();
|
|
assert_eq!(closed_with(&b).await, SessionError::INVALID_REQUEST_ID.0);
|
|
|
|
// Two requests open at once; a third waits for credit until one of them ends.
|
|
let t = Tracks::new(ns(&["live", "r"]), "cam", 10, 6, 128);
|
|
let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
let (c, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let (_ok1, tx1, _rx1) = c.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap();
|
|
let (_ok2, _tx2, _rx2) = c.subscribe(t.ns.clone(), t.receipts.clone(), Parameters::new()).await.unwrap();
|
|
assert!(timeout(Duration::from_millis(400), c.conn.open_bi()).await.is_err(), "a third request opened past the stream limit");
|
|
// Cancelling the first returns its credit (§6.4.2.3).
|
|
tx1.reset(StreamResetCode::CANCELLED.0).await;
|
|
let third = timeout(STEP, c.conn.open_bi()).await;
|
|
assert!(matches!(third, Ok(Ok(_))), "the third request opened once the first ended");
|
|
|
|
c.close(0, "done");
|
|
p.close(0, "done");
|
|
server.shutdown().await;
|
|
}
|