611f3159f1
The bridge refused a -23 SUBSCRIBE_TRACKS with FILL_PARAMETERS NOT_SUPPORTED, as in a -23 PUBLISH. -23 §3.6.2 has a SUBSCRIBE_TRACKS's parameters be the initial ones of each PUBLISH it brings and be included in it, §9.19.15 lists SUBSCRIBE_TRACKS among the messages that carry FILL_PARAMETERS, and §3.4.1 has "A publisher opens a fill fetch stream when it ... sends a PUBLISH, that carries FILL_PARAMETERS while the subscription is not paused". The -23 bridge now carries them beside -22's SUBSCRIBE_TRACKS (draft23::bridge::TRACKS_BESIDE), the relay's PUBLISH carries them (added by the session's bridge, Connection::publish_fill, as SUBSCRIBE_TRACKS_ID is), and a fill fetch stream named by the PUBLISH's Request ID brings the fill range (the Location filter inside them, else the subscription's, else the track to its Largest Object, §3.4) from the cache or upstream as a SUBSCRIBE's fill does. A paused subscription opens none; a fill that seeks below the live edge without the seek power (WAI live-rooms §10.6) has no request to refuse and is its stream reset after its header (§3.4.1: "it MUST open a fill fetch stream and reset it immediately after the FETCH_HEADER if necessary"). A -23 PUBLISH this relay receives with FILL_PARAMETERS is still refused. tests/draft23.rs, a_23_subscribe_tracks_with_fill_parameters, writes the request raw at -23 and was run against three faults planted in turn (no fill opened, FILL_PARAMETERS left out of the PUBLISH, a fill opened while paused): each fails it. Co-Authored-By: Claude Opus 5.5 <[email protected]>
729 lines
43 KiB
Rust
729 lines
43 KiB
Rust
//! Sessions at `draft-ietf-moq-transport-23` through the relay, beside -22 ones: the relay
|
|
//! chooses -23 by ALPN, `WT-Available-Protocols` or the WebSocket subprotocol and bridges it
|
|
//! to its -22 model (`transport::bridge`). WAI objects and their receipts cross -23 and -22
|
|
//! sessions both ways and verify at each subscriber; a FETCH from the cache returns them at
|
|
//! -23, its FETCH_OK a REQUEST_OK; a tampered object is refused at the relay's door. Raw
|
|
//! sessions written byte by byte from -23's own messages, not through this crate's client,
|
|
//! check the relay's -23 wire, and each planted fault closes the session with -23's code
|
|
//! for it.
|
|
//!
|
|
//! A -23 session's subgroup streams are read by -23's object rules, where they differ from
|
|
//! -22's (a supported property outside its scope or repeated makes the track malformed; a
|
|
//! grease value in the Mandatory Track Property range is no Mandatory Track Property), each
|
|
//! held against the same object from a -22 session, which keeps -22's reading. And the relay's
|
|
//! LARGEST_OBJECT is held to -23 §7.5 by a vector of what a publisher states and sends.
|
|
//!
|
|
//! A raw -23 SUBSCRIBE_TRACKS: the subscription parameters it carries that -22's has no place for
|
|
//! (both delivery timeouts, SUBSCRIBER_PRIORITY, LOCATION_FILTER) are stated in each PUBLISH it
|
|
//! brings and govern the subscription; its FILL_PARAMETERS are carried in each PUBLISH, which
|
|
//! opens a fill fetch stream named by the PUBLISH's Request ID, and none for a paused one.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::SigningKey;
|
|
use joule_moq::carriage::{object_properties, ObjectExtras, ReceiptBundle};
|
|
use joule_moq::draft::Draft;
|
|
use joule_moq::draft22::data::{FetchEntry, ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::{Location, LocationFilter, Parameter, Parameters};
|
|
use joule_moq::draft23 as m23;
|
|
use joule_moq::messages::ObjectHeader;
|
|
use joule_moq::receipt::ReceiptVerifier;
|
|
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::transport::{quic, BoxRecv, SharedConnection, TransportError};
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
const LIMIT: Duration = Duration::from_secs(120);
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn every_session_at_23_over_quic() {
|
|
timeout(LIMIT, two_publishers_three_subscribers(Via::QuicDraft(Draft::D23))).await.expect("in time");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn every_session_at_23_over_webtransport() {
|
|
timeout(LIMIT, two_publishers_three_subscribers(Via::WtDraft(Draft::D23))).await.expect("in time");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn every_session_at_23_over_websocket() {
|
|
timeout(LIMIT, two_publishers_three_subscribers(Via::WsDraft(Draft::D23))).await.expect("in time");
|
|
}
|
|
|
|
/// A -23 publisher and a -22 one; subscribers at -22 (the namespace subscriber), -23 over
|
|
/// WebTransport and -23 over the WebSocket binding: every object of both publishers reaches
|
|
/// every subscriber and verifies against its publisher's receipts.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_23_and_a_22_client_through_one_relay() {
|
|
let vias = [Via::QuicDraft(Draft::D23), Via::Quic, Via::Quic, Via::WtDraft(Draft::D23), Via::WsDraft(Draft::D23)];
|
|
timeout(LIMIT, two_publishers_three_subscribers_on(vias)).await.expect("in time");
|
|
}
|
|
|
|
/// The reverse: a -22 publisher by namespace, a -23 one by PUBLISH, and the namespace
|
|
/// subscriber at -23.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_22_publisher_and_23_subscribers() {
|
|
let vias = [Via::Quic, Via::QuicDraft(Draft::D23), Via::QuicDraft(Draft::D23), Via::Quic, Via::WtDraft(Draft::D22)];
|
|
timeout(LIMIT, two_publishers_three_subscribers_on(vias)).await.expect("in time");
|
|
}
|
|
|
|
/// A late joiner at -23 FETCHes a track and its receipts, published at -22, from the
|
|
/// relay's cache: -23's FETCH_OK is a REQUEST_OK whose LARGEST_OBJECT gives the end, and the
|
|
/// fetch stream's Object ID deltas and indicators are -23's. The same objects come back and
|
|
/// verify; a range comes back whole; a relative filter is refused (INVALID_RANGE, §3.2).
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_fetch_from_the_cache_at_23() {
|
|
timeout(LIMIT, async {
|
|
const GROUPS: u64 = 6;
|
|
let server = relay(|_| {}).await;
|
|
let key = SigningKey::from_bytes(&[0x2c; 32]);
|
|
let t = Tracks::new(ns(&["vod", "d23"]), "ep", 10, 5, 100);
|
|
let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let _held = publish_tracks(&publisher, &t, None).await;
|
|
for g in 0..GROUPS {
|
|
publish_group(&publisher, &key, &t, g, &payload("ep", g, 1500)).await;
|
|
}
|
|
let relay = server.relay().clone();
|
|
let (mk, rk) = ((t.ns.clone(), t.media.clone()), (t.ns.clone(), t.receipts.clone()));
|
|
assert!(
|
|
until(STEP, || relay.largest(&mk) == Some(Location::new(GROUPS - 1, 0)) && relay.cache_covers(&mk, Location::new(0, 0)) && relay.cache_covers(&rk, Location::new(0, 0))).await,
|
|
"the cache holds the track"
|
|
);
|
|
let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
for via in [Via::QuicDraft(Draft::D23), Via::WtDraft(Draft::D23)] {
|
|
let (late, _ep) = connect(&server, via, ClientOptions::default()).await;
|
|
let (ok, media) = late.fetch(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap_or_else(|e| panic!("FETCH at -23: {e}"));
|
|
assert_eq!(ok.end_location, Location::new(GROUPS - 1, 0), "the end, from -23's LARGEST_OBJECT");
|
|
let (_, rec) = late.fetch(t.ns.clone(), t.receipts.clone(), Parameters::new()).await.expect("FETCH the receipts");
|
|
let objs = |v: &[FetchEntry]| v.iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some(o.clone()) } else { None }).collect::<Vec<_>>();
|
|
let (media, rec) = (objs(&media), objs(&rec));
|
|
assert_eq!((media.len(), rec.len()), (GROUPS as usize, GROUPS as usize));
|
|
for (m, r) in media.iter().zip(&rec) {
|
|
let b = ReceiptBundle::from_payload(&r.payload).unwrap();
|
|
let vg = verifier.verify_group(&b.group).expect("a receipt verifies");
|
|
verifier.verify_payload(&m.payload, &b.objects[0], &vg).expect("an object verifies");
|
|
assert_eq!(m.payload, payload("ep", m.group_id, 1500));
|
|
}
|
|
let range = Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStartGroupEnd { start: Location::new(2, 0), end_group_delta: 1 }));
|
|
let (ok, part) = late.fetch(t.ns.clone(), t.media.clone(), range).await.expect("FETCH a range");
|
|
assert_eq!(objs(&part).iter().map(|o| o.group_id).collect::<Vec<_>>(), vec![2, 3]);
|
|
assert_eq!(ok.end_location.group, 3);
|
|
let rel = Parameters::new().with(Parameter::LocationFilter(LocationFilter::NextObject));
|
|
let e = late.fetch(t.ns.clone(), t.media.clone(), rel).await.expect_err("a relative filter in FETCH");
|
|
assert!(e.contains("INVALID_RANGE"), "{e}");
|
|
assert!(late.conn.notes().iter().any(|n| n.contains("relative Location filter")), "{:?}", late.conn.notes());
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// Group `g` of `t` signed over `signed`, its object carrying `sent`.
|
|
async fn publish_group_as(c: &Client, key: &SigningKey, t: &Tracks, g: u64, signed: &[u8], sent: &[u8]) {
|
|
let (receipt, bundle) = signed_group(key, t.receipt_alias, g, signed);
|
|
let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
|
|
c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.unwrap();
|
|
let header = ObjectHeader { track_alias: TrackAlias(t.media_alias), group_id: GroupId(g), object_id: ObjectId(0), publisher_priority: t.priority, status: ObjectStatus::Normal, payload_len: sent.len() as u64, mrl_layer: 0, mrl_total_layers: 0, wai_envelope: false };
|
|
let props = object_properties(&header, &ObjectExtras { receipt: Some(receipt), wai_receipt: None, delivery_timeout_ms: None }).unwrap();
|
|
let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
|
|
c.send_subgroup(mh, &[SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: sent.to_vec() }]).await.unwrap();
|
|
}
|
|
|
|
/// A -23 publisher sends one group whose object is not the one its receipt signs: the
|
|
/// relay's door refuses that object alone, and a -23 subscriber gets every other group,
|
|
/// each verifying.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_tampered_object_at_23_is_refused_at_the_door() {
|
|
timeout(LIMIT, async {
|
|
const G: u64 = 5;
|
|
const BAD: u64 = 2;
|
|
let key = SigningKey::from_bytes(&[0xd3; 32]);
|
|
let server = relay(|c| {
|
|
c.receipts.door = vec![joule_moq_relay::config::DoorRule { prefix: vec!["live".into()], trust: vec![hex::encode(key.verifying_key().to_bytes())], trust_field: None, hold_ms: 5_000 }];
|
|
})
|
|
.await;
|
|
let t = Tracks::new(ns(&["live", "t23"]), "cam", 30, 7, 80);
|
|
let (p, _ep) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
|
|
let _held = publish_tracks(&p, &t, None).await;
|
|
let (s, _es) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
|
|
let view = subscribe_both(&s, &t, Parameters::new()).await;
|
|
for g in 0..G {
|
|
let good = payload("t23", g, 700);
|
|
if g == BAD {
|
|
let mut bad = good.clone();
|
|
bad[0] ^= 0xff;
|
|
publish_group_as(&p, &key, &t, g, &good, &bad).await;
|
|
} else {
|
|
publish_group(&p, &key, &t, g, &good).await;
|
|
}
|
|
}
|
|
assert!(until(STEP, || server.relay().stats().refused_at_door >= 1).await, "the door refused the tampered object");
|
|
// Every media group but the tampered one, and the receipts of each: the two tracks
|
|
// ride streams of their own, so a group's object may arrive before its receipts.
|
|
let got = collect(&s, STEP, |got| {
|
|
let media: Vec<u64> = got.iter().filter(|x| x.header.track_alias == view.media_alias).map(|x| x.header.group_id).collect();
|
|
media.len() as u64 >= G - 1 && media.iter().all(|g| got.iter().any(|x| x.header.track_alias == view.receipts_alias && x.header.group_id == *g))
|
|
})
|
|
.await;
|
|
let (objects, bundles, _, _) = verify(&got, &view, &key);
|
|
assert_eq!(objects, (G - 1) as usize, "every object but the tampered one, each verifying");
|
|
assert!(bundles >= (G - 1) as usize);
|
|
assert!(got.iter().filter(|x| x.header.track_alias == view.media_alias).all(|x| x.header.group_id != BAD), "the tampered object is not forwarded");
|
|
assert_eq!(server.relay().stats().refused_at_door, 1);
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
// ── Raw -23 sessions ────────────────────────────────────────────────────────
|
|
|
|
async fn raw_session(server: &joule_moq_relay::RelayServer) -> (SharedConnection, quinn::Endpoint) {
|
|
let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![m23::ALPN.to_vec()], None).unwrap();
|
|
let (conn, got) = timeout(STEP, quic::connect_any(&ep, server.quic_addr.unwrap(), "localhost")).await.expect("in time").expect("a QUIC session");
|
|
assert_eq!(got, Draft::D23);
|
|
(Arc::new(conn), ep)
|
|
}
|
|
|
|
/// Read one framed -23 message (Type, a varint Length, the body).
|
|
async fn read23(r: &mut BoxRecv, buf: &mut Vec<u8>) -> m23::ControlMessage {
|
|
loop {
|
|
let mut rd = joule_moq::draft22::io::Reader::new(buf);
|
|
if let Ok(m) = m23::ControlMessage::decode(&mut rd) {
|
|
let n = rd.position();
|
|
buf.drain(..n);
|
|
return m;
|
|
}
|
|
match timeout(STEP, r.read_chunk()).await.expect("in time").expect("a chunk") {
|
|
Some(b) => buf.extend_from_slice(&b),
|
|
None => panic!("the stream ended"),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A -23 session with its SETUP sent and the relay's read; the control stream is returned
|
|
/// to be held open.
|
|
async fn setup23(server: &joule_moq_relay::RelayServer) -> (SharedConnection, quinn::Endpoint, m23::Setup, joule_moq_relay::transport::BoxSend) {
|
|
let (conn, ep) = raw_session(server).await;
|
|
let mut ctl = conn.open_uni().await.unwrap();
|
|
let mut o = joule_moq::draft22::kvp::SetupOptions::new();
|
|
o.push(joule_moq::draft22::kvp::KeyValuePair::bytes(joule_moq::draft22::kvp::opt::MOQT_IMPLEMENTATION, b"raw-23".to_vec()));
|
|
ctl.write_all(&m23::ControlMessage::Setup(m23::Setup { options: o }).encode().unwrap()).await.unwrap();
|
|
let mut theirs = timeout(STEP, conn.accept_uni()).await.expect("in time").expect("a stream");
|
|
let mut d = m23::UniStreamDecoder::new(joule_moq::draft22::GroupOrder::Ascending, Default::default());
|
|
let setup = loop {
|
|
match d.next().unwrap() {
|
|
Some(m23::UniEvent::Control(m)) => break m,
|
|
Some(_) => continue,
|
|
None => d.push(&timeout(STEP, theirs.read_chunk()).await.unwrap().unwrap().unwrap()),
|
|
}
|
|
};
|
|
let m23::ControlMessage::Setup(s) = setup else { panic!("{setup:?}") };
|
|
(conn, ep, s, ctl)
|
|
}
|
|
|
|
fn closed_with(e: TransportError) -> u64 {
|
|
match e {
|
|
TransportError::Closed { code, .. } => code,
|
|
other => panic!("{other:?}"),
|
|
}
|
|
}
|
|
|
|
/// A -23 session written from -23's messages: SETUP each way, with a varint Length; a
|
|
/// PUBLISH_NAMESPACE answered with REQUEST_OK; two SUBSCRIBE_NAMESPACEs whose prefixes
|
|
/// overlap, both accepted (-23 §4.2: -22 refused the second with PREFIX_OVERLAP).
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_raw_23_session() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (conn, _ep, s, _ctl) = setup23(&server).await;
|
|
assert!(s.options.implementation().is_some());
|
|
let mut held = Vec::new();
|
|
for (id, req) in [
|
|
(0, m23::ControlMessage::PublishNamespace(m23::NamespaceRequest { request_id: 0, prefix: ns(&["raw23"]), params: m23::Parameters::new() })),
|
|
(2, m23::ControlMessage::SubscribeNamespace(m23::NamespaceRequest { request_id: 2, prefix: ns(&["over"]), params: m23::Parameters::new() })),
|
|
(4, m23::ControlMessage::SubscribeNamespace(m23::NamespaceRequest { request_id: 4, prefix: ns(&["over", "lap"]), params: m23::Parameters::new() })),
|
|
] {
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
tx.write_all(&req.encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
let m = read23(&mut rx, &mut buf).await;
|
|
assert!(matches!(m, m23::ControlMessage::RequestOk(_)), "request {id}: {m:?}");
|
|
held.push((tx, rx));
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// Open a -23 session, write `bytes` on a request stream of its own (after `first`, when
|
|
/// given, and its answer), and return the code the relay closes the session with.
|
|
async fn fault(server: &joule_moq_relay::RelayServer, first: Option<m23::ControlMessage>, bytes: Vec<u8>) -> u64 {
|
|
let (conn, _ep, _, _ctl) = setup23(server).await;
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
if let Some(m) = first {
|
|
tx.write_all(&m.encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
let a = read23(&mut rx, &mut buf).await;
|
|
assert!(matches!(a, m23::ControlMessage::RequestOk(_)), "{a:?}");
|
|
}
|
|
tx.write_all(&bytes).await.unwrap();
|
|
closed_with(timeout(STEP, conn.closed()).await.expect("the relay closes the session"))
|
|
}
|
|
|
|
/// Planted faults, each on a session of its own, each closed with -23's code for it:
|
|
///
|
|
/// - a -22-only message, FETCH_OK (`0x18`, reserved at -23): UNEXPECTED_MESSAGE;
|
|
/// - a -22-only parameter, FORWARD (`0x10`), in a SUBSCRIBE: INVALID_PARAMETER;
|
|
/// - a message framed as -22 frames it (a 16-bit Length): MALFORMED_MESSAGE;
|
|
/// - a missing -23 field, PUBLISH_NAMESPACE without its Number of Parameters: MALFORMED_MESSAGE;
|
|
/// - a GOAWAY from the requester on its own request stream (-22 allowed it, -23 §9.2 does
|
|
/// not): PROTOCOL_VIOLATION;
|
|
/// - an empty Track Name (-23 §2.4.1): PROTOCOL_VIOLATION.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn each_planted_fault_closes_the_session_with_its_code() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let v = |x: u64| joule_moq::draft22::varint::to_vec(x);
|
|
let frame = |ty: u64, body: &[u8]| [v(ty), v(body.len() as u64), body.to_vec()].concat();
|
|
// -22's FETCH_OK: End Of Track 0, End Location {1, 0}, no parameters.
|
|
let fetch_ok = frame(0x18, &[0x00, 0x01, 0x00, 0x00]);
|
|
// A SUBSCRIBE (request 0, namespace ["a"], name "t") with FORWARD = 1.
|
|
let forward = frame(0x03, &[0x00, 0x01, 0x01, b'a', 0x01, b't', 0x01, 0x10, 0x01]);
|
|
// PUBLISH_NAMESPACE with a 16-bit Length, as -22 frames it.
|
|
let body22 = [0x00u8, 0x01, 0x01, b'p', 0x00];
|
|
let framed22 = [vec![0x06, 0x00, body22.len() as u8], body22.to_vec()].concat();
|
|
// PUBLISH_NAMESPACE (request 0, prefix ["p"]) and no Number of Parameters.
|
|
let missing = frame(0x06, &[0x00, 0x01, 0x01, b'p']);
|
|
// GOAWAY with no URI and no timeout, on the stream of an accepted PUBLISH_NAMESPACE.
|
|
let goaway = frame(0x10, &[0x00, 0x00]);
|
|
let pn = m23::ControlMessage::PublishNamespace(m23::NamespaceRequest { request_id: 0, prefix: ns(&["held"]), params: m23::Parameters::new() });
|
|
// SUBSCRIBE with an empty Track Name.
|
|
let empty = frame(0x03, &[0x00, 0x01, 0x01, b'a', 0x00, 0x00]);
|
|
let cases: Vec<(&str, Option<m23::ControlMessage>, Vec<u8>, u64)> = vec![
|
|
("FETCH_OK, a -22-only message", None, fetch_ok, 0x303),
|
|
("FORWARD, a -22-only parameter", None, forward, 0x305),
|
|
("a 16-bit Length", None, framed22, 0x300),
|
|
("no Number of Parameters", None, missing, 0x300),
|
|
("the requester's GOAWAY on its request stream", Some(pn), goaway, 0x3),
|
|
("an empty Track Name", None, empty, 0x3),
|
|
];
|
|
let mut caught = 0;
|
|
for (what, first, bytes, want) in cases {
|
|
let got = fault(&server, first, bytes).await;
|
|
assert_eq!(got, want, "{what}: closed with {got:#x}, not {want:#x}");
|
|
caught += 1;
|
|
}
|
|
assert_eq!(caught, 6);
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// What a session left: a -23 session's summary records `moqt-23`.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_23_sessions_summary_names_its_draft() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (c, _ep) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
|
|
let e = c.subscribe(ns(&["nothing", "here"]), name("b"), Parameters::new()).await.expect_err("no publisher");
|
|
assert!(e.contains("DOES_NOT_EXIST"), "{e}");
|
|
c.close(0, "done");
|
|
assert!(until(STEP, || !server.relay().summaries().is_empty()).await);
|
|
assert_eq!(server.relay().summaries()[0].summary.draft, "moqt-23");
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
// ── -23's object rules on a -23 session's subgroup streams ──────────────────
|
|
|
|
/// One object of group `g` on the track of alias `alias`, with `properties`, on a stream of
|
|
/// its own.
|
|
async fn send_object(c: &Client, alias: u64, g: u64, properties: joule_moq::draft22::kvp::Properties) {
|
|
let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(100), properties: true, end_of_group: true, first_object: true };
|
|
// A stream the relay stops for a malformed object may be stopped before its FIN is sent.
|
|
let _ = c.send_subgroup(h, &[SubgroupObject { object_id: 0, properties, status: WireStatus::Normal, payload: payload("o", g, 64) }]).await;
|
|
}
|
|
|
|
/// The groups of the objects `v` is delivered within `wait`, once `n` have come or it passes.
|
|
async fn groups_delivered(v: &Client, n: usize, wait: Duration) -> Vec<u64> {
|
|
let mut g: Vec<u64> = collect(v, wait, |x| x.iter().filter(|s| !s.objects.is_empty()).count() >= n).await.iter().filter(|s| !s.objects.is_empty()).map(|s| s.header.group_id).collect();
|
|
g.sort_unstable();
|
|
g
|
|
}
|
|
|
|
/// The same four objects from a -23 publisher and from a -22 one, each to a viewer at its own
|
|
/// draft. What -23 changed is what differs:
|
|
///
|
|
/// - a grease value in the Mandatory Track Property range on an object: at -23 no Mandatory
|
|
/// Track Property, forwarded; at -22 one, the track malformed;
|
|
/// - a Track-only property (DEFAULT_PUBLISHER_PRIORITY) on an object: at -23 outside its
|
|
/// scope, the track malformed; at -22 forwarded;
|
|
/// - OBJECT_DELIVERY_TIMEOUT twice, once in each property list: at -23 repeated, the track
|
|
/// malformed; at -22 forwarded;
|
|
/// - an object with an Object-scoped property once: forwarded at both.
|
|
///
|
|
/// A malformed object ends its stream and no more: the session stands, and the next object
|
|
/// is forwarded.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_23_sessions_objects_are_held_to_23s_rules() {
|
|
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
// 0x7f * 129 + 0x9d: a grease value (§13) inside 0x4000-0x7FFF.
|
|
let grease: u64 = 0x7f * 129 + 0x9d;
|
|
assert!(joule_moq::draft::is_grease(grease) && prop::MANDATORY_TRACK.contains(&grease) && grease.is_multiple_of(2));
|
|
let one = |kv: KeyValuePair| {
|
|
let mut p = Properties::new();
|
|
p.push(kv);
|
|
p
|
|
};
|
|
let twice = {
|
|
let mut p = one(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, 500));
|
|
p.set_immutable(&one(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, 700))).unwrap();
|
|
p
|
|
};
|
|
// (what, the object's properties, forwarded at -23, forwarded at -22)
|
|
let cases: Vec<(&str, Properties, bool, bool)> = vec![
|
|
("an Object-scoped property, once", one(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, 500)), true, true),
|
|
("a grease value in the Mandatory range", one(KeyValuePair::varint(grease, 1)), true, false),
|
|
("a Track-only property on an object", one(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 7)), false, true),
|
|
("a property that allows one instance, twice", twice, false, true),
|
|
];
|
|
let mut differ = 0;
|
|
for (draft, via) in [(Draft::D23, Via::QuicDraft(Draft::D23)), (Draft::D22, Via::Quic)] {
|
|
let nsp = ns(&["objects", draft.alpn_str()]);
|
|
let (p, _ep) = connect(&server, via, ClientOptions::default()).await;
|
|
let _held = p.publish(nsp.clone(), name("t"), 7, Parameters::new(), track_props(100, None)).await.expect("PUBLISH");
|
|
let (v, _ev) = connect(&server, via, ClientOptions::default()).await;
|
|
let _sub = v.subscribe(nsp.clone(), name("t"), Parameters::new()).await.expect("SUBSCRIBE");
|
|
let mut expect = Vec::new();
|
|
for (g, (what, props, at23, at22)) in cases.iter().enumerate() {
|
|
send_object(&p, 7, g as u64, props.clone()).await;
|
|
let forwarded = if draft == Draft::D23 { *at23 } else { *at22 };
|
|
if forwarded {
|
|
expect.push(g as u64);
|
|
}
|
|
differ += usize::from(at23 != at22 && draft == Draft::D23);
|
|
let _ = what;
|
|
}
|
|
// A last object, well formed: forwarded after every malformed one, the session open.
|
|
send_object(&p, 7, 9, Properties::new()).await;
|
|
expect.push(9);
|
|
let got = groups_delivered(&v, expect.len(), Duration::from_secs(5)).await;
|
|
assert_eq!(got, expect, "at {draft}: the objects forwarded, of {:?}", cases.iter().map(|c| c.0).collect::<Vec<_>>());
|
|
assert!(timeout(Duration::from_millis(300), p.conn.closed()).await.is_err(), "at {draft}: a malformed object closed its publisher's session");
|
|
p.close(0, "done");
|
|
v.close(0, "done");
|
|
}
|
|
assert_eq!(differ, 3, "three of the four are read differently at the two drafts");
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
// ── LARGEST_OBJECT, by -23 §7.5 ─────────────────────────────────────────────
|
|
|
|
/// What a TRACK_STATUS of `key`'s track answers for its Largest Object, asked by `c`.
|
|
async fn status_largest(c: &Client, nsp: &joule_moq::TrackNamespace) -> Option<Location> {
|
|
use joule_moq::draft22::{ControlMessage, TrackRequest};
|
|
let m = ControlMessage::TrackStatus(TrackRequest { request_id: c.next_request_id(), namespace: nsp.clone(), name: name("t"), params: Parameters::new() });
|
|
let (_tx, mut rx) = c.request(&m).await.expect("TRACK_STATUS");
|
|
match Client::response(&mut rx).await.expect("an answer") {
|
|
ControlMessage::RequestOk(ok) => ok.params.largest_object(),
|
|
other => panic!("{other:?}"),
|
|
}
|
|
}
|
|
|
|
/// -23 §7.5: a relay's LARGEST_OBJECT is the largest of the LARGEST_OBJECT values its upstream
|
|
/// publisher stated (in PUBLISH and PUBLISH_STATE_NOTIFY here) and the largest Location of an
|
|
/// object received on the upstream subscription. The vector: what the publisher states and
|
|
/// sends, step by step, and the Largest Object the relay answers after each, in TRACK_STATUS_OK
|
|
/// and in the SUBSCRIBE_OK of a subscription made then. A value stated in PUBLISH_STATE_NOTIFY
|
|
/// was not taken before; with it left out, step 3 answers {7, 4}.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_relays_largest_object_is_the_largest_stated_or_received() {
|
|
use joule_moq::draft22::ControlMessage;
|
|
#[derive(Clone, Copy)]
|
|
enum Step {
|
|
Notify(u64, u64),
|
|
Object(u64, u64),
|
|
}
|
|
// (what the publisher does, the Largest Object after it)
|
|
let vector: [(Step, (u64, u64)); 6] = [
|
|
(Step::Object(7, 4), (7, 4)), // an object past the stated {7, 3}
|
|
(Step::Notify(9, 0), (9, 0)), // stated in PUBLISH_STATE_NOTIFY, no object of it yet
|
|
(Step::Object(8, 5), (9, 0)), // an object of an earlier group arrives: no smaller
|
|
(Step::Notify(2, 0), (9, 0)), // a smaller value stated: the largest stands
|
|
(Step::Object(9, 1), (9, 1)), // an object past what was stated
|
|
(Step::Notify(9, 1), (9, 1)), // stated again as it is
|
|
];
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
for (draft, via) in [(Draft::D23, Via::QuicDraft(Draft::D23)), (Draft::D22, Via::Quic)] {
|
|
let nsp = ns(&["largest", draft.alpn_str()]);
|
|
let key = (nsp.clone(), name("t"));
|
|
let (p, _ep) = connect(&server, via, ClientOptions::default()).await;
|
|
let stated = Parameters::new().with(Parameter::LargestObject(Location::new(7, 3)));
|
|
let (tx, _rx) = p.publish(nsp.clone(), name("t"), 7, stated, track_props(100, None)).await.expect("PUBLISH");
|
|
let (v, _ev) = connect(&server, via, ClientOptions::default()).await;
|
|
assert_eq!(status_largest(&v, &nsp).await, Some(Location::new(7, 3)), "at {draft}: what PUBLISH stated, with no object received");
|
|
for (i, (step, (g, o))) in vector.iter().enumerate() {
|
|
match *step {
|
|
Step::Notify(g, o) => tx.send(&ControlMessage::PublishStateNotify(Parameters::new().with(Parameter::LargestObject(Location::new(g, o))))).await.expect("PUBLISH_STATE_NOTIFY"),
|
|
Step::Object(g, o) => {
|
|
let h = SubgroupHeader { track_alias: 7, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(100), properties: false, end_of_group: false, first_object: false };
|
|
p.send_subgroup(h, &[SubgroupObject::normal(o, payload("l", g, 32))]).await.expect("the object");
|
|
}
|
|
}
|
|
let want = Some(Location::new(*g, *o));
|
|
// A step that leaves the Largest Object where it was shows only by not moving it.
|
|
let _ = until(Duration::from_millis(400), || server.relay().largest(&key) != vector[..i].last().map_or(Some(Location::new(7, 3)), |(_, (g, o))| Some(Location::new(*g, *o)))).await;
|
|
assert!(until(STEP, || server.relay().largest(&key) == want).await, "at {draft}, step {}: the relay holds {:?}, not {want:?}", i + 1, server.relay().largest(&key));
|
|
assert_eq!(status_largest(&v, &nsp).await, want, "at {draft}, step {}: TRACK_STATUS_OK", i + 1);
|
|
let (ok, stx, _srx) = v.subscribe(nsp.clone(), name("t"), Parameters::new()).await.expect("SUBSCRIBE");
|
|
assert_eq!(ok.params.largest_object(), want, "at {draft}, step {}: SUBSCRIBE_OK", i + 1);
|
|
stx.reset(0x1).await;
|
|
}
|
|
p.close(0, "done");
|
|
v.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
// ── SUBSCRIBE_TRACKS at -23, written raw ───────────────────────────────────
|
|
|
|
/// What one data stream of a raw -23 session carried: its header, its subgroup objects or
|
|
/// fetch entries, and whether it ended in a reset.
|
|
#[derive(Debug)]
|
|
struct Arrived23 {
|
|
header: Option<m23::StreamHeader>,
|
|
objects: Vec<SubgroupObject>,
|
|
entries: Vec<m23::FetchEntry>,
|
|
reset: bool,
|
|
}
|
|
|
|
/// The data streams the relay opens to a raw -23 session, each read to its end, until `done`
|
|
/// has held of what arrived for `quiet` (what comes meanwhile is kept too) or `limit` passes.
|
|
async fn data23(conn: &SharedConnection, limit: Duration, quiet: Duration, mut done: impl FnMut(&[Arrived23]) -> bool) -> Vec<Arrived23> {
|
|
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Arrived23>();
|
|
let c = conn.clone();
|
|
let acceptor = tokio::spawn(async move {
|
|
while let Ok(mut r) = c.accept_uni().await {
|
|
let tx = tx.clone();
|
|
tokio::spawn(async move {
|
|
let mut d = m23::UniStreamDecoder::new(joule_moq::draft22::GroupOrder::Ascending, Default::default());
|
|
let mut a = Arrived23 { header: None, objects: Vec::new(), entries: Vec::new(), reset: false };
|
|
'stream: loop {
|
|
loop {
|
|
match d.next() {
|
|
Ok(Some(m23::UniEvent::Header(h))) => a.header = Some(h),
|
|
Ok(Some(m23::UniEvent::SubgroupObject(o))) => a.objects.push(o),
|
|
Ok(Some(m23::UniEvent::FetchEntry(e))) => a.entries.push(e),
|
|
Ok(Some(_)) => {}
|
|
Ok(None) => break,
|
|
Err(_) => break 'stream,
|
|
}
|
|
}
|
|
match r.read_chunk().await {
|
|
Ok(Some(b)) => d.push(&b),
|
|
Ok(None) => break,
|
|
Err(_) => {
|
|
a.reset = true;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
let _ = tx.send(a);
|
|
});
|
|
}
|
|
});
|
|
let deadline = tokio::time::Instant::now() + limit;
|
|
let mut got = Vec::new();
|
|
let mut since: Option<tokio::time::Instant> = None;
|
|
loop {
|
|
if done(&got) {
|
|
let s = *since.get_or_insert_with(tokio::time::Instant::now);
|
|
if s.elapsed() >= quiet {
|
|
break;
|
|
}
|
|
}
|
|
let until = since.map_or(deadline, |s| deadline.min(s + quiet));
|
|
match tokio::time::timeout_at(until, rx.recv()).await {
|
|
Ok(Some(a)) => got.push(a),
|
|
Ok(None) => break,
|
|
Err(_) if tokio::time::Instant::now() >= deadline => break,
|
|
Err(_) => {}
|
|
}
|
|
}
|
|
acceptor.abort();
|
|
got
|
|
}
|
|
|
|
/// A raw -23 SUBSCRIBE_TRACKS on a stream of its own, and the relay's answer; the stream is
|
|
/// returned to be held open.
|
|
async fn subscribe_tracks23(conn: &SharedConnection, request_id: u64, prefix: joule_moq::TrackNamespace, params: m23::Parameters) -> (joule_moq_relay::transport::BoxSend, BoxRecv, m23::ControlMessage) {
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
tx.write_all(&m23::ControlMessage::SubscribeTracks(m23::NamespaceRequest { request_id, prefix, params }).encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
let answer = read23(&mut rx, &mut buf).await;
|
|
(tx, rx, answer)
|
|
}
|
|
|
|
/// The relay's next PUBLISH to a raw -23 session, within `wait`, accepted with REQUEST_OK; its
|
|
/// stream returned to be held open. `None` if none comes.
|
|
async fn publish23(conn: &SharedConnection, wait: Duration) -> Option<(m23::Publish, joule_moq_relay::transport::BoxSend, BoxRecv)> {
|
|
let (mut tx, mut rx) = timeout(wait, conn.accept_bi()).await.ok()?.ok()?;
|
|
let mut buf = Vec::new();
|
|
let m = read23(&mut rx, &mut buf).await;
|
|
let m23::ControlMessage::Publish(p) = m else { panic!("the relay opened {m:?} where a PUBLISH was expected") };
|
|
tx.write_all(&m23::ControlMessage::RequestOk(m23::RequestOk { params: m23::Parameters::new(), track_properties: Default::default() }).encode().unwrap()).await.unwrap();
|
|
Some((p, tx, rx))
|
|
}
|
|
|
|
/// One object of group `g` of the track a -22 publisher published under `alias`.
|
|
async fn object22(c: &Client, alias: u64, g: u64, payload: &[u8]) {
|
|
let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: false, first_object: true };
|
|
c.send_subgroup(h, &[SubgroupObject::normal(0, payload.to_vec())]).await.expect("the object goes");
|
|
}
|
|
|
|
/// The groups whose objects arrived on the subscription `alias`, by their payloads.
|
|
fn payloads_of(got: &[Arrived23], alias: u64) -> Vec<Vec<u8>> {
|
|
let mut v: Vec<Vec<u8>> = got
|
|
.iter()
|
|
.filter(|a| matches!(&a.header, Some(m23::StreamHeader::Subgroup(h)) if h.track_alias == alias))
|
|
.flat_map(|a| a.objects.iter().map(|o| o.payload.clone()))
|
|
.collect();
|
|
v.sort();
|
|
v
|
|
}
|
|
|
|
/// -23 §3.6.2, §9.17: the subscription parameters a SUBSCRIBE_TRACKS carries that -22's has no
|
|
/// place for (both delivery timeouts, SUBSCRIBER_PRIORITY, LOCATION_FILTER), carried beside it by
|
|
/// the bridge (`Connection::carried`), are "the initial Subscription parameters when a PUBLISH is
|
|
/// sent as a result of SUBSCRIBE_TRACKS", and the relay "includes those in the PUBLISH": each is
|
|
/// in the PUBLISH, with SUBSCRIBE_TRACKS_ID, and the Location filter is where the subscription
|
|
/// starts, so groups before it are not sent. A request that carries none of them brings a
|
|
/// PUBLISH that states none.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_23_subscribe_tracks_carries_its_subscription_parameters() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (publ, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let (_p, _prx) = publ.publish(ns(&["carry", "a"]), name("t"), 61, Parameters::new(), Default::default()).await.expect("PUBLISH");
|
|
object22(&publ, 61, 0, b"g0").await;
|
|
let key = (ns(&["carry", "a"]), name("t"));
|
|
assert!(until(STEP, || server.relay().largest(&key) == Some(Location::new(0, 0))).await, "the relay holds group 0");
|
|
|
|
let (conn, _ep, _s, _ctl) = setup23(&server).await;
|
|
let filter = LocationFilter::AbsoluteStart { start: Location::new(3, 0) };
|
|
let asked = m23::Parameters::new()
|
|
.with(m23::Parameter::ObjectDeliveryTimeout(60_000))
|
|
.with(m23::Parameter::SubgroupDeliveryTimeout(90_000))
|
|
.with(m23::Parameter::SubscriberPriority(7))
|
|
.with(m23::Parameter::LocationFilter(filter));
|
|
let (_ttx, _trx, answer) = subscribe_tracks23(&conn, 0, ns(&["carry"]), asked).await;
|
|
assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "{answer:?}");
|
|
let (p, _ptx, _prx2) = publish23(&conn, STEP).await.expect("a PUBLISH of the track");
|
|
let has = |ty: u64| p.params.get(ty).cloned();
|
|
use m23::params::param;
|
|
assert_eq!(has(param::OBJECT_DELIVERY_TIMEOUT), Some(m23::Parameter::ObjectDeliveryTimeout(60_000)), "{:?}", p.params);
|
|
assert_eq!(has(param::SUBGROUP_DELIVERY_TIMEOUT), Some(m23::Parameter::SubgroupDeliveryTimeout(90_000)), "{:?}", p.params);
|
|
assert_eq!(has(param::SUBSCRIBER_PRIORITY), Some(m23::Parameter::SubscriberPriority(7)), "{:?}", p.params);
|
|
assert_eq!(p.params.location_filter(), Some(filter), "{:?}", p.params);
|
|
assert_eq!(p.params.subscribe_tracks_id(), Some(0), "the PUBLISH names its request (§9.19.21)");
|
|
assert_eq!(p.params.largest_object(), Some(Location::new(0, 0)), "and the Largest Object");
|
|
|
|
// Groups 1 and 2 are before the filter's start, 3 and 4 in it.
|
|
for g in 1..=4u64 {
|
|
object22(&publ, 61, g, format!("g{g}").as_bytes()).await;
|
|
}
|
|
let got = data23(&conn, STEP, Duration::from_millis(500), |a| {
|
|
let have = payloads_of(a, p.track_alias);
|
|
have.contains(&b"g3".to_vec()) && have.contains(&b"g4".to_vec())
|
|
})
|
|
.await;
|
|
assert_eq!(payloads_of(&got, p.track_alias), vec![b"g3".to_vec(), b"g4".to_vec()], "the subscription starts where its Location filter does");
|
|
assert!(got.iter().all(|a| !a.reset && a.entries.is_empty()), "every stream whole, and no fill, which the request did not ask for: {got:?}");
|
|
|
|
// A request that carries none of them: its PUBLISH states none.
|
|
let (conn2, _ep2, _s2, _ctl2) = setup23(&server).await;
|
|
let (_ttx2, _trx2, answer) = subscribe_tracks23(&conn2, 0, ns(&["carry"]), m23::Parameters::new()).await;
|
|
assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "{answer:?}");
|
|
let (q, _qtx, _qrx) = publish23(&conn2, STEP).await.expect("a PUBLISH of the track");
|
|
for ty in [param::OBJECT_DELIVERY_TIMEOUT, param::SUBGROUP_DELIVERY_TIMEOUT, param::SUBSCRIBER_PRIORITY, param::LOCATION_FILTER, param::FILL_PARAMETERS] {
|
|
assert!(q.params.get(ty).is_none(), "{ty:#x} in a PUBLISH whose request carried none: {:?}", q.params);
|
|
}
|
|
publ.close(0, "done");
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// -23 §3.4, §3.4.1, §3.6.2, §9.19.15: FILL_PARAMETERS in a SUBSCRIBE_TRACKS, which the relay
|
|
/// refused NOT_SUPPORTED before. Each PUBLISH the request brings carries them, and opens a fill
|
|
/// fetch stream whose FETCH_HEADER names the PUBLISH's Request ID, carrying the fill range (the
|
|
/// Location filter inside them, here the whole track to its Largest Object); the subscription's
|
|
/// own objects follow on its subgroups. A request that begins its subscriptions paused (PAUSED
|
|
/// 1) has its PUBLISHes carry the parameters and open no fill ("FILL_PARAMETERS carried while
|
|
/// the subscription is paused opens no fill fetch stream").
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_23_subscribe_tracks_with_fill_parameters() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (publ, _pe) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let tname = name("t");
|
|
let (_p, _prx) = publ.publish(ns(&["fill", "a"]), tname.clone(), 62, Parameters::new(), Default::default()).await.expect("PUBLISH");
|
|
for g in 0..3u64 {
|
|
object22(&publ, 62, g, format!("f{g}").as_bytes()).await;
|
|
}
|
|
let key = (ns(&["fill", "a"]), tname);
|
|
assert!(until(STEP, || server.relay().largest(&key) == Some(Location::new(2, 0))).await, "the relay holds groups 0 to 2");
|
|
|
|
let (conn, _ep, _s, _ctl) = setup23(&server).await;
|
|
let inner = m23::Parameters::new().with(m23::Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }));
|
|
let asked = m23::Parameters::new().with(m23::Parameter::FillParameters(inner.clone()));
|
|
let (_ttx, _trx, answer) = subscribe_tracks23(&conn, 0, ns(&["fill"]), asked).await;
|
|
assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "SUBSCRIBE_TRACKS with FILL_PARAMETERS is accepted: {answer:?}");
|
|
let (p, _ptx, _prx2) = publish23(&conn, STEP).await.expect("a PUBLISH of the track");
|
|
assert_eq!(p.params.get(m23::params::param::FILL_PARAMETERS).cloned(), Some(m23::Parameter::FillParameters(inner.clone())), "the PUBLISH carries them: {:?}", p.params);
|
|
assert_eq!(p.params.subscribe_tracks_id(), Some(0));
|
|
object22(&publ, 62, 3, b"f3").await;
|
|
let fill_of = |a: &Arrived23| matches!(&a.header, Some(m23::StreamHeader::Fetch(h)) if h.request_id == p.request_id);
|
|
let got = data23(&conn, STEP, Duration::from_millis(300), |a| a.iter().any(fill_of) && payloads_of(a, p.track_alias).contains(&b"f3".to_vec())).await;
|
|
let fill = got.iter().find(|a| fill_of(a)).unwrap_or_else(|| panic!("a fill fetch stream named by the PUBLISH's Request ID: {got:?}"));
|
|
assert!(!fill.reset, "the fill is served, not reset");
|
|
let filled: Vec<(u64, Vec<u8>)> = fill.entries.iter().filter_map(|e| if let m23::FetchEntry::Object(o) = e { Some((o.group_id, o.payload.clone())) } else { None }).collect();
|
|
assert_eq!(filled, vec![(0, b"f0".to_vec()), (1, b"f1".to_vec()), (2, b"f2".to_vec())], "the fill range: the track to its Largest Object");
|
|
assert_eq!(payloads_of(&got, p.track_alias), vec![b"f3".to_vec()], "the subscription's own objects, from the Next Object");
|
|
|
|
// Paused: the PUBLISH carries FILL_PARAMETERS and no fill is opened.
|
|
let (conn2, _ep2, _s2, _ctl2) = setup23(&server).await;
|
|
let paused = m23::Parameters::new().with(m23::Parameter::Paused(true)).with(m23::Parameter::FillParameters(inner.clone()));
|
|
let (_ttx2, _trx2, answer) = subscribe_tracks23(&conn2, 0, ns(&["fill"]), paused).await;
|
|
assert!(matches!(answer, m23::ControlMessage::RequestOk(_)), "{answer:?}");
|
|
let (q, _qtx, _qrx) = publish23(&conn2, STEP).await.expect("a PUBLISH of the track");
|
|
assert_eq!(q.params.paused(), Some(true));
|
|
assert!(q.params.get(m23::params::param::FILL_PARAMETERS).is_some(), "{:?}", q.params);
|
|
let got = data23(&conn2, Duration::from_millis(1_500), Duration::ZERO, |_| false).await;
|
|
assert!(got.iter().all(|a| !matches!(a.header, Some(m23::StreamHeader::Fetch(_)))), "no fill for a paused subscription: {got:?}");
|
|
publ.close(0, "done");
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|