d2c5dbd483
-16 §9.1 asks for each new Request ID to be "the next in sequence", but its SUBSCRIBE_NAMESPACE rides a bidirectional stream of its own while the other requests ride the control stream, and QUIC does not order one stream against another. A public relay network's -16 endpoint opened a SUBSCRIBE_NAMESPACE (ID 1) and sent a PUBLISH_NAMESPACE (ID 3) on the control stream; the 3 came first and the bridge closed the session. The bridge now admits a new ID of the peer's parity within the grant and leaves the gap for the one in flight; an ID seen before still closes the session with INVALID_REQUEST_ID, and one past the grant with TOO_MANY_REQUESTS. a_raw_16_session sends 4 before 2 and then 4 again. wai-receipts collects until every object has arrived (by alias, group and object), not a count of streams: a relay may carry a subgroup on more streams than its publisher opened, and the count stopped the subscriber before the last group. Co-Authored-By: Claude Opus 5.5 <[email protected]>
357 lines
22 KiB
Rust
357 lines
22 KiB
Rust
//! Sessions at `draft-ietf-moq-transport-18` and `-16` through the relay, beside -22
|
|
//! ones: the relay chooses each session's draft by ALPN or `WT-Available-Protocols` and
|
|
//! bridges it to its -22 model (`transport::bridge`). WAI objects and their receipts
|
|
//! cross every pairing of drafts and verify at each subscriber; a FETCH from the cache
|
|
//! returns them at -18 and -16; what one draft cannot say is refused by name; and two
|
|
//! sessions written byte by byte from the drafts' own messages, not through this crate's
|
|
//! client, check the relay's -18 and -16 wire.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use joule_moq::draft::Draft;
|
|
use joule_moq::draft22::data::FetchEntry;
|
|
use joule_moq::draft22::{Location, Parameter, Parameters};
|
|
use joule_moq::receipt::ReceiptVerifier;
|
|
use joule_moq::{draft16 as m16, draft18 as m18};
|
|
use joule_moq_relay::client::ClientOptions;
|
|
use joule_moq_relay::transport::{quic, 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_18_over_quic() {
|
|
timeout(LIMIT, two_publishers_three_subscribers(Via::QuicDraft(Draft::D18))).await.expect("in time");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn every_session_at_16_over_quic() {
|
|
timeout(LIMIT, two_publishers_three_subscribers(Via::QuicDraft(Draft::D16))).await.expect("in time");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn every_session_at_18_over_webtransport() {
|
|
timeout(LIMIT, two_publishers_three_subscribers(Via::WtDraft(Draft::D18))).await.expect("in time");
|
|
}
|
|
|
|
/// Publishers at -16 and -18, subscribers at -22, -18 (WebTransport) and -16: every
|
|
/// object reaches every subscriber and verifies against its publisher's receipts, which
|
|
/// crossed two bridges unchanged.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn publishers_and_subscribers_at_different_drafts() {
|
|
let vias = [Via::QuicDraft(Draft::D16), Via::QuicDraft(Draft::D18), Via::Quic, Via::WtDraft(Draft::D18), Via::QuicDraft(Draft::D16)];
|
|
timeout(LIMIT, two_publishers_three_subscribers_on(vias)).await.expect("in time");
|
|
}
|
|
|
|
/// A late joiner at -18 and one at -16 FETCH a track and its receipts from the relay's
|
|
/// cache, published at -22: the exclusive End Locations of -18 and -16 and the absolute
|
|
/// Group IDs of a -16 fetch stream come back to the same objects, which verify.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_fetch_from_the_cache_at_18_and_16() {
|
|
timeout(LIMIT, async {
|
|
const GROUPS: u64 = 6;
|
|
let server = relay(|_| {}).await;
|
|
let key = ed25519_dalek::SigningKey::from_bytes(&[0x2b; 32]);
|
|
let t = Tracks::new(ns(&["vod", "drafts"]), "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 d in [Draft::D18, Draft::D16] {
|
|
let (late, _ep) = connect(&server, Via::QuicDraft(d), ClientOptions::default()).await;
|
|
let (ok, media) = late.fetch(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap_or_else(|e| panic!("FETCH at {d}: {e}"));
|
|
assert_eq!(ok.end_location, Location::new(GROUPS - 1, 0), "{d}: the inclusive end, back from the draft's exclusive one");
|
|
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), "{d}");
|
|
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));
|
|
}
|
|
// A range: groups 2 to 3, whole, which -18 and -16 say as End {4, 0}.
|
|
let range = Parameters::new().with(Parameter::LocationFilter(joule_moq::draft22::LocationFilter::AbsoluteStartGroupEnd { start: Location::new(2, 0), end_group_delta: 1 }));
|
|
let (_, 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], "{d}");
|
|
assert!(late.conn.notes().iter().all(|n| !n.starts_with("refused")), "{d}: {:?}", late.conn.notes());
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// A SUBSCRIBE whose Range Filter -18 cannot say is refused at this end, by name, and
|
|
/// never reaches the relay; the session goes on.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn what_18_cannot_say_is_refused_here() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (c, _ep) = connect(&server, Via::QuicDraft(Draft::D18), ClientOptions::default()).await;
|
|
let p = Parameters::new().with(Parameter::RangeFilter { ty: joule_moq::draft22::params::param::OBJECTID_FILTER, filter: None });
|
|
let e = c.subscribe(ns(&["a"]), name("b"), p).await.expect_err("refused");
|
|
assert!(e.contains("REQUEST_ERROR NOT_SUPPORTED") && e.contains("Range Filter"), "{e}");
|
|
assert!(c.conn.notes().iter().any(|n| n.starts_with("refused: a Range Filter has no -18 form")), "{:?}", c.conn.notes());
|
|
// The session is whole: a request -18 can say gets the relay's answer.
|
|
let e = c.subscribe(ns(&["nothing", "here"]), name("b"), Parameters::new()).await.expect_err("no publisher");
|
|
assert!(e.contains("DOES_NOT_EXIST"), "{e}");
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
async fn raw_session(server: &joule_moq_relay::RelayServer, d: Draft) -> (SharedConnection, quinn::Endpoint) {
|
|
let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![d.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, d);
|
|
(Arc::new(conn), ep)
|
|
}
|
|
|
|
/// Read framed messages (a 16-bit Length after the type) of the draft whose integer's
|
|
/// length `len` reads from a first byte, until one is whole.
|
|
async fn read_framed(r: &mut joule_moq_relay::transport::BoxRecv, buf: &mut Vec<u8>, len: fn(u8) -> usize) -> Vec<u8> {
|
|
loop {
|
|
if let Some(&b0) = buf.first() {
|
|
let tl = len(b0);
|
|
if buf.len() >= tl + 2 {
|
|
let n = tl + 2 + u16::from_be_bytes([buf[tl], buf[tl + 1]]) as usize;
|
|
if buf.len() >= n {
|
|
return buf.drain(..n).collect();
|
|
}
|
|
}
|
|
}
|
|
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"),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// An -18 session written from -18's messages: SETUP each way on unidirectional streams,
|
|
/// a PUBLISH_NAMESPACE answered with REQUEST_OK, a Joining FETCH (no -22 form) refused
|
|
/// with NOT_SUPPORTED, a FETCH with Fetch Type 4 closing the session.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_raw_18_session() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (conn, _ep) = raw_session(&server, Draft::D18).await;
|
|
let mut ctl = conn.open_uni().await.unwrap();
|
|
let mut o = m18::kvp::SetupOptions::new();
|
|
o.push(m18::kvp::KeyValuePair::bytes(m18::kvp::opt::MOQT_IMPLEMENTATION, b"raw-18".to_vec()));
|
|
ctl.write_all(&m18::ControlMessage::Setup(m18::Setup { options: o }).encode().unwrap()).await.unwrap();
|
|
// The relay's SETUP, on its control stream.
|
|
let mut theirs = timeout(STEP, conn.accept_uni()).await.expect("in time").expect("a stream");
|
|
let mut d = m18::UniStreamDecoder::new(joule_moq::draft22::GroupOrder::Ascending, Default::default());
|
|
let setup = loop {
|
|
match d.next().unwrap() {
|
|
Some(m18::UniEvent::Control(m)) => break m,
|
|
Some(_) => continue,
|
|
None => d.push(&timeout(STEP, theirs.read_chunk()).await.unwrap().unwrap().unwrap()),
|
|
}
|
|
};
|
|
let m18::ControlMessage::Setup(s) = setup else { panic!("{setup:?}") };
|
|
assert!(s.options.implementation().is_some());
|
|
assert!(s.options.0.iter().all(|p| p.ty != 0x06 && p.ty != 0x08), "no -22-only option reaches an -18 peer");
|
|
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
let pn = m18::ControlMessage::PublishNamespace(m18::NamespaceRequest { request_id: 0, prefix: ns(&["raw18"]), params: m18::Parameters::new() });
|
|
tx.write_all(&pn.encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
let m = m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap();
|
|
assert!(matches!(m, m18::ControlMessage::RequestOk(_)), "{m:?}");
|
|
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
let j = m18::ControlMessage::Fetch(m18::Fetch { request_id: 2, kind: m18::FetchKind::RelativeJoining { joining_request_id: 0, joining_start: 1 }, params: m18::Parameters::new() });
|
|
tx.write_all(&j.encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
match m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap() {
|
|
m18::ControlMessage::RequestError(e) => assert_eq!(e.code, m18::codes::RequestErrorCode::NOT_SUPPORTED),
|
|
other => panic!("{other:?}"),
|
|
}
|
|
|
|
let (mut tx, _rx) = conn.open_bi().await.unwrap();
|
|
tx.write_all(&[0x16, 0x00, 0x03, 0x04, 0x04, 0x00]).await.unwrap();
|
|
match timeout(STEP, conn.closed()).await.expect("the relay closes the session") {
|
|
TransportError::Closed { code, .. } => assert_eq!(code, 0x3, "PROTOCOL_VIOLATION"),
|
|
other => panic!("{other:?}"),
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// A FIN from an -18 requester on PUBLISH_NAMESPACE withdraws the namespace (-18 §6.2
|
|
/// withdraws it by cancelling the request; -18 has no FIN that leaves a request
|
|
/// standing), while a -22 publisher's FIN does not (-22 §6.4.2.2). A -22 client's FIN
|
|
/// sent through an -18 session is withheld, so it does not withdraw either.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_18_publishers_fin_withdraws_its_namespace() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (sub, _e1) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let (_stx, mut srx) = sub.subscribe_namespace(ns(&["fin"])).await.expect("SUBSCRIBE_NAMESPACE");
|
|
// An -18 publisher written from -18's messages: PUBLISH_NAMESPACE, then a FIN.
|
|
let (conn, _ep) = raw_session(&server, Draft::D18).await;
|
|
let mut ctl = conn.open_uni().await.unwrap();
|
|
ctl.write_all(&m18::ControlMessage::Setup(m18::Setup { options: m18::kvp::SetupOptions::new() }).encode().unwrap()).await.unwrap();
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
let pn = m18::ControlMessage::PublishNamespace(m18::NamespaceRequest { request_id: 0, prefix: ns(&["fin", "raw18"]), params: m18::Parameters::new() });
|
|
tx.write_all(&pn.encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
let m = m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap();
|
|
assert!(matches!(m, m18::ControlMessage::RequestOk(_)), "{m:?}");
|
|
match timeout(STEP, srx.next()).await {
|
|
Ok(Ok(Some(joule_moq::draft22::ControlMessage::Namespace(s)))) => assert_eq!(s, ns(&["raw18"])),
|
|
other => panic!("no NAMESPACE: {other:?}"),
|
|
}
|
|
tx.finish().unwrap();
|
|
match timeout(STEP, srx.next()).await {
|
|
Ok(Ok(Some(joule_moq::draft22::ControlMessage::NamespaceDone(s)))) => assert_eq!(s, ns(&["raw18"])),
|
|
other => panic!("the -18 FIN did not withdraw the namespace: {other:?}"),
|
|
}
|
|
|
|
// A -22 publisher, and a -22 client over -18: each FINs its PUBLISH_NAMESPACE, and
|
|
// each namespace stands.
|
|
for (via, field) in [(Via::Quic, "at22"), (Via::QuicDraft(Draft::D18), "through18")] {
|
|
let (p, _e) = connect(&server, via, ClientOptions::default()).await;
|
|
let (ptx, _prx) = p.publish_namespace(ns(&["fin", field]), Parameters::new()).await.expect("PUBLISH_NAMESPACE");
|
|
match timeout(STEP, srx.next()).await {
|
|
Ok(Ok(Some(joule_moq::draft22::ControlMessage::Namespace(s)))) => assert_eq!(s, ns(&[field])),
|
|
other => panic!("{via:?}: no NAMESPACE: {other:?}"),
|
|
}
|
|
ptx.fin().await;
|
|
let after = timeout(Duration::from_millis(500), srx.next()).await;
|
|
assert!(after.is_err(), "{via:?}: a FIN that is not a withdrawal withdrew it: {after:?}");
|
|
// The namespace goes with the session.
|
|
p.close(0, "done");
|
|
match timeout(STEP, srx.next()).await {
|
|
Ok(Ok(Some(joule_moq::draft22::ControlMessage::NamespaceDone(s)))) => assert_eq!(s, ns(&[field])),
|
|
other => panic!("{via:?}: no NAMESPACE_DONE at the session's end: {other:?}"),
|
|
}
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// An -18 publisher that closes its control stream while a data stream of its is still
|
|
/// arriving and before its PUBLISH_DONE: closing the control stream is a protocol
|
|
/// violation (-18 §3.3), but the relay reads the stream to its end and takes the
|
|
/// PUBLISH_DONE before it closes the session for it, so the subscriber gets every object
|
|
/// and the publisher's own status.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_18_publisher_that_closes_its_control_stream_first_loses_nothing() {
|
|
timeout(LIMIT, async {
|
|
use joule_moq::draft22::data::{SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
let server = relay(|_| {}).await;
|
|
let (sub, _e) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let (conn, _ep) = raw_session(&server, Draft::D18).await;
|
|
let mut ctl = conn.open_uni().await.unwrap();
|
|
ctl.write_all(&m18::ControlMessage::Setup(m18::Setup { options: m18::kvp::SetupOptions::new() }).encode().unwrap()).await.unwrap();
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
let p = m18::ControlMessage::Publish(m18::Publish { request_id: 0, namespace: ns(&["late"]), name: name("track"), track_alias: 1, params: m18::Parameters::new(), track_properties: Default::default() });
|
|
tx.write_all(&p.encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
let m = m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap();
|
|
assert!(matches!(m, m18::ControlMessage::RequestOk(_)), "{m:?}");
|
|
let (_ok, _stx, mut srx) = sub.subscribe(ns(&["late"]), name("track"), Parameters::new()).await.expect("SUBSCRIBE");
|
|
|
|
// The data stream's header and first object; the control stream closed; then the
|
|
// second object, the stream's FIN and PUBLISH_DONE.
|
|
let h = SubgroupHeader { track_alias: 1, group_id: 0, subgroup_id: SubgroupId::Zero, publisher_priority: Some(0), properties: false, end_of_group: true, first_object: true };
|
|
let mut codec = SubgroupCodec::new(h);
|
|
let (mut first, mut second) = (Vec::new(), Vec::new());
|
|
h.encode(&mut first);
|
|
codec.encode_object(&SubgroupObject::normal(0, b"before".to_vec()), &mut first).unwrap();
|
|
codec.encode_object(&SubgroupObject::normal(1, b"after".to_vec()), &mut second).unwrap();
|
|
let mut data = conn.open_uni().await.unwrap();
|
|
data.write_all(&first).await.unwrap();
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
ctl.finish().unwrap();
|
|
tokio::time::sleep(Duration::from_millis(300)).await;
|
|
data.write_all(&second).await.unwrap();
|
|
data.finish().unwrap();
|
|
let done = m18::ControlMessage::PublishDone(m18::PublishDone { status: m18::codes::PublishDoneCode::TRACK_ENDED, stream_count: 1, reason: Vec::new() });
|
|
tx.write_all(&done.encode().unwrap()).await.unwrap();
|
|
|
|
let got = collect(&sub, STEP, |g| g.iter().map(|x| x.objects.len()).sum::<usize>() >= 2).await;
|
|
let payloads: Vec<&[u8]> = got.iter().flat_map(|g| &g.objects).map(|o| o.payload.as_slice()).collect();
|
|
assert_eq!(payloads, vec![&b"before"[..], &b"after"[..]]);
|
|
match timeout(STEP, srx.next()).await {
|
|
Ok(Ok(Some(joule_moq::draft22::ControlMessage::PublishDone(d)))) => assert_eq!(d.status, joule_moq::draft22::PublishDoneCode::TRACK_ENDED),
|
|
other => panic!("no PUBLISH_DONE: {other:?}"),
|
|
}
|
|
// And then the session closes, for the closed control stream.
|
|
match timeout(STEP, conn.closed()).await.expect("the relay closes the session") {
|
|
TransportError::Closed { code, .. } => assert_eq!(code, 0x3, "PROTOCOL_VIOLATION"),
|
|
other => panic!("{other:?}"),
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// A -16 session written from -16's messages: CLIENT_SETUP on the bidirectional control
|
|
/// stream the client opens, SERVER_SETUP back with MAX_REQUEST_ID, a PUBLISH_NAMESPACE
|
|
/// answered by Request ID; a request whose ID is ahead of one still in flight on another
|
|
/// stream admitted (a SUBSCRIBE at 4 before the SUBSCRIBE_NAMESPACE at 2 on its own
|
|
/// stream, which then fills the gap), and a Request ID seen before closing the session
|
|
/// with INVALID_REQUEST_ID.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_raw_16_session() {
|
|
timeout(LIMIT, async {
|
|
let server = relay(|_| {}).await;
|
|
let (conn, _ep) = raw_session(&server, Draft::D16).await;
|
|
let (mut tx, mut rx) = conn.open_bi().await.unwrap();
|
|
let mut p = m16::kvp::SetupParameters::new();
|
|
p.push(m16::kvp::KeyValuePair::varint(m16::kvp::setup::MAX_REQUEST_ID, 100)).push(m16::kvp::KeyValuePair::bytes(m16::kvp::setup::MOQT_IMPLEMENTATION, b"raw-16".to_vec()));
|
|
tx.write_all(&m16::ControlMessage::ClientSetup(p).encode().unwrap()).await.unwrap();
|
|
let mut buf = Vec::new();
|
|
let m = m16::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, m16::io::len_from_first).await).unwrap();
|
|
let m16::ControlMessage::ServerSetup(sp) = m else { panic!("{m:?}") };
|
|
assert!(sp.max_request_id() > 0, "the relay grants requests");
|
|
|
|
let pn = m16::ControlMessage::PublishNamespace(m16::control::PublishNamespace { request_id: 0, namespace: ns(&["raw16"]), params: m16::params::Parameters::new() });
|
|
tx.write_all(&pn.encode().unwrap()).await.unwrap();
|
|
let m = m16::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, m16::io::len_from_first).await).unwrap();
|
|
assert!(matches!(&m, m16::ControlMessage::RequestOk(o) if o.request_id == 0), "{m:?}");
|
|
|
|
let s = m16::ControlMessage::Subscribe(m16::control::TrackRequest { request_id: 4, namespace: ns(&["elsewhere"]), name: name("t"), params: m16::params::Parameters::new() });
|
|
tx.write_all(&s.encode().unwrap()).await.unwrap();
|
|
let m = m16::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, m16::io::len_from_first).await).unwrap();
|
|
assert!(matches!(&m, m16::ControlMessage::RequestError(e) if e.request_id == 4), "the request ahead is answered: {m:?}");
|
|
let (mut ntx, mut nrx) = conn.open_bi().await.unwrap();
|
|
let sn = m16::ControlMessage::SubscribeNamespace(m16::control::SubscribeNamespace { request_id: 2, prefix: ns(&["somewhere"]), options: m16::control::SubscribeOptions::Namespace, params: m16::params::Parameters::new() });
|
|
ntx.write_all(&sn.encode().unwrap()).await.unwrap();
|
|
let mut nbuf = Vec::new();
|
|
let m = m16::ControlMessage::decode_exact(&read_framed(&mut nrx, &mut nbuf, m16::io::len_from_first).await).unwrap();
|
|
assert!(matches!(&m, m16::ControlMessage::RequestOk(o) if o.request_id == 2), "the request behind it, on its own stream: {m:?}");
|
|
|
|
tx.write_all(&s.encode().unwrap()).await.unwrap();
|
|
match timeout(STEP, conn.closed()).await.expect("the relay closes the session") {
|
|
TransportError::Closed { code, .. } => assert_eq!(code, 0x4, "INVALID_REQUEST_ID"),
|
|
other => panic!("{other:?}"),
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|