moqtap_client/dispatch.rs
1//! Unified multi-draft entry-point types.
2//!
3//! This module is the facade downstream consumers use to hold a MoQT
4//! connection without caring which draft was negotiated.
5//! It mirrors [`moqtap_codec::dispatch`]: one enum variant per enabled draft,
6//! gated on its feature flag.
7//!
8//! Four types live here:
9//!
10//! - `AnyConnection` — wraps a draft-specific `Connection`.
11//! - `AnyClientEvent` — wraps a draft-specific `ClientEvent`.
12//! - `AnyConnectionObserver` — a trait that receives `AnyClientEvent`s.
13//! Attached to an `AnyConnection` via `AnyConnection::set_observer`,
14//! which installs a per-draft adapter on the inner connection.
15//! - `AnyRequest` — what a request made through `AnyConnection` leaves the
16//! caller holding. Drafts 07-16 put every request on the one control
17//! stream and hand back only a request ID; from draft-17 on each request
18//! owns a bidirectional stream, and the handle owns that stream.
19//!
20//! `AnyConnection` carries only the handful of protocol methods whose
21//! arguments can be reconciled across every draft that has the message
22//! (`subscribe`, `fetch`, `track_status`, `subscribe_namespace`,
23//! `publish_namespace`, `publish`, `subscribe_tracks`). The rest differ too
24//! much in signature — match on the variant to reach them.
25
26use std::sync::Arc;
27
28use moqtap_codec::kvp::KeyValuePair;
29use moqtap_codec::version::DraftVersion;
30
31/// Generates the `AnyConnection` and `AnyClientEvent` enums plus the per-draft
32/// observer adapter, with one variant per enabled draft feature.
33macro_rules! dispatch_all {
34 (
35 $(
36 #[cfg(feature = $feat:literal)]
37 $variant:ident => $module:ident,
38 )+
39 ) => {
40 /// A MoQT client connection of any enabled draft version.
41 ///
42 /// Wraps the draft-specific `Connection` type. Methods common to all
43 /// drafts are forwarded; for draft-specific protocol calls, match on
44 /// the variant.
45 pub enum AnyConnection {
46 $(
47 #[cfg(feature = $feat)]
48 #[doc = concat!("A draft-", $feat, " connection.")]
49 $variant(crate::$module::connection::Connection),
50 )+
51 }
52
53 impl AnyConnection {
54 /// Returns the draft version this connection is using.
55 #[allow(unreachable_code)]
56 pub fn draft(&self) -> DraftVersion {
57 match self {
58 $(
59 #[cfg(feature = $feat)]
60 Self::$variant(_) => DraftVersion::$variant,
61 )+
62 #[allow(unreachable_patterns)]
63 _ => unreachable!("AnyConnection has no enabled variants"),
64 }
65 }
66
67 /// The SETUP message the server answered the handshake with.
68 ///
69 /// `SERVER_SETUP` through draft-16, the server's half of the
70 /// unified `SETUP` from draft-17.
71 /// [`fields`](moqtap_codec::dispatch::AnyControlMessage::fields)
72 /// renders its parameters under the negotiated draft's own names,
73 /// in the order they arrived — which is what makes one relay's
74 /// setup response comparable to another's.
75 #[allow(unreachable_code)]
76 pub fn server_setup(&self) -> &moqtap_codec::dispatch::AnyControlMessage {
77 match self {
78 $(
79 #[cfg(feature = $feat)]
80 Self::$variant(c) => c.server_setup(),
81 )+
82 #[allow(unreachable_patterns)]
83 _ => unreachable!("AnyConnection has no enabled variants"),
84 }
85 }
86
87 /// The framed wire bytes of [`Self::server_setup`], as they
88 /// arrived.
89 pub fn server_setup_raw(&self) -> Option<&[u8]> {
90 match self {
91 $(
92 #[cfg(feature = $feat)]
93 Self::$variant(c) => c.server_setup_raw(),
94 )+
95 #[allow(unreachable_patterns)]
96 _ => None,
97 }
98 }
99
100 /// Attach an observer. The observer is adapted into the
101 /// draft-specific observer trait and installed on the inner
102 /// connection; events are forwarded as [`AnyClientEvent`].
103 ///
104 /// Replaces any previously attached observer.
105 #[allow(unused_variables)]
106 pub fn set_observer(&mut self, observer: Arc<dyn AnyConnectionObserver>) {
107 match self {
108 $(
109 #[cfg(feature = $feat)]
110 Self::$variant(c) => {
111 c.set_observer(Box::new($variant::Adapter(observer)));
112 }
113 )+
114 #[allow(unreachable_patterns)]
115 _ => {}
116 }
117 }
118
119 /// Remove any attached observer.
120 pub fn clear_observer(&mut self) {
121 match self {
122 $(
123 #[cfg(feature = $feat)]
124 Self::$variant(c) => c.clear_observer(),
125 )+
126 #[allow(unreachable_patterns)]
127 _ => {}
128 }
129 }
130
131 /// Close the connection with the given application error code
132 /// and reason.
133 #[allow(unused_variables)]
134 pub fn close(&self, code: u32, reason: &[u8]) {
135 match self {
136 $(
137 #[cfg(feature = $feat)]
138 Self::$variant(c) => c.close(code, reason),
139 )+
140 #[allow(unreachable_patterns)]
141 _ => {}
142 }
143 }
144 }
145
146 /// An event from a MoQT connection of any enabled draft version.
147 ///
148 /// Event shapes differ across drafts (e.g. draft-17's
149 /// `SubgroupObjectReceived` carries header types, while earlier
150 /// drafts carry decoded objects). Match on the variant to inspect
151 /// the draft-specific event.
152 #[non_exhaustive]
153 #[derive(Debug, Clone)]
154 pub enum AnyClientEvent {
155 $(
156 #[cfg(feature = $feat)]
157 #[doc = concat!("A draft-", $feat, " event.")]
158 $variant(crate::$module::event::ClientEvent),
159 )+
160 }
161
162 impl AnyClientEvent {
163 /// Returns the draft version this event belongs to.
164 #[allow(unreachable_code)]
165 pub fn draft(&self) -> DraftVersion {
166 match self {
167 $(
168 #[cfg(feature = $feat)]
169 Self::$variant(_) => DraftVersion::$variant,
170 )+
171 #[allow(unreachable_patterns)]
172 _ => unreachable!("AnyClientEvent has no enabled variants"),
173 }
174 }
175
176 /// This event as a [`ControlFrame`], or `None` if it is not a
177 /// control message.
178 ///
179 /// The wildcard arm is over the *other* event variants — a stream
180 /// opening, an object arriving — and not over drafts, so it stays
181 /// reachable in every build and says nothing about which drafts are
182 /// enabled.
183 ///
184 /// `stream_id` is deliberately not carried. Drafts 07 through 15
185 /// have no such field on this event, request streams arriving with
186 /// draft-16, so one arm cannot read it from every draft — and a
187 /// second accessor split across two draft lists is a cost to pay
188 /// when something needs the correlation, not before.
189 pub fn control_frame(&self) -> Option<ControlFrame<'_>> {
190 match self {
191 $(
192 #[cfg(feature = $feat)]
193 Self::$variant(crate::$module::event::ClientEvent::ControlMessage {
194 direction,
195 message,
196 raw,
197 ..
198 }) => Some(ControlFrame {
199 draft: DraftVersion::$variant,
200 outbound: matches!(
201 direction,
202 crate::$module::event::Direction::Send
203 ),
204 message,
205 raw: raw.as_deref(),
206 }),
207 )+
208 _ => None,
209 }
210 }
211 }
212
213 // The classification, written once and instantiated per draft.
214 //
215 // Per draft rather than generic because the two facts worth keeping are
216 // both the draft's: `ConnectionError` is a distinct type on each, and
217 // `close` comes from `codec_session_error_code`, which is the draft's
218 // own reading of its own text. A blanket impl could reach neither.
219 $(
220 #[cfg(feature = $feat)]
221 impl From<crate::$module::connection::ConnectionError> for AnyConnectionError {
222 fn from(err: crate::$module::connection::ConnectionError) -> Self {
223 use crate::above_codec_rules::{DraftSpecificCause, EndpointFault};
224 use crate::$module::connection::{Connection, ConnectionError};
225 use crate::transport::TransportError;
226
227 let message = err.to_string();
228 // Read before the match below moves `err`. Answers `None`
229 // for every variant that match names, so the two tables
230 // partition this draft's error type between them.
231 let own = Connection::draft_specific_cause(&err);
232 let cause = match err {
233 ConnectionError::Codec(error) => ErrorCause::Codec {
234 close: Connection::codec_session_error_code(&error)
235 .map(|code| code.as_u64()),
236 error,
237 },
238 // A varint that would not decode is a decode failure
239 // like any other, and every draft's `CodecError` has a
240 // variant that says so — so it is reported as the codec
241 // error it is rather than as a class of its own.
242 ConnectionError::VarInt(e) => {
243 let error = moqtap_codec::error::CodecError::VarInt(e);
244 ErrorCause::Codec {
245 close: Connection::codec_session_error_code(&error)
246 .map(|code| code.as_u64()),
247 error,
248 }
249 }
250 // The endpoint's error type holds both findings at
251 // once. `fault` says which end,
252 // `session_error_code` says what the draft requires be
253 // done about it, and the pair is read here rather than
254 // guessed at from either half alone.
255 ConnectionError::Endpoint(e) => match e.fault() {
256 EndpointFault::Peer(rule) => ErrorCause::PeerViolation {
257 rule,
258 close: e.session_error_code().map(|code| code.as_u64()),
259 },
260 EndpointFault::ThisEndpoint | EndpointFault::EitherEnd => {
261 ErrorCause::Endpoint
262 }
263 },
264 ConnectionError::Transport(TransportError::StreamReset(code)) => {
265 ErrorCause::StreamReset(code)
266 }
267 ConnectionError::Transport(TransportError::Stopped(code)) => {
268 ErrorCause::Stopped(code)
269 }
270 ConnectionError::Transport(TransportError::SessionClosed {
271 code, ..
272 }) => ErrorCause::SessionClosed(code),
273 ConnectionError::Transport(TransportError::StreamClosed) => {
274 ErrorCause::StreamEnded
275 }
276 ConnectionError::Transport(_) => ErrorCause::Transport,
277 ConnectionError::UnexpectedEnd | ConnectionError::StreamFinished => {
278 ErrorCause::StreamEnded
279 }
280 // Nothing was written: no control stream to write on, an
281 // address that is not one — or a socket this machine
282 // would not open, which `From<DialError>` folds in here
283 // for exactly this reason — a TLS config this build will
284 // not build, an object asked for before the header it is
285 // framed against. `DataStreamState` is the caller
286 // reaching for a data stream out of order, which is this
287 // side's mistake.
288 ConnectionError::NoControlStream
289 | ConnectionError::InvalidAddress(_)
290 | ConnectionError::TlsConfig(_)
291 | ConnectionError::DataStreamState(_) => ErrorCause::Facade,
292 // Whatever this draft adds of its own, read by the draft
293 // rather than guessed at here.
294 #[allow(unreachable_patterns)]
295 _ => match own {
296 Some(DraftSpecificCause::LocalRefusal) => ErrorCause::Facade,
297 Some(DraftSpecificCause::PeerViolation { rule, close }) => {
298 ErrorCause::PeerViolation { rule, close }
299 }
300 Some(DraftSpecificCause::Codec { error, close }) => {
301 ErrorCause::Codec { error, close }
302 }
303 None => ErrorCause::Unclassified,
304 },
305 };
306 AnyConnectionError { message, cause }
307 }
308 }
309 )+
310
311 // Per-draft adapter modules. Each holds an `Adapter` struct that
312 // implements the draft's `ConnectionObserver` trait by forwarding to
313 // an `AnyConnectionObserver`.
314 $(
315 #[cfg(feature = $feat)]
316 #[allow(non_snake_case)]
317 mod $variant {
318 use super::{AnyClientEvent, AnyConnectionObserver};
319 use std::sync::Arc;
320
321 pub(super) struct Adapter(pub(super) Arc<dyn AnyConnectionObserver>);
322
323 impl crate::$module::observer::ConnectionObserver for Adapter {
324 fn on_event(&self, event: &crate::$module::event::ClientEvent) {
325 self.0.on_event(&AnyClientEvent::$variant(event.clone()));
326 }
327
328 fn on_event_owned(&self, event: crate::$module::event::ClientEvent) {
329 self.0.on_event(&AnyClientEvent::$variant(event));
330 }
331 }
332 }
333 )+
334 };
335}
336
337dispatch_all! {
338 #[cfg(feature = "draft07")]
339 Draft07 => draft07,
340 #[cfg(feature = "draft08")]
341 Draft08 => draft08,
342 #[cfg(feature = "draft09")]
343 Draft09 => draft09,
344 #[cfg(feature = "draft10")]
345 Draft10 => draft10,
346 #[cfg(feature = "draft11")]
347 Draft11 => draft11,
348 #[cfg(feature = "draft12")]
349 Draft12 => draft12,
350 #[cfg(feature = "draft13")]
351 Draft13 => draft13,
352 #[cfg(feature = "draft14")]
353 Draft14 => draft14,
354 #[cfg(feature = "draft15")]
355 Draft15 => draft15,
356 #[cfg(feature = "draft16")]
357 Draft16 => draft16,
358 #[cfg(feature = "draft17")]
359 Draft17 => draft17,
360 #[cfg(feature = "draft18")]
361 Draft18 => draft18,
362 #[cfg(feature = "draft19")]
363 Draft19 => draft19,
364 #[cfg(feature = "draft20")]
365 Draft20 => draft20,
366 #[cfg(feature = "draft21")]
367 Draft21 => draft21,
368 #[cfg(feature = "draft22")]
369 Draft22 => draft22,
370 #[cfg(feature = "draft23")]
371 Draft23 => draft23,
372}
373
374/// Draft-agnostic transport choice for [`AnyConnection::connect`].
375#[derive(Debug, Clone)]
376pub enum AnyTransportType {
377 /// Raw QUIC via quinn. The `addr` passed to `connect` should be `host:port`.
378 Quic,
379 /// WebTransport via wtransport. The `url` is the WebTransport endpoint.
380 WebTransport {
381 /// The WebTransport endpoint URL (e.g., `https://host:port/path`).
382 url: String,
383 },
384}
385
386/// Draft-agnostic client configuration. The exact per-draft `ClientConfig`
387/// is constructed internally by [`AnyConnection::connect`] based on `draft`.
388///
389/// Fields that aren't meaningful for the selected draft are ignored:
390/// `additional_versions` is not carried by drafts 15–22 (single-version
391/// setup) and drafts 07–14 always offer their own draft first.
392#[derive(Debug, Clone)]
393pub struct AnyClientConfig {
394 /// Primary draft version for the connection.
395 pub draft: DraftVersion,
396 /// Additional draft versions to offer in CLIENT_SETUP.
397 pub additional_versions: Vec<DraftVersion>,
398 /// Transport type (QUIC or WebTransport).
399 pub transport: AnyTransportType,
400 /// Whether to skip TLS certificate verification (for testing).
401 pub skip_cert_verification: bool,
402 /// Custom CA certificates to trust (DER-encoded).
403 pub ca_certs: Vec<Vec<u8>>,
404 /// Setup parameters to include in CLIENT_SETUP (e.g., auth tokens).
405 pub setup_parameters: Vec<KeyValuePair>,
406}
407
408/// Why a call through this facade stopped, kept as a value rather than as
409/// words.
410///
411/// # What flattening cost, and what it bought
412///
413/// The drafts each state their own `ConnectionError`, and this facade
414/// exists so that a caller never has to branch on which. Rendering every draft
415/// through `Display` bought exactly that — at the price of the *variant*, which
416/// is the half a caller most often needs. Three questions could not be asked of
417/// a sentence:
418///
419/// - **Did the peer reset this stream, or finish it?** A subgroup ends when its
420/// stream ends, so every reader's last read fails; whether it failed because
421/// the peer abandoned the stream or because there was nothing more to send is
422/// the difference between two entirely different findings, and both arrived
423/// spelled as prose.
424/// - **Which rule stopped a decode?** A decoder built for the negotiated draft
425/// refusing a frame is sometimes this build failing to keep up and sometimes
426/// this build doing what the draft *requires*. [`Self::Codec`] carries the
427/// error itself and the code the draft answers it with, so the two are
428/// separable without reading the message.
429/// - **Was it the peer's fault at all?** [`AnyConnectionError::is_local`].
430///
431/// # The ten every draft shares, and the ones only some do
432///
433/// The first ten variants of `ConnectionError` are identical across all
434/// drafts — `Endpoint`, `Codec`, `Transport`, `VarInt`,
435/// `NoControlStream`, `UnexpectedEnd`, `StreamFinished`, `InvalidAddress`,
436/// `TlsConfig`, `DataStreamState` — and those are classified here, once, rather
437/// than once per draft.
438///
439/// Six of the drafts add variants of their own, and those are read by
440/// `Connection::draft_specific_cause` on each draft, beside the doc comment
441/// quoting the sentence it enforces. They divide into this endpoint refusing to
442/// write something ([`Self::Facade`]) and a peer breaking a rule the decoder
443/// could not see ([`Self::PeerViolation`]) — two opposite findings that reached
444/// a caller as prose and read exactly alike. See
445/// [`crate::above_codec_rules`].
446///
447/// # The same split, one layer in
448///
449/// `ConnectionError::Endpoint` wraps one more type per draft — that draft's
450/// `EndpointError`, twenty-six to forty-one variants holding the same two
451/// findings under one name. `EndpointError::fault` divides it the same way and
452/// by the same rule: a variant raised while **reading** what the peer sent is
453/// the peer's, and one raised while **writing**, or refusing to, is this
454/// side's. A handful are raised on both paths and the variant cannot say which;
455/// those stay on this side, where they already were.
456#[derive(Debug, Clone, PartialEq, Eq)]
457pub enum ErrorCause {
458 /// A decoder built for the negotiated draft refused what arrived.
459 ///
460 /// `close` is the session error code **that draft's own text** requires be
461 /// sent for this error, where it names one — which is the whole of what
462 /// separates a relay's defect from this build's shortfall. A decoder
463 /// stopping is not by itself a finding about the peer; a decoder stopping on
464 /// a rule the draft answers with "MUST close the session" is.
465 ///
466 /// `None` covers both a rule the draft states without a close and an error
467 /// the draft says nothing about, and deliberately does not distinguish
468 /// them: neither is grounds to name a relay.
469 Codec {
470 /// The decoder's own error, unflattened.
471 error: moqtap_codec::error::CodecError,
472 /// The code the negotiated draft answers it with, where it names one.
473 close: Option<u64>,
474 },
475 /// A message that did not fit the session's state, and the fault is not the
476 /// peer's.
477 ///
478 /// Two things reach here, and `EndpointError::fault` tells them apart on
479 /// each draft. Most are this endpoint refusing on the way out — a response
480 /// offered for a request the draft says to refuse, an alias it was asked to
481 /// give to a second track, a session that has closed — where nothing
482 /// reached the wire. The rest are variants raised on **both** a receive path
483 /// and a send path, which the variant alone cannot tell apart: the state
484 /// machines, which render as *invalid transition from X on event Y*
485 /// whichever end asked for the transition, and the unknown-request errors.
486 ///
487 /// Those land here because that is the safe direction: a failure that has
488 /// not been told apart is not evidence against a relay. The peer's half
489 /// does not land here at all — it is [`Self::PeerViolation`], which
490 /// [`AnyConnectionError::is_local`] answers false for.
491 ///
492 /// The draft's `EndpointError` is not carried through because it is one of
493 /// one per draft, unrelated, with no shared spine; the message has it.
494 Endpoint,
495 /// The peer abandoned this stream by resetting it, with the application
496 /// error code it named.
497 ///
498 /// Distinct from [`Self::StreamEnded`] and the distinction is the point: a
499 /// reset says the peer stopped on purpose, and several of the drafts' rules
500 /// turn on exactly that. Section 2.2 forbids one Subgroup's Objects on
501 /// different streams "unless one of the streams was reset prematurely" —
502 /// a sentence no caller can apply without being able to see a reset.
503 StreamReset(u64),
504 /// The peer stopped reading this stream (`STOP_SENDING`), with its code.
505 Stopped(u64),
506 /// The peer closed the whole session, with the code it named.
507 ///
508 /// Over MoQT that code is a draft's own session error, which is the
509 /// sharpest thing a refusal says.
510 SessionClosed(u64),
511 /// The stream ended and the peer did not reset it.
512 ///
513 /// Every way a stream can run out short of a reset: a clean FIN with a read
514 /// still wanting bytes, a truncation, a closed stream. They are together
515 /// because most drafts' readers do not tell them apart — their
516 /// `ConnectionError` raises `UnexpectedEnd` for the first two alike — and
517 /// putting a name on a distinction that is not observable would invent it.
518 /// A draft whose reader does tell a truncation apart reports it as a
519 /// [`Self::PeerViolation`].
520 StreamEnded,
521 /// A transport error naming none of the above — a lost connection, a write
522 /// that failed, a datagram that would not send.
523 Transport,
524 /// This facade refused the call itself. Nothing was written and nothing
525 /// reached the wire.
526 ///
527 /// A value that will not fit the field the draft puts it in, a request
528 /// handle from a different draft than the connection, a draft whose feature
529 /// this build was compiled without, an object asked for before the header it
530 /// is framed against, a message handed to the control stream that belongs on
531 /// a request stream of its own. Local by construction, which is why
532 /// [`AnyConnectionError::is_local`] counts it: without a value saying so, a
533 /// facade refusal carries no prefix and reads to a caller exactly like a
534 /// relay hanging up.
535 Facade,
536 /// The peer broke a rule this endpoint enforces **above** its decoder.
537 ///
538 /// The frame read perfectly well and is forbidden anyway, and the fact that
539 /// forbids it is one of two kinds. Some are a comparison inside the frame:
540 /// properties on an Object whose status permits none, a payload after a
541 /// datagram header that permits none, a bidirectional stream opened with a
542 /// message type the draft does not let one open with. The rest are a
543 /// comparison against the session — a second GOAWAY, a Request ID out of
544 /// the peer's own sequence, a Track Alias already naming another track, an
545 /// Object past the one the track ended at.
546 ///
547 /// A decoder can see none of it, so none of it ever reaches [`Self::Codec`]
548 /// and a caller reading only that would find the peer blameless. Without
549 /// this variant the second group arrives as [`Self::Endpoint`], which
550 /// [`AnyConnectionError::is_local`] counts as **this** side's fault, so a
551 /// relay breaking one of these rules is filed against this build's own
552 /// state machine.
553 ///
554 /// `close` has exactly [`Self::Codec`]'s contract — the code **this draft's
555 /// own text** names for the rule, or `None` where it states the rule and
556 /// attaches no consequence. It comes from the draft's own
557 /// `EndpointError::session_error_code` or
558 /// `Connection::codec_session_error_code`, so the rule and its consequence
559 /// are never two readings of one sentence.
560 PeerViolation {
561 /// Which rule, named the same way on every draft that states it.
562 rule: crate::above_codec_rules::AboveCodecRule,
563 /// The code the negotiated draft answers it with, where it names one.
564 close: Option<u64>,
565 },
566 /// A variant neither this facade nor its draft has classified.
567 ///
568 /// Reachable only if the two tables disagree: a variant the match above does
569 /// not name, and that the draft's own `draft_specific_cause` answered `None`
570 /// for. Both are exhaustive today — neither has a wildcard arm — so adding a
571 /// variant to a draft's `ConnectionError` is a compile error in that draft's
572 /// file rather than a silent arrival here.
573 ///
574 /// Kept because the alternative in that arm is a panic, and a facade that
575 /// panics on an error is worse than one that declines to characterise it.
576 /// It answers [`AnyConnectionError::is_local`] false, which is the safe
577 /// direction: an unclassified failure is not evidence about anybody.
578 Unclassified,
579}
580
581/// Error returned by [`AnyConnection::connect`], [`AnyConnection::recv_response`]
582/// and the rest of this facade.
583///
584/// Renders as the draft-specific error it came from, so nothing that read the
585/// message reads anything different. What is new beside it is
586/// [`Self::cause`] — the same failure as a value, so a caller can ask which
587/// kind of failure it was without branching on draft and without parsing the
588/// sentence. See [`ErrorCause`] for why the sentence was not enough.
589#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
590#[error("{message}")]
591pub struct AnyConnectionError {
592 message: String,
593 cause: ErrorCause,
594}
595
596impl AnyConnectionError {
597 /// A refusal by this facade itself: nothing was written, nothing reached
598 /// the wire. [`ErrorCause::Facade`].
599 pub fn facade(message: impl Into<String>) -> Self {
600 Self { message: message.into(), cause: ErrorCause::Facade }
601 }
602
603 /// The draft-specific error's own words, unedited.
604 pub fn message(&self) -> &str {
605 &self.message
606 }
607
608 /// What stopped the call, as a value.
609 pub fn cause(&self) -> &ErrorCause {
610 &self.cause
611 }
612
613 /// Whether the failure was on **this** side rather than the peer's.
614 ///
615 /// True for a decode this build would not complete, a message that did not
616 /// fit the state machine this side drives, and a call this facade refused
617 /// before writing anything. False for every way a peer can end a stream or
618 /// a session — which is what an unimplemented message type or an unwelcome
619 /// parameter looks like from here — and false for
620 /// [`ErrorCause::PeerViolation`], which is the peer's doing by definition.
621 ///
622 /// [`ErrorCause::Endpoint`] answers **true**, and it is a narrow claim: the
623 /// peer's half of the endpoint's error type leaves through
624 /// [`ErrorCause::PeerViolation`] before it gets here. What reaches this
625 /// answer is this endpoint refusing on the way out, plus the variants
626 /// raised on both paths that no table can tell apart. Those answer true
627 /// because that is the safe direction, not because they have been shown to
628 /// be this side's.
629 ///
630 /// [`ErrorCause::Unclassified`] answers false, which is the safe direction:
631 /// a failure that has not been classified is not evidence of anything.
632 pub fn is_local(&self) -> bool {
633 matches!(self.cause, ErrorCause::Codec { .. } | ErrorCause::Endpoint | ErrorCause::Facade)
634 }
635}
636
637/// Where a FETCH's range ends, said once in a way no draft can read two ways.
638///
639/// The drafts do not agree about what a number in an "end object" field means,
640/// and the disagreement is silent: draft-19 Section 10.13 defines `End
641/// Location` as "the last Object, plus 1; or 0 to indicate the entire Group",
642/// while draft-20 Section 5.1.2 and draft-20 Section 10.13 make the
643/// `LOCATION_FILTER` range "inclusive" at both ends and delete both conventions
644/// without a note in the change log. The same `end_object = 10` therefore asks
645/// for objects 0 through 9 on one draft and 0 through 10 on the other, and
646/// nothing on the wire says which was meant.
647///
648/// So this enum, not a number. [`FetchEnd::Object`] is the last Object the
649/// fetch covers and the range **holds** it; [`FetchEnd::EntireGroup`] is
650/// draft-19's `0` spelled out. [`AnyConnection::fetch`] converts to whichever
651/// the negotiated draft writes.
652#[derive(Debug, Clone, Copy, PartialEq, Eq)]
653pub enum FetchEnd {
654 /// Through this Object ID, **inclusive** — it is the last Object the fetch
655 /// covers, and the range holds it.
656 ///
657 /// `Object(0)` is a range ending at Object 0, one object long if it also
658 /// starts there. It is not "the whole group"; that is
659 /// [`FetchEnd::EntireGroup`].
660 Object(u64),
661 /// Through the last Object of the end Group, however many it turns out to
662 /// hold.
663 ///
664 /// Drafts 14-19 write this as `End Location.Object = 0`. Draft-20 writes it
665 /// as a three-field `LOCATION_FILTER`, which Section 5.1.2 defines as
666 /// covering all Objects in the end Group.
667 EntireGroup,
668}
669
670/// The range one [`AnyConnection::fetch`] asks for.
671///
672/// Four numbers on drafts 14 through 19, a `LOCATION_FILTER` parameter on
673/// draft-20, and one meaning here. The end Group is **absolute** on both sides
674/// of that split — draft-20's `EndGroupDelta` is derived from it, not asked for
675/// — and the end Object is [`FetchEnd`], which is the whole point of the type.
676///
677/// # Porting a draft-19 call
678///
679/// A caller that wrote `end_object` as "the last Object plus one" writes
680/// [`FetchEnd::Object`] with the last Object, and one that wrote `0` for a whole
681/// group writes [`FetchEnd::EntireGroup`]. Both send exactly the bytes they sent
682/// before on drafts 14 through 19, and the same request on draft-20.
683#[derive(Debug, Clone, Copy, PartialEq, Eq)]
684pub struct FetchRange {
685 /// The Group the range starts in.
686 pub start_group: u64,
687 /// The Object within `start_group` the range starts at, inclusive.
688 pub start_object: u64,
689 /// The Group the range ends in, absolute. Must not be below
690 /// `start_group`: draft-20 encodes it as an unsigned delta from the start
691 /// and has no way to say "backwards".
692 pub end_group: u64,
693 /// Where the range stops inside `end_group`.
694 pub end: FetchEnd,
695}
696
697impl FetchRange {
698 /// A range ending at `end_object` in `end_group`, **inclusive** — that
699 /// Object is fetched.
700 pub fn through_object(
701 start_group: u64,
702 start_object: u64,
703 end_group: u64,
704 end_object: u64,
705 ) -> Self {
706 Self { start_group, start_object, end_group, end: FetchEnd::Object(end_object) }
707 }
708
709 /// A range covering every Object of `end_group`, however many there are.
710 pub fn through_end_of_group(start_group: u64, start_object: u64, end_group: u64) -> Self {
711 Self { start_group, start_object, end_group, end: FetchEnd::EntireGroup }
712 }
713
714 /// A range holding exactly one Object.
715 ///
716 /// The shape an off-by-one is loudest in: a draft-19 encoder writing
717 /// `end_object = object` rather than `object + 1` asks for nothing at all,
718 /// and one that ports that arithmetic to draft-20 unchanged asks for two.
719 pub fn one_object(group: u64, object: u64) -> Self {
720 Self::through_object(group, object, group, object)
721 }
722
723 /// The `End Location.Object` drafts 14 through 19 carry inline in FETCH.
724 ///
725 /// Their field is "The end Location, plus 1. A Location.Object value of 0
726 /// means the entire group is requested." — draft-19 Section 10.12.1, and
727 /// drafts 14 through 18 in the same words. So [`FetchEnd::Object`] gains
728 /// one here and
729 /// [`FetchEnd::EntireGroup`] is `0`. **This is the only place that `+ 1`
730 /// lives**, which is what keeps it from reaching draft-20.
731 ///
732 /// # Errors
733 ///
734 /// `FetchEnd::Object(u64::MAX)` has no encoding in that field — the plus
735 /// one leaves the number space — and is the one range these drafts cannot
736 /// express that draft-20 can. Refused rather than wrapped to `0`, which
737 /// would silently ask for the whole group.
738 pub fn inline_end_object(&self) -> Result<u64, AnyConnectionError> {
739 match self.end {
740 FetchEnd::EntireGroup => Ok(0),
741 FetchEnd::Object(last) => last.checked_add(1).ok_or_else(|| {
742 AnyConnectionError::facade(format!(
743 "fetch: a range ending at Object {last} cannot be expressed on drafts 14 \
744 through 19, whose End Location.Object is the last Object plus 1"
745 ))
746 }),
747 }
748 }
749
750 /// The draft-20 `LOCATION_FILTER` that carries this range.
751 ///
752 /// Draft-20 Section 10.13 deleted `Start Location` and `End Location` from
753 /// FETCH and moved the range into the parameter, whose ranges Section 5.1.2
754 /// calls inclusive. So [`FetchEnd::Object`] is written **as it stands** —
755 /// nothing here adds one — and [`FetchEnd::EntireGroup`] becomes the
756 /// three-field filter, which Section 5.1.2 defines as covering all Objects
757 /// in the end Group. The end Group travels as `EndGroupDelta`, "delta
758 /// encoded from StartGroup", so it is `end_group - start_group`.
759 ///
760 /// # Errors
761 ///
762 /// [`AnyConnectionError`] when `end_group` is below `start_group`: the
763 /// delta is unsigned and there is no such filter. Drafts 14 through 19 have
764 /// two absolute fields and would put such a range on the wire, where the
765 /// publisher answers it with `INVALID_RANGE`; the difference is where the
766 /// refusal happens, not whether the fetch is legal.
767 /// The name carries the draft because the type does: `LOCATION_FILTER`
768 /// is draft-20's and later's, and each draft's `fill` module declares its
769 /// own `LocationFilter`. An unsuffixed name would be two inherent methods
770 /// of one name the moment a second draft defines the parameter.
771 #[cfg(feature = "draft20")]
772 pub fn location_filter_draft20(
773 &self,
774 ) -> Result<crate::draft20::fill::LocationFilter, AnyConnectionError> {
775 use crate::draft20::fill::LocationFilter;
776
777 let delta = self.end_group.checked_sub(self.start_group).ok_or_else(|| {
778 AnyConnectionError::facade(format!(
779 "fetch: a range from Group {} to Group {} runs backwards, and draft-20 \
780 Section 5.1.2 encodes the end Group as an unsigned delta from the start",
781 self.start_group, self.end_group
782 ))
783 })?;
784 let filter = match self.end {
785 FetchEnd::EntireGroup => {
786 LocationFilter::range(self.start_group, self.start_object, delta)
787 }
788 FetchEnd::Object(last) => {
789 LocationFilter::range_to(self.start_group, self.start_object, delta, last)
790 }
791 };
792 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
793 }
794 /// The draft-21 `LOCATION_FILTER` that carries this range.
795 ///
796 /// Draft-21 Section 9.11 deleted `Start Location` and `End Location` from
797 /// FETCH and moved the range into the parameter, whose ranges Section 3.3.1
798 /// calls inclusive. So [`FetchEnd::Object`] is written **as it stands** —
799 /// nothing here adds one — and [`FetchEnd::EntireGroup`] becomes the
800 /// three-field filter, which Section 9.20.10 defines as covering all Objects
801 /// in the end Group. The end Group travels as `EndGroupDelta`, "delta
802 /// encoded from StartGroup", so it is `end_group - start_group`.
803 ///
804 /// # Errors
805 ///
806 /// [`AnyConnectionError`] when `end_group` is below `start_group`: the
807 /// delta is unsigned and there is no such filter. Drafts 14 through 19 have
808 /// two absolute fields and would put such a range on the wire, where the
809 /// publisher answers it with `INVALID_RANGE`; the difference is where the
810 /// refusal happens, not whether the fetch is legal.
811 /// The name carries the draft because the type does: `LOCATION_FILTER`
812 /// is draft-21's and later's, and each draft's `fill` module declares its
813 /// own `LocationFilter`. An unsuffixed name would be two inherent methods
814 /// of one name the moment a second draft defines the parameter.
815 #[cfg(feature = "draft21")]
816 pub fn location_filter_draft21(
817 &self,
818 ) -> Result<crate::draft21::fill::LocationFilter, AnyConnectionError> {
819 use crate::draft21::fill::LocationFilter;
820
821 let delta = self.end_group.checked_sub(self.start_group).ok_or_else(|| {
822 AnyConnectionError::facade(format!(
823 "fetch: a range from Group {} to Group {} runs backwards, and draft-21 \
824 Section 9.20.10 encodes the end Group as an unsigned delta from the start",
825 self.start_group, self.end_group
826 ))
827 })?;
828 let filter = match self.end {
829 FetchEnd::EntireGroup => {
830 LocationFilter::range(self.start_group, self.start_object, delta)
831 }
832 FetchEnd::Object(last) => {
833 LocationFilter::range_to(self.start_group, self.start_object, delta, last)
834 }
835 };
836 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
837 }
838 /// The draft-22 `LOCATION_FILTER` that carries this range.
839 ///
840 /// Draft-22 Section 9.11 gives FETCH no `Start Location` or `End Location`
841 /// field: the range is the parameter, and draft-22 Section 3.2 has a FETCH
842 /// request Objects "between a Start Location and an End Location,
843 /// inclusive". So [`FetchEnd::Object`] is written **as it stands** —
844 /// Location Filter Type 0x04, Absolute Range — and
845 /// [`FetchEnd::EntireGroup`] is type 0x03, which draft-22 Section 9.20.9
846 /// ends at the "last Object of Group StartGroup + EndGroupDelta". The end
847 /// Group travels as `EndGroupDelta`, "delta encoded from StartGroup", so it
848 /// is `end_group - start_group`.
849 ///
850 /// # Errors
851 ///
852 /// [`AnyConnectionError`] when `end_group` is below `start_group`: the
853 /// delta is unsigned and there is no such filter.
854 /// The name carries the draft because the type does: each draft's `fill`
855 /// module declares its own `LocationFilter`, and draft-22's writes a
856 /// `Location Filter Type` that draft-21's has no field for.
857 #[cfg(feature = "draft22")]
858 pub fn location_filter_draft22(
859 &self,
860 ) -> Result<crate::draft22::fill::LocationFilter, AnyConnectionError> {
861 use crate::draft22::fill::LocationFilter;
862
863 let delta = self.end_group.checked_sub(self.start_group).ok_or_else(|| {
864 AnyConnectionError::facade(format!(
865 "fetch: a range from Group {} to Group {} runs backwards, and draft-22 \
866 Section 9.20.9 encodes the end Group as an unsigned delta from the start",
867 self.start_group, self.end_group
868 ))
869 })?;
870 let filter = match self.end {
871 FetchEnd::EntireGroup => {
872 LocationFilter::range(self.start_group, self.start_object, delta)
873 }
874 FetchEnd::Object(last) => {
875 LocationFilter::range_to(self.start_group, self.start_object, delta, last)
876 }
877 };
878 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
879 }
880 /// The draft-23 `LOCATION_FILTER` that carries this range.
881 ///
882 /// Draft-23 Section 9.11 gives FETCH no `Start Location` or `End Location`
883 /// field: the range is the parameter, and draft-23 Section 3.2 has a FETCH
884 /// request Objects "between a Start Location and an End Location,
885 /// inclusive". So [`FetchEnd::Object`] is written as it stands, Location
886 /// Filter Type 0x04 (Absolute Range), and [`FetchEnd::EntireGroup`] is type
887 /// 0x03, which draft-23 Section 9.19.9 ends at the "last Object of Group
888 /// StartGroup + EndGroupDelta". The end Group travels as `EndGroupDelta`,
889 /// "delta encoded from StartGroup", so it is `end_group - start_group`.
890 /// Both types are absolute, so neither meets draft-23 Section 3.2's "If the
891 /// Location filter is relative (see Section 9.19.9), the publisher MUST
892 /// return REQUEST_ERROR with error code INVALID_RANGE."
893 ///
894 /// # Errors
895 ///
896 /// [`AnyConnectionError`] when `end_group` is below `start_group`: the
897 /// delta is unsigned and there is no such filter.
898 #[cfg(feature = "draft23")]
899 pub fn location_filter_draft23(
900 &self,
901 ) -> Result<crate::draft23::fill::LocationFilter, AnyConnectionError> {
902 use crate::draft23::fill::LocationFilter;
903
904 let delta = self.end_group.checked_sub(self.start_group).ok_or_else(|| {
905 AnyConnectionError::facade(format!(
906 "fetch: a range from Group {} to Group {} runs backwards, and draft-23 \
907 Section 9.19.9 encodes the end Group as an unsigned delta from the start",
908 self.start_group, self.end_group
909 ))
910 })?;
911 let filter = match self.end {
912 FetchEnd::EntireGroup => {
913 LocationFilter::range(self.start_group, self.start_object, delta)
914 }
915 FetchEnd::Object(last) => {
916 LocationFilter::range_to(self.start_group, self.start_object, delta, last)
917 }
918 };
919 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
920 }
921}
922
923/// Where a **Joining** FETCH starts, said once for the two forms the drafts
924/// give it.
925///
926/// A Joining FETCH names no track and no end. Draft-19 Section 10.12.2: "A
927/// Joining Fetch is associated with a Subscribe request by specifying the
928/// Request ID of an active subscription. A publisher receiving a Joining Fetch
929/// uses properties of the associated Subscribe to determine the Track
930/// Namespace, Track Name and End Location such that it is contiguous with the
931/// associated Subscribe." So the only thing a subscriber still has to say is
932/// where the range *begins* — and the field that says it is one varint called
933/// Joining Start, which means two different things depending on a **Fetch
934/// Type** written six bytes earlier.
935///
936/// That is why this is an enum and not a `u64`. Section 10.12.2.1 gives the
937/// publisher two sentences for the same field: for a Relative Joining Fetch it
938/// sets the Start Location to "{Subscribe Largest Location.Group - Joining
939/// Start, 0}", and for an Absolute Joining Fetch it sets the Start Location "to
940/// Joining Start". A caller handing `3` to a single-number entry point would be
941/// asking for three groups of history on one call and for the whole track from
942/// Group 3 on the other, with nothing in the argument to say which was meant.
943///
944/// [`AnyConnection::fetch_joining`] turns the variant into the Fetch Type as
945/// well as the number, so the two cannot come apart.
946#[derive(Debug, Clone, Copy, PartialEq, Eq)]
947pub enum JoiningStart {
948 /// Start this many Groups **before** the joined subscription's Largest
949 /// Location — Fetch Type `0x2`, Relative Joining.
950 ///
951 /// `GroupsBefore(0)` is the subscription's own current Group and nothing
952 /// earlier; `GroupsBefore(1)` adds the Group before it. The publisher
953 /// clamps at zero — "{Subscribe Largest Location.Group - Joining Start,
954 /// 0}" — so an offset larger than the track is long asks for the whole of
955 /// it rather than being an error.
956 ///
957 /// This is the form a subscriber that has **not** been told a Largest
958 /// Location can still use, which is most of them: it is the publisher that
959 /// does the subtraction.
960 GroupsBefore(u64),
961 /// Start at this Group, absolutely — Fetch Type `0x3`, Absolute Joining.
962 ///
963 /// The Start Location is `{Group, 0}`. Drafts 08 through 10 have no such
964 /// Fetch Type — their FETCH offers Standalone and one Joining kind, and
965 /// draft-11 is where the pair arrives — so this variant is refused there
966 /// rather than sent as the relative one, which would ask a completely
967 /// different question.
968 Group(u64),
969}
970
971/// Where a subscription's range stops, said once in a way no draft can read two
972/// ways.
973///
974/// A SUBSCRIBE that names a start location may also name an end, and the
975/// drafts spell that end three different ways. Draft-07 Section 6.4
976/// gives the AbsoluteRange filter an End Group **and** an End Object, with
977/// FETCH's own conventions — "the end Object ID, plus 1. A value of 0 means the
978/// entire group is requested." Draft-08 deleted the End Object and redefined
979/// the End Group as "the end Group ID, inclusive", and drafts 09 through 19
980/// kept it that way. Draft-20 Section 5.1.2 then brought an end Object back, as
981/// the fourth field of a `LOCATION_FILTER` whose range is **inclusive** at both
982/// ends with no plus one anywhere.
983///
984/// One `end_object: u64` at this boundary would therefore have meant the last
985/// Object on four drafts, the last Object plus one on one of them, and nothing
986/// at all on the other twelve — which is [`FetchEnd`]'s problem a second time,
987/// in a place where it is worse: there, every draft could at least carry the
988/// field.
989///
990/// So [`SubscribeEnd::EndOfGroup`] is the end every draft with a range can
991/// express, and [`SubscribeEnd::ThroughObject`] is the one only draft-07 and
992/// drafts 20 through 23 can. The twelve drafts between them **refuse** it rather than
993/// rounding it up to the whole group, because a subscription that quietly
994/// covers more than it asked for is one whose extra objects look like a relay
995/// ignoring the range.
996#[derive(Debug, Clone, Copy, PartialEq, Eq)]
997pub enum SubscribeEnd {
998 /// No end at all — the AbsoluteStart filter. Every draft has it.
999 ///
1000 /// On drafts 20 and 21 this is not expressible from the start location
1001 /// `{0, 0}`; see [`SubscribeRange::location_filter_draft20`] for the collision and
1002 /// for the two ways round it.
1003 Open,
1004 /// Through the whole of this Group, however many Objects it turns out to
1005 /// hold. The Group is **absolute** on every draft, including the ones that
1006 /// put a delta on the wire.
1007 EndOfGroup(u64),
1008 /// Through this Object of this Group, **inclusive** — it is the last Object
1009 /// the subscription covers, and the range holds it.
1010 ///
1011 /// Expressible on draft-07 and drafts 20 through 23, and on nothing between
1012 /// them. See [`SubscribeEnd`] for why those and not the twelve in the middle, and
1013 /// [`SubscribeRange::inline_end_location`] for the plus one draft-07 needs
1014 /// and draft-20 must not have.
1015 ThroughObject {
1016 /// The Group the range ends in, absolute.
1017 end_group: u64,
1018 /// The last Object of `end_group` the range covers.
1019 end_object: u64,
1020 },
1021}
1022
1023/// The range one [`AnyConnection::subscribe_range`] asks for.
1024///
1025/// The start is a plain Location on all the drafts and needs no type; the
1026/// end is [`SubscribeEnd`], which is the whole point. What the drafts do to the
1027/// **end Group** is handled here rather than by the caller: drafts 08 through 16
1028/// write it out in full, drafts 17 through 20 write it as a delta from the start
1029/// Group, and this type takes the absolute value and derives the delta — the
1030/// same split, and the same direction of travel, as [`FetchRange`].
1031///
1032/// # Which filter this is
1033///
1034/// [`SubscribeEnd::Open`] is the AbsoluteStart filter and everything else is
1035/// AbsoluteRange. The Filter Type is never taken as an argument beside the
1036/// fields, because that is the bug this type exists to make unavailable: a
1037/// SUBSCRIBE naming AbsoluteStart with no Start Location beside it is a frame
1038/// whose declared length is short by the fields its own type promised, and the
1039/// publisher reading it runs off the end of the message with nothing local to
1040/// complain. See `AnyConnection::subscribe`, which refuses both filters for
1041/// exactly that reason and points here.
1042#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1043pub struct SubscribeRange {
1044 /// The Group the range starts in.
1045 pub start_group: u64,
1046 /// The Object within `start_group` the range starts at, inclusive.
1047 pub start_object: u64,
1048 /// Where the range stops.
1049 pub end: SubscribeEnd,
1050}
1051
1052impl SubscribeRange {
1053 /// From this Location onward, with no end: the AbsoluteStart filter.
1054 pub fn starting_at(start_group: u64, start_object: u64) -> Self {
1055 Self { start_group, start_object, end: SubscribeEnd::Open }
1056 }
1057
1058 /// From this Location through the whole of `end_group`.
1059 ///
1060 /// `end_group` is absolute and must not be below `start_group` — drafts 17
1061 /// and later encode it as an unsigned delta from the start and have no way
1062 /// to say "backwards".
1063 pub fn through_end_of_group(start_group: u64, start_object: u64, end_group: u64) -> Self {
1064 Self { start_group, start_object, end: SubscribeEnd::EndOfGroup(end_group) }
1065 }
1066
1067 /// From this Location through `end_object` of `end_group`, **inclusive**.
1068 ///
1069 /// Drafts 07 and 20 through 23 only. See [`SubscribeEnd::ThroughObject`].
1070 pub fn through_object(
1071 start_group: u64,
1072 start_object: u64,
1073 end_group: u64,
1074 end_object: u64,
1075 ) -> Self {
1076 Self {
1077 start_group,
1078 start_object,
1079 end: SubscribeEnd::ThroughObject { end_group, end_object },
1080 }
1081 }
1082
1083 /// The whole of one Group, from its first Object.
1084 ///
1085 /// The shape a caller asking "what has this track carried in group N"
1086 /// wants, and the one that is expressible on all the drafts.
1087 pub fn whole_group(group: u64) -> Self {
1088 Self::through_end_of_group(group, 0, group)
1089 }
1090
1091 /// Which of the two filters this range names.
1092 pub fn filter_type(&self) -> moqtap_codec::types::FilterType {
1093 match self.end {
1094 SubscribeEnd::Open => moqtap_codec::types::FilterType::AbsoluteStart,
1095 _ => moqtap_codec::types::FilterType::AbsoluteRange,
1096 }
1097 }
1098
1099 /// The Start Location every draft carrying this range puts on the wire.
1100 pub fn start_location(&self) -> moqtap_codec::types::Location {
1101 moqtap_codec::types::Location {
1102 group: moqtap_codec::varint::VarInt::from_u64_moqt(self.start_group),
1103 object: moqtap_codec::varint::VarInt::from_u64_moqt(self.start_object),
1104 }
1105 }
1106
1107 /// The end Group in full, or `None` for an open-ended range.
1108 pub fn end_group(&self) -> Option<u64> {
1109 match self.end {
1110 SubscribeEnd::Open => None,
1111 SubscribeEnd::EndOfGroup(group)
1112 | SubscribeEnd::ThroughObject { end_group: group, .. } => Some(group),
1113 }
1114 }
1115
1116 /// The `End Location` draft-07 carries inline: the end Group, and the end
1117 /// Object **plus one**, with `0` meaning the whole Group.
1118 ///
1119 /// Draft-07 Section 6.4 gives its AbsoluteRange filter FETCH's two fields
1120 /// and FETCH's two conventions, in the same words, which is why
1121 /// `moqtap_codec::types::check_location_range` checks both. So this is
1122 /// [`FetchRange::inline_end_object`]'s arithmetic on the one draft where a
1123 /// *subscription* needs it, and it lives here rather than being reached for
1124 /// from there so that the two cannot drift into disagreeing about a
1125 /// sentence they share.
1126 ///
1127 /// # Errors
1128 ///
1129 /// [`SubscribeEnd::ThroughObject`] with `end_object` at `u64::MAX`: the plus
1130 /// one leaves the number space, and wrapping it to `0` would silently widen
1131 /// the range to the whole Group. Draft-20 can express that range and
1132 /// draft-07 cannot.
1133 pub fn inline_end_location(
1134 &self,
1135 ) -> Result<Option<moqtap_codec::types::Location>, AnyConnectionError> {
1136 let raw = match self.end {
1137 SubscribeEnd::Open => return Ok(None),
1138 SubscribeEnd::EndOfGroup(group) => (group, 0),
1139 SubscribeEnd::ThroughObject { end_group, end_object } => {
1140 let plus_one = end_object.checked_add(1).ok_or_else(|| {
1141 AnyConnectionError::facade(format!(
1142 "subscribe_range: a range ending at Object {end_object} cannot be \
1143 expressed on draft-07, whose End Location.Object is the last Object plus 1"
1144 ))
1145 })?;
1146 (end_group, plus_one)
1147 }
1148 };
1149 Ok(Some(moqtap_codec::types::Location {
1150 group: moqtap_codec::varint::VarInt::from_u64_moqt(raw.0),
1151 object: moqtap_codec::varint::VarInt::from_u64_moqt(raw.1),
1152 }))
1153 }
1154
1155 /// The end Group drafts 08 through 19 carry, which is a Group and never a
1156 /// Location.
1157 ///
1158 /// # Errors
1159 ///
1160 /// [`SubscribeEnd::ThroughObject`], which those twelve drafts deleted the
1161 /// field for. Refused rather than widened to the whole Group: see
1162 /// [`SubscribeEnd`].
1163 fn group_only_end(&self, draft: DraftVersion) -> Result<Option<u64>, AnyConnectionError> {
1164 match self.end {
1165 SubscribeEnd::Open => Ok(None),
1166 SubscribeEnd::EndOfGroup(group) => Ok(Some(group)),
1167 SubscribeEnd::ThroughObject { .. } => Err(AnyConnectionError::facade(format!(
1168 "subscribe_range: draft-{:02} carries no End Object on SUBSCRIBE — draft-08 \
1169 deleted the field and draft-20 restored it inside LOCATION_FILTER, so a range \
1170 ending inside a group is expressible on draft-07 and drafts 20 through 23 \
1171 alone. Ask for \
1172 the whole group with SubscribeEnd::EndOfGroup",
1173 draft.number()
1174 ))),
1175 }
1176 }
1177
1178 /// The end Group as drafts 17 and later write it: a delta from the start
1179 /// Group.
1180 ///
1181 /// # Errors
1182 ///
1183 /// An `end_group` below `start_group`. The delta is unsigned and there is no
1184 /// such filter; drafts 08 through 16 have an absolute field and would put
1185 /// the range on the wire, where the publisher refuses it. The difference is
1186 /// where the refusal happens, not whether the range is legal.
1187 pub fn end_group_delta(&self) -> Result<Option<u64>, AnyConnectionError> {
1188 let Some(end_group) = self.end_group() else {
1189 return Ok(None);
1190 };
1191 end_group.checked_sub(self.start_group).map(Some).ok_or_else(|| {
1192 AnyConnectionError::facade(format!(
1193 "subscribe_range: a range from Group {} to Group {end_group} runs backwards, and \
1194 drafts 17 and later encode the end Group as an unsigned delta from the start",
1195 self.start_group
1196 ))
1197 })
1198 }
1199
1200 /// This range as the filter parameter drafts 15 through 19 carry it in.
1201 ///
1202 /// `delta` is what those drafts disagree about among themselves: 15 and 16
1203 /// write the End Group out in full, and 17 introduced the delta. The codec's
1204 /// [`SubscriptionFilter`](moqtap_codec::subscription_filter::SubscriptionFilter)
1205 /// refuses a filter whose end is spelled the other draft's way, so the two
1206 /// cannot be mixed up silently here.
1207 ///
1208 /// # Errors
1209 ///
1210 /// [`SubscribeEnd::ThroughObject`], which these five drafts have no field
1211 /// for — the same wall drafts 08 through 14 hit, one layer along. It is
1212 /// worth saying twice because it is not obvious from the shape of the code:
1213 /// the filter these drafts carry has a Start *Location* and an End *Group*,
1214 /// so a range ending inside a group has nowhere to put its Object and would
1215 /// otherwise be written out as the whole group with nothing to say it had
1216 /// been widened.
1217 ///
1218 /// And an `end_group` below `start_group` on the drafts that write a delta.
1219 pub fn subscription_filter(
1220 &self,
1221 draft: DraftVersion,
1222 delta: bool,
1223 ) -> Result<moqtap_codec::subscription_filter::SubscriptionFilter, AnyConnectionError> {
1224 use moqtap_codec::subscription_filter::{FilterEnd, SubscriptionFilter};
1225
1226 let end_group = match (self.group_only_end(draft)?, delta) {
1227 (None, _) => None,
1228 (Some(group), false) => Some(FilterEnd::Group(group)),
1229 (Some(_), true) => {
1230 Some(FilterEnd::GroupDelta(self.end_group_delta()?.unwrap_or_default()))
1231 }
1232 };
1233 Ok(SubscriptionFilter {
1234 filter_type: self.filter_type(),
1235 start_location: Some(self.start_location()),
1236 end_group,
1237 })
1238 }
1239
1240 /// This range as the draft-20 `LOCATION_FILTER` that carries it.
1241 ///
1242 /// Two fields for an open-ended range, three for one through the end of a
1243 /// Group, four for one ending at an Object — Section 5.1.2 selects the shape
1244 /// by how many fields the value holds, so each of the three is a different
1245 /// constructor rather than the same one with values left out.
1246 ///
1247 /// # `{0, 0}` means the opposite here, and is refused
1248 ///
1249 /// On drafts 07 through 19 and on draft-22 an AbsoluteStart at `{0, 0}` is
1250 /// the beginning of the track. Draft-20 Section 5.1.2 gives the two-field filter `{0, 0}` to
1251 /// **Next Object** — `{Largest Object.Group, Largest Object.Object + 1}`,
1252 /// or `{0,0}` where nothing has been delivered — which is the live edge and
1253 /// not the beginning. The identical call would therefore ask every other draft
1254 /// for everything and draft-20 for nothing that has already happened, and
1255 /// nothing on the wire says which was meant.
1256 ///
1257 /// So it is refused, and the error names both ways round it: a range
1258 /// (`through_end_of_group`, which is three fields and unambiguous) for the
1259 /// beginning of the track, and
1260 /// [`LocationFilter::next_object`](crate::draft20::fill::LocationFilter::next_object)
1261 /// through the draft-20 variant for the live edge. This is the only value, on
1262 /// drafts 20 and 21, where the two readings collide: a `{0, 0}` start with an
1263 /// end beside it is three or four fields and means what it says, and any
1264 /// other start location is unambiguous with or without one.
1265 ///
1266 /// # Errors
1267 ///
1268 /// The `{0, 0}` collision above, and an `end_group` below `start_group`, for
1269 /// which see [`Self::end_group_delta`].
1270 /// The name carries the draft because the type does: `LOCATION_FILTER`
1271 /// is draft-20's and later's, and each draft's `fill` module declares its
1272 /// own `LocationFilter`. An unsuffixed name would be two inherent methods
1273 /// of one name the moment a second draft defines the parameter.
1274 #[cfg(feature = "draft20")]
1275 pub fn location_filter_draft20(
1276 &self,
1277 ) -> Result<crate::draft20::fill::LocationFilter, AnyConnectionError> {
1278 use crate::draft20::fill::LocationFilter;
1279
1280 let filter = match self.end {
1281 SubscribeEnd::Open if self.start_group == 0 && self.start_object == 0 => {
1282 return Err(AnyConnectionError::facade(
1283 "subscribe_range: draft-20 Section 5.1.2 reads a two-field LOCATION_FILTER of \
1284 {0, 0} as Next Object — the live edge — where drafts 07 through 19 read an \
1285 AbsoluteStart at {0, 0} as the beginning of the track. For the beginning, \
1286 give the range an end: SubscribeRange::through_end_of_group. For the live \
1287 edge, LocationFilter::next_object through the Draft20 variant",
1288 ));
1289 }
1290 SubscribeEnd::Open => {
1291 Ok(LocationFilter::absolute_start(self.start_group, self.start_object))
1292 }
1293 SubscribeEnd::EndOfGroup(_) => LocationFilter::range(
1294 self.start_group,
1295 self.start_object,
1296 self.end_group_delta()?.unwrap_or_default(),
1297 ),
1298 SubscribeEnd::ThroughObject { end_object, .. } => LocationFilter::range_to(
1299 self.start_group,
1300 self.start_object,
1301 self.end_group_delta()?.unwrap_or_default(),
1302 end_object,
1303 ),
1304 };
1305 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
1306 }
1307 /// This range as the draft-21 `LOCATION_FILTER` that carries it.
1308 ///
1309 /// Two fields for an open-ended range, three for one through the end of a
1310 /// Group, four for one ending at an Object — Section 9.20.10 selects the shape
1311 /// by how many fields the value holds, so each of the three is a different
1312 /// constructor rather than the same one with values left out.
1313 ///
1314 /// # `{0, 0}` means the opposite here, and is refused
1315 ///
1316 /// On drafts 07 through 19 and on draft-22 an AbsoluteStart at `{0, 0}` is
1317 /// the beginning of the track. Draft-21 Section 9.20.10 gives the two-field filter `{0, 0}` to
1318 /// **Next Object** — `{Largest Object.Group, Largest Object.Object + 1}`,
1319 /// or `{0,0}` where nothing has been delivered — which is the live edge and
1320 /// not the beginning. The identical call would therefore ask every other draft
1321 /// for everything and draft-21 for nothing that has already happened, and
1322 /// nothing on the wire says which was meant.
1323 ///
1324 /// So it is refused, and the error names both ways round it: a range
1325 /// (`through_end_of_group`, which is three fields and unambiguous) for the
1326 /// beginning of the track, and
1327 /// [`LocationFilter::next_object`](crate::draft21::fill::LocationFilter::next_object)
1328 /// through the draft-21 variant for the live edge. This is the only value, on
1329 /// drafts 20 and 21, where the two readings collide: a `{0, 0}` start with an
1330 /// end beside it is three or four fields and means what it says, and any
1331 /// other start location is unambiguous with or without one.
1332 ///
1333 /// # Errors
1334 ///
1335 /// The `{0, 0}` collision above, and an `end_group` below `start_group`, for
1336 /// which see [`Self::end_group_delta`].
1337 /// The name carries the draft because the type does: `LOCATION_FILTER`
1338 /// is draft-21's and later's, and each draft's `fill` module declares its
1339 /// own `LocationFilter`. An unsuffixed name would be two inherent methods
1340 /// of one name the moment a second draft defines the parameter.
1341 #[cfg(feature = "draft21")]
1342 pub fn location_filter_draft21(
1343 &self,
1344 ) -> Result<crate::draft21::fill::LocationFilter, AnyConnectionError> {
1345 use crate::draft21::fill::LocationFilter;
1346
1347 let filter = match self.end {
1348 SubscribeEnd::Open if self.start_group == 0 && self.start_object == 0 => {
1349 return Err(AnyConnectionError::facade(
1350 "subscribe_range: draft-21 Section 9.20.10 reads a two-field LOCATION_FILTER of \
1351 {0, 0} as Next Object — the live edge — where drafts 07 through 19 read an \
1352 AbsoluteStart at {0, 0} as the beginning of the track. For the beginning, \
1353 give the range an end: SubscribeRange::through_end_of_group. For the live \
1354 edge, LocationFilter::next_object through the Draft21 variant",
1355 ));
1356 }
1357 SubscribeEnd::Open => {
1358 Ok(LocationFilter::absolute_start(self.start_group, self.start_object))
1359 }
1360 SubscribeEnd::EndOfGroup(_) => LocationFilter::range(
1361 self.start_group,
1362 self.start_object,
1363 self.end_group_delta()?.unwrap_or_default(),
1364 ),
1365 SubscribeEnd::ThroughObject { end_object, .. } => LocationFilter::range_to(
1366 self.start_group,
1367 self.start_object,
1368 self.end_group_delta()?.unwrap_or_default(),
1369 end_object,
1370 ),
1371 };
1372 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
1373 }
1374 /// This range as the draft-22 `LOCATION_FILTER` that carries it.
1375 ///
1376 /// Location Filter Type 0x02 for an open-ended range, 0x03 for one through
1377 /// the end of a Group, 0x04 for one ending at an Object — Section 9.20.9
1378 /// names the shape by its type, so each is a different constructor.
1379 ///
1380 /// # `{0, 0}` means what it says here
1381 ///
1382 /// An open range from `{0, 0}` is type 0x02 with both fields zero, which
1383 /// draft-22 reads as an absolute start at the beginning of the track, as
1384 /// an AbsoluteStart at `{0, 0}` was read before the field count chose the
1385 /// shape. The live edge is a type of its own, "If Location Filter Type is
1386 /// 0x05, no fields follow and it specifies the Next Object", so nothing on
1387 /// this draft collides with the beginning of the track and nothing is
1388 /// refused for it.
1389 /// [`LocationFilter::next_object`](crate::draft22::fill::LocationFilter::next_object)
1390 /// through the draft-22 variant is the live edge.
1391 ///
1392 /// # Errors
1393 ///
1394 /// An `end_group` below `start_group`, for which see
1395 /// [`Self::end_group_delta`].
1396 /// The name carries the draft because the type does: each draft's `fill`
1397 /// module declares its own `LocationFilter`, and draft-22's writes a
1398 /// `Location Filter Type` that draft-21's has no field for.
1399 #[cfg(feature = "draft22")]
1400 pub fn location_filter_draft22(
1401 &self,
1402 ) -> Result<crate::draft22::fill::LocationFilter, AnyConnectionError> {
1403 use crate::draft22::fill::LocationFilter;
1404
1405 let filter = match self.end {
1406 SubscribeEnd::Open => {
1407 Ok(LocationFilter::absolute_start(self.start_group, self.start_object))
1408 }
1409 SubscribeEnd::EndOfGroup(_) => LocationFilter::range(
1410 self.start_group,
1411 self.start_object,
1412 self.end_group_delta()?.unwrap_or_default(),
1413 ),
1414 SubscribeEnd::ThroughObject { end_object, .. } => LocationFilter::range_to(
1415 self.start_group,
1416 self.start_object,
1417 self.end_group_delta()?.unwrap_or_default(),
1418 end_object,
1419 ),
1420 };
1421 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
1422 }
1423 /// This range as the draft-23 `LOCATION_FILTER` that carries it.
1424 ///
1425 /// Location Filter Type 0x02 for an open-ended range, 0x03 for one through
1426 /// the end of a Group, 0x04 for one ending at an Object (draft-23 Section
1427 /// 9.19.9). An open range from `{0, 0}` is type 0x02 with both fields zero,
1428 /// the beginning of the track; the live edge is type 0x05, "If Location
1429 /// Filter Type is 0x05, no fields follow and it specifies the Next Object",
1430 /// built by
1431 /// [`LocationFilter::next_object`](crate::draft23::fill::LocationFilter::next_object).
1432 ///
1433 /// # Errors
1434 ///
1435 /// An `end_group` below `start_group`, for which see
1436 /// [`Self::end_group_delta`].
1437 #[cfg(feature = "draft23")]
1438 pub fn location_filter_draft23(
1439 &self,
1440 ) -> Result<crate::draft23::fill::LocationFilter, AnyConnectionError> {
1441 use crate::draft23::fill::LocationFilter;
1442
1443 let filter = match self.end {
1444 SubscribeEnd::Open => {
1445 Ok(LocationFilter::absolute_start(self.start_group, self.start_object))
1446 }
1447 SubscribeEnd::EndOfGroup(_) => LocationFilter::range(
1448 self.start_group,
1449 self.start_object,
1450 self.end_group_delta()?.unwrap_or_default(),
1451 ),
1452 SubscribeEnd::ThroughObject { end_object, .. } => LocationFilter::range_to(
1453 self.start_group,
1454 self.start_object,
1455 self.end_group_delta()?.unwrap_or_default(),
1456 end_object,
1457 ),
1458 };
1459 filter.map_err(|e| AnyConnectionError::facade(e.to_string()))
1460 }
1461}
1462
1463/// A request made through [`AnyConnection`], in whichever form the negotiated
1464/// draft carries it.
1465///
1466/// Drafts 07-15 put every request on the single bidirectional control stream,
1467/// so all a requester keeps is the request ID it was allocated. Draft-16 moved
1468/// one request off it: Section 3.3 there names "two uses of bidirectional
1469/// streams, the control stream, which begins with CLIENT_SETUP, and
1470/// SUBSCRIBE_NAMESPACE", so a namespace subscription on that draft owns a
1471/// stream and everything else does not. From draft-17 on every request does,
1472/// and the stream *is* the correlation — responses on those drafts carry no
1473/// request ID at all. The variants reflect that split rather than hiding it.
1474///
1475/// Hold on to this value for as long as the request is live. Dropping a
1476/// per-request variant resets its stream, which the peer reads as a
1477/// cancellation; dropping [`AnyRequest::ControlPlane`] does nothing, because
1478/// there is no stream to reset.
1479///
1480/// The per-request variants are large — a `RequestStream` holds both halves of
1481/// a QUIC stream — and `ControlPlane` is two small fields. That disparity is
1482/// only visible when a single draft from 17 on is the only one enabled; with
1483/// more than one, the largest variants are the same size as each other.
1484/// Boxing them to close it would put an allocation on every request of every
1485/// draft to flatter a build no released configuration uses.
1486#[allow(clippy::large_enum_variant)]
1487#[must_use = "dropping a request stream cancels the request"]
1488pub enum AnyRequest {
1489 /// Drafts 07-16: the request was written on the control stream and is
1490 /// identified only by its request ID.
1491 ControlPlane {
1492 /// The request ID the endpoint allocated.
1493 request_id: moqtap_codec::varint::VarInt,
1494 /// The draft that allocated it.
1495 draft: DraftVersion,
1496 },
1497 /// Draft-16: a namespace subscription, the one request on that draft that
1498 /// owns a bidirectional stream. Every other draft-16 request comes back as
1499 /// [`AnyRequest::ControlPlane`].
1500 #[cfg(feature = "draft16")]
1501 Draft16(crate::draft16::connection::NamespaceStream),
1502 /// Draft-17: the request owns a bidirectional stream.
1503 #[cfg(feature = "draft17")]
1504 Draft17(crate::draft17::connection::RequestStream),
1505 /// Draft-18: the request owns a bidirectional stream.
1506 #[cfg(feature = "draft18")]
1507 Draft18(crate::draft18::connection::RequestStream),
1508 /// Draft-19: the request owns a bidirectional stream.
1509 #[cfg(feature = "draft19")]
1510 Draft19(crate::draft19::connection::RequestStream),
1511 /// Draft-20: the request owns a bidirectional stream.
1512 #[cfg(feature = "draft20")]
1513 Draft20(crate::draft20::connection::RequestStream),
1514 /// Draft-21: the request owns a bidirectional stream.
1515 #[cfg(feature = "draft21")]
1516 Draft21(crate::draft21::connection::RequestStream),
1517 /// Draft-22: the request owns a bidirectional stream.
1518 #[cfg(feature = "draft22")]
1519 Draft22(crate::draft22::connection::RequestStream),
1520 /// Draft-23: the request owns a bidirectional stream.
1521 #[cfg(feature = "draft23")]
1522 Draft23(crate::draft23::connection::RequestStream),
1523}
1524
1525impl AnyRequest {
1526 /// The request ID this request was allocated.
1527 pub fn request_id(&self) -> moqtap_codec::varint::VarInt {
1528 match self {
1529 Self::ControlPlane { request_id, .. } => *request_id,
1530 #[cfg(feature = "draft16")]
1531 Self::Draft16(r) => r.request_id(),
1532 #[cfg(feature = "draft17")]
1533 Self::Draft17(r) => r.request_id(),
1534 #[cfg(feature = "draft18")]
1535 Self::Draft18(r) => r.request_id(),
1536 #[cfg(feature = "draft19")]
1537 Self::Draft19(r) => r.request_id(),
1538 #[cfg(feature = "draft20")]
1539 Self::Draft20(r) => r.request_id(),
1540 #[cfg(feature = "draft21")]
1541 Self::Draft21(r) => r.request_id(),
1542 #[cfg(feature = "draft22")]
1543 Self::Draft22(r) => r.request_id(),
1544 #[cfg(feature = "draft23")]
1545 Self::Draft23(r) => r.request_id(),
1546 }
1547 }
1548
1549 /// The draft that carries this request.
1550 pub fn draft(&self) -> DraftVersion {
1551 match self {
1552 Self::ControlPlane { draft, .. } => *draft,
1553 #[cfg(feature = "draft16")]
1554 Self::Draft16(r) => r.draft(),
1555 #[cfg(feature = "draft17")]
1556 Self::Draft17(r) => r.draft(),
1557 #[cfg(feature = "draft18")]
1558 Self::Draft18(r) => r.draft(),
1559 #[cfg(feature = "draft19")]
1560 Self::Draft19(r) => r.draft(),
1561 #[cfg(feature = "draft20")]
1562 Self::Draft20(r) => r.draft(),
1563 #[cfg(feature = "draft21")]
1564 Self::Draft21(r) => r.draft(),
1565 #[cfg(feature = "draft22")]
1566 Self::Draft22(r) => r.draft(),
1567 #[cfg(feature = "draft23")]
1568 Self::Draft23(r) => r.draft(),
1569 }
1570 }
1571
1572 /// Which stream this request owns, or `None` on a draft that carries
1573 /// requests on the shared control stream.
1574 ///
1575 /// This is the observable difference between the two variants: a caller
1576 /// that needs to correlate a response by stream — which is the only
1577 /// correlation drafts 17-19 offer — gets `Some` exactly when the draft
1578 /// provides one.
1579 ///
1580 /// The number is the QUIC stream ID (RFC 9000 Section 2.1), as on
1581 /// [`AnySubgroupWriter::stream_id`].
1582 pub fn stream_id(&self) -> Option<u64> {
1583 match self {
1584 Self::ControlPlane { .. } => None,
1585 #[cfg(feature = "draft16")]
1586 Self::Draft16(r) => Some(r.stream_id()),
1587 #[cfg(feature = "draft17")]
1588 Self::Draft17(r) => Some(r.stream_id()),
1589 #[cfg(feature = "draft18")]
1590 Self::Draft18(r) => Some(r.stream_id()),
1591 #[cfg(feature = "draft19")]
1592 Self::Draft19(r) => Some(r.stream_id()),
1593 #[cfg(feature = "draft20")]
1594 Self::Draft20(r) => Some(r.stream_id()),
1595 #[cfg(feature = "draft21")]
1596 Self::Draft21(r) => Some(r.stream_id()),
1597 #[cfg(feature = "draft22")]
1598 Self::Draft22(r) => Some(r.stream_id()),
1599 #[cfg(feature = "draft23")]
1600 Self::Draft23(r) => Some(r.stream_id()),
1601 }
1602 }
1603
1604 /// Cancel the request by resetting its stream with `code`.
1605 ///
1606 /// Only a request that owns a stream can do this, which is a draft-16
1607 /// namespace subscription and every request from draft-17 on. Cancelling a
1608 /// request that lives on the control stream means sending a message
1609 /// (UNSUBSCRIBE, FETCH_CANCEL, and so on), which needs the connection and
1610 /// is therefore not reachable from the request handle alone. There this
1611 /// refuses rather than silently doing nothing.
1612 #[allow(unused_variables)]
1613 pub fn cancel(&mut self, code: u64) -> Result<(), AnyConnectionError> {
1614 match self {
1615 Self::ControlPlane { draft, .. } => Err(AnyConnectionError::facade(format!(
1616 "cancel: draft {draft:?} carries this request on the control stream, so the \
1617 request handle has no stream to reset; send the draft's own cancellation \
1618 message instead"
1619 ))),
1620 #[cfg(feature = "draft16")]
1621 Self::Draft16(r) => r.cancel(code).map_err(AnyConnectionError::from),
1622 #[cfg(feature = "draft17")]
1623 Self::Draft17(r) => r.cancel(code).map_err(AnyConnectionError::from),
1624 #[cfg(feature = "draft18")]
1625 Self::Draft18(r) => r.cancel(code).map_err(AnyConnectionError::from),
1626 #[cfg(feature = "draft19")]
1627 Self::Draft19(r) => r.cancel(code).map_err(AnyConnectionError::from),
1628 #[cfg(feature = "draft20")]
1629 Self::Draft20(r) => r.cancel(code).map_err(AnyConnectionError::from),
1630 #[cfg(feature = "draft21")]
1631 Self::Draft21(r) => r.cancel(code).map_err(AnyConnectionError::from),
1632 #[cfg(feature = "draft22")]
1633 Self::Draft22(r) => r.cancel(code).map_err(AnyConnectionError::from),
1634 #[cfg(feature = "draft23")]
1635 Self::Draft23(r) => r.cancel(code).map_err(AnyConnectionError::from),
1636 }
1637 }
1638}
1639
1640/// A request the **peer** made, and the handle this side answers it through.
1641///
1642/// The mirror of [`AnyRequest`], and split the same way and for the same
1643/// reason: drafts 07 through 16 carry every request on one shared control
1644/// stream and identify it by its Request ID, and from draft-17 each request
1645/// owns a bidirectional stream that *is* its identity. What differs is who
1646/// allocated the ID — here the peer did, so it is read off the message rather
1647/// than out of this endpoint's own sequence.
1648///
1649/// There is no `Draft16` stream variant, unlike [`AnyRequest`]. Draft-16 gave a
1650/// stream to namespace *subscriptions* alone, and an inbound SUBSCRIBE on that
1651/// draft still arrives on the control stream.
1652///
1653/// Dropping an unanswered request from draft-17 on resets its stream, which the
1654/// peer reads as a refusal. Dropping a [`AnyInboundRequest::ControlPlane`] does
1655/// nothing at all, and the peer is left waiting — refusing there means sending
1656/// the draft's own error message through the variant.
1657#[allow(clippy::large_enum_variant)]
1658#[must_use = "dropping an unanswered request refuses it on the drafts that can"]
1659pub enum AnyInboundRequest {
1660 /// Drafts 07-16: identified only by the Request ID the peer allocated.
1661 ControlPlane {
1662 /// The Request ID the peer put on the message.
1663 request_id: moqtap_codec::varint::VarInt,
1664 /// The draft that carried it.
1665 draft: DraftVersion,
1666 },
1667 /// Draft-17: the request arrived on a bidirectional stream of its own.
1668 #[cfg(feature = "draft17")]
1669 Draft17(crate::draft17::connection::RequestStream),
1670 /// Draft-18: the request arrived on a bidirectional stream of its own.
1671 #[cfg(feature = "draft18")]
1672 Draft18(crate::draft18::connection::RequestStream),
1673 /// Draft-19: the request arrived on a bidirectional stream of its own.
1674 #[cfg(feature = "draft19")]
1675 Draft19(crate::draft19::connection::RequestStream),
1676 /// Draft-20: the request arrived on a bidirectional stream of its own.
1677 #[cfg(feature = "draft20")]
1678 Draft20(crate::draft20::connection::RequestStream),
1679 /// Draft-21: the request arrived on a bidirectional stream of its own.
1680 #[cfg(feature = "draft21")]
1681 Draft21(crate::draft21::connection::RequestStream),
1682 /// Draft-22: the request arrived on a bidirectional stream of its own.
1683 #[cfg(feature = "draft22")]
1684 Draft22(crate::draft22::connection::RequestStream),
1685 /// Draft-23: the request arrived on a bidirectional stream of its own.
1686 #[cfg(feature = "draft23")]
1687 Draft23(crate::draft23::connection::RequestStream),
1688}
1689
1690impl AnyInboundRequest {
1691 /// The Request ID the peer allocated for this request.
1692 pub fn request_id(&self) -> moqtap_codec::varint::VarInt {
1693 match self {
1694 Self::ControlPlane { request_id, .. } => *request_id,
1695 #[cfg(feature = "draft17")]
1696 Self::Draft17(r) => r.request_id(),
1697 #[cfg(feature = "draft18")]
1698 Self::Draft18(r) => r.request_id(),
1699 #[cfg(feature = "draft19")]
1700 Self::Draft19(r) => r.request_id(),
1701 #[cfg(feature = "draft20")]
1702 Self::Draft20(r) => r.request_id(),
1703 #[cfg(feature = "draft21")]
1704 Self::Draft21(r) => r.request_id(),
1705 #[cfg(feature = "draft22")]
1706 Self::Draft22(r) => r.request_id(),
1707 #[cfg(feature = "draft23")]
1708 Self::Draft23(r) => r.request_id(),
1709 }
1710 }
1711
1712 /// The draft that carries this request.
1713 pub fn draft(&self) -> DraftVersion {
1714 match self {
1715 Self::ControlPlane { draft, .. } => *draft,
1716 #[cfg(feature = "draft17")]
1717 Self::Draft17(r) => r.draft(),
1718 #[cfg(feature = "draft18")]
1719 Self::Draft18(r) => r.draft(),
1720 #[cfg(feature = "draft19")]
1721 Self::Draft19(r) => r.draft(),
1722 #[cfg(feature = "draft20")]
1723 Self::Draft20(r) => r.draft(),
1724 #[cfg(feature = "draft21")]
1725 Self::Draft21(r) => r.draft(),
1726 #[cfg(feature = "draft22")]
1727 Self::Draft22(r) => r.draft(),
1728 #[cfg(feature = "draft23")]
1729 Self::Draft23(r) => r.draft(),
1730 }
1731 }
1732
1733 /// The transport stream this request arrived on, or `None` on a draft that
1734 /// carries inbound requests on the shared control stream.
1735 ///
1736 /// The same observable difference [`AnyRequest::stream_id`] exposes, in the
1737 /// other direction.
1738 pub fn stream_id(&self) -> Option<u64> {
1739 match self {
1740 Self::ControlPlane { .. } => None,
1741 #[cfg(feature = "draft17")]
1742 Self::Draft17(r) => Some(r.stream_id()),
1743 #[cfg(feature = "draft18")]
1744 Self::Draft18(r) => Some(r.stream_id()),
1745 #[cfg(feature = "draft19")]
1746 Self::Draft19(r) => Some(r.stream_id()),
1747 #[cfg(feature = "draft20")]
1748 Self::Draft20(r) => Some(r.stream_id()),
1749 #[cfg(feature = "draft21")]
1750 Self::Draft21(r) => Some(r.stream_id()),
1751 #[cfg(feature = "draft22")]
1752 Self::Draft22(r) => Some(r.stream_id()),
1753 #[cfg(feature = "draft23")]
1754 Self::Draft23(r) => Some(r.stream_id()),
1755 }
1756 }
1757}
1758
1759/// What arrived from the peer, and whether this facade has a handle for it.
1760///
1761/// [`AnyConnection::recv_inbound`] returns one of these per message. Everything
1762/// is returned rather than filtered, because on drafts 07 through 16 the same
1763/// stream carries the peer's requests *and* the answers to this side's own —
1764/// a reader that quietly dropped what it was not looking for would swallow a
1765/// SUBSCRIBE_OK somebody was waiting on.
1766///
1767/// The variants differ in size for the reason [`AnyRequest`]'s do — one of them
1768/// carries a request stream and the other does not — and boxing to close the
1769/// gap would put an allocation on every message of every draft to flatter a
1770/// build no released configuration uses.
1771#[allow(clippy::large_enum_variant)]
1772pub enum AnyArrival {
1773 /// The peer subscribed to a track this session publishes.
1774 ///
1775 /// Answer it with [`AnyConnection::accept_subscribe`]. The message is the
1776 /// SUBSCRIBE as it arrived, which is where the track it names lives — and,
1777 /// on drafts 07 through 11, the Track Alias the subscriber chose.
1778 Subscribe {
1779 /// The SUBSCRIBE, decoded by the negotiated draft.
1780 message: moqtap_codec::dispatch::AnyControlMessage,
1781 /// The handle to answer it through.
1782 request: AnyInboundRequest,
1783 },
1784 /// Anything else the peer sent.
1785 ///
1786 /// On drafts 07 through 16 that is any control message which is not a
1787 /// SUBSCRIBE: a response to one of this side's own requests, or something
1788 /// sent unprompted such as MAX_REQUEST_ID. It has already been dispatched
1789 /// into the endpoint by the time it arrives here, so the session's state is
1790 /// correct whether the caller reads it or not.
1791 ///
1792 /// From draft-17 it is a request of a kind this facade has no entry point
1793 /// for, and **it has already been refused**: REQUEST_ERROR with
1794 /// NOT_SUPPORTED and a retry interval of 0, then FIN. There is no handle in
1795 /// this variant to answer it otherwise. That asymmetry is real: on the
1796 /// older drafts this variant is passive, and on the newer ones producing it
1797 /// refuses something.
1798 Other(moqtap_codec::dispatch::AnyControlMessage),
1799}
1800
1801/// One object on a subgroup stream, in the terms every draft shares.
1802///
1803/// The drafts give a subgroup object five different struct shapes, and
1804/// what varies between them is bookkeeping rather than content: whether the
1805/// extension block is counted or measured, whether the status is a field that
1806/// is always present or an `Option`, whether the declared length is stored
1807/// beside the payload or derived from it. None of that is a choice a caller
1808/// makes. What a caller has is an ID and some bytes.
1809///
1810/// The extension block is absent here for a different reason than the rest. It
1811/// is a property of the **stream**, not of the object — the header settles it
1812/// once, and an object with nothing to put in the block still writes a length
1813/// of zero on a stream that carries one, which is why
1814/// [`moqtap_codec::dispatch::AnySubgroupHeader::carries_extension_block`] is
1815/// asked of the header. [`AnyConnection::open_subgroup`] opens streams that
1816/// carry none wherever a draft has a way to say so — eleven of the drafts;
1817/// [`AnyConnection::accept_subgroup`] reads whichever kind the peer opened, and
1818/// hands the header over so a caller can ask.
1819#[derive(Debug, Clone, PartialEq, Eq)]
1820pub struct AnyObject {
1821 /// The Object ID, resolved.
1822 ///
1823 /// Drafts 14 and later encode it as a delta against the object before it on
1824 /// the same stream. The per-draft reader has already undone that, so this
1825 /// is the absolute ID on every draft and a caller never sees the encoding.
1826 pub object_id: u64,
1827 /// The payload. Empty when `status` is `Some`.
1828 pub payload: Vec<u8>,
1829 /// The Object Status wire code, or `None` when the object carried a payload
1830 /// instead.
1831 ///
1832 /// `None` means the same thing on all the drafts, which is why this is
1833 /// an `Option` even though half of them model the status as a field that is
1834 /// always there. Every draft writes the status **only** when the declared
1835 /// payload length is zero, because the status and the payload occupy the
1836 /// same position on the wire: no sequence of bytes states both. So the
1837 /// question "did this object carry a status?" has one answer per object,
1838 /// and it is the answer the wire gives rather than the one a particular
1839 /// draft's struct happens to hold.
1840 pub status: Option<u64>,
1841}
1842
1843/// A subgroup stream this endpoint opened and is writing objects to.
1844///
1845/// Returned by [`AnyConnection::open_subgroup`]. The header is already on the
1846/// wire by the time this exists — opening the stream and framing it are one
1847/// step, because a unidirectional stream with no header on it is not a subgroup
1848/// stream and there is nothing a caller could do with one.
1849///
1850/// Unlike [`AnyRequest`], there is no era split here and no variant that stands
1851/// for "the draft has no stream for this". Every draft from 07 on carries
1852/// subgroup objects on a unidirectional stream of their own, and this is the
1853/// first part of the facade that spans every draft without a footnote. The
1854/// walls that stop the control plane at draft-11 — no ANNOUNCE, no
1855/// TRACK_STATUS_REQUEST, a request ceiling granted by a message this crate
1856/// cannot send — are all about *requests*, and none of them touches a data
1857/// stream.
1858#[allow(clippy::large_enum_variant)]
1859pub enum AnySubgroupWriter {
1860 /// Draft-07's subgroup stream.
1861 #[cfg(feature = "draft07")]
1862 Draft07(crate::draft07::connection::FramedSendStream),
1863 /// Draft-08's subgroup stream.
1864 #[cfg(feature = "draft08")]
1865 Draft08(crate::draft08::connection::FramedSendStream),
1866 /// Draft-09's subgroup stream.
1867 #[cfg(feature = "draft09")]
1868 Draft09(crate::draft09::connection::FramedSendStream),
1869 /// Draft-10's subgroup stream.
1870 #[cfg(feature = "draft10")]
1871 Draft10(crate::draft10::connection::FramedSendStream),
1872 /// Draft-11's subgroup stream.
1873 #[cfg(feature = "draft11")]
1874 Draft11(crate::draft11::connection::FramedSendStream),
1875 /// Draft-12's subgroup stream.
1876 #[cfg(feature = "draft12")]
1877 Draft12(crate::draft12::connection::FramedSendStream),
1878 /// Draft-13's subgroup stream.
1879 #[cfg(feature = "draft13")]
1880 Draft13(crate::draft13::connection::FramedSendStream),
1881 /// Draft-14's subgroup stream.
1882 #[cfg(feature = "draft14")]
1883 Draft14(crate::draft14::connection::FramedSendStream),
1884 /// Draft-15's subgroup stream.
1885 #[cfg(feature = "draft15")]
1886 Draft15(crate::draft15::connection::FramedSendStream),
1887 /// Draft-16's subgroup stream.
1888 #[cfg(feature = "draft16")]
1889 Draft16(crate::draft16::connection::FramedSendStream),
1890 /// Draft-17's subgroup stream.
1891 #[cfg(feature = "draft17")]
1892 Draft17(crate::draft17::connection::FramedSendStream),
1893 /// Draft-18's subgroup stream.
1894 #[cfg(feature = "draft18")]
1895 Draft18(crate::draft18::connection::FramedSendStream),
1896 /// Draft-19's subgroup stream.
1897 #[cfg(feature = "draft19")]
1898 Draft19(crate::draft19::connection::FramedSendStream),
1899 /// Draft-20's subgroup stream.
1900 #[cfg(feature = "draft20")]
1901 Draft20(crate::draft20::connection::FramedSendStream),
1902 /// Draft-21's subgroup stream.
1903 #[cfg(feature = "draft21")]
1904 Draft21(crate::draft21::connection::FramedSendStream),
1905 /// Draft-22's subgroup stream.
1906 #[cfg(feature = "draft22")]
1907 Draft22(crate::draft22::connection::FramedSendStream),
1908 /// Draft-23's subgroup stream.
1909 #[cfg(feature = "draft23")]
1910 Draft23(crate::draft23::connection::FramedSendStream),
1911}
1912
1913/// Expands one body per group of drafts that share a shape, over
1914/// [`AnySubgroupWriter`], [`AnySubgroupReader`] or [`AnyFetchReader`].
1915///
1916/// Every arm is `#[cfg]`-gated on its own draft feature and a catch-all closes
1917/// the match, so a single-draft build compiles with every other draft's arm removed and a
1918/// no-draft build compiles with all of them removed. The same shape
1919/// `moqtap_codec`'s `subgroup_header_accessor!` generates, and for the same
1920/// reason: the alternative is a hand-written arm per draft per method, of which
1921/// at most five differ.
1922///
1923/// Named for data streams rather than for subgroups because a fetch stream is
1924/// the other kind and reads through the same shape. Every variant it is used
1925/// over therefore holds **exactly one** field — which is why
1926/// [`AnyFetchReader::Draft16`] wraps its stream and its resolver in a
1927/// [`Draft16FetchStream`] instead of being a two-field variant.
1928macro_rules! data_stream_dispatch {
1929 (
1930 $enum:ident, $self:expr,
1931 $( [ $( $variant:ident @ $feat:literal ),+ $(,)? ] => |$s:ident, $draft:ident| $body:expr ),+ $(,)?
1932 ) => {
1933 match $self {
1934 $($(
1935 #[cfg(feature = $feat)]
1936 $enum::$variant($s) => {
1937 #[allow(unused_variables)]
1938 let $draft = DraftVersion::$variant;
1939 $body
1940 }
1941 )+)+
1942 #[allow(unreachable_patterns)]
1943 _ => unreachable!("no draft feature is enabled"),
1944 }
1945 };
1946}
1947
1948impl AnySubgroupWriter {
1949 /// The draft framing this stream.
1950 pub fn draft(&self) -> DraftVersion {
1951 data_stream_dispatch! {
1952 AnySubgroupWriter, self,
1953 [
1954 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
1955 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
1956 Draft13 @ "draft13", Draft14 @ "draft14", Draft15 @ "draft15",
1957 Draft16 @ "draft16", Draft17 @ "draft17", Draft18 @ "draft18",
1958 Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
1959 Draft22 @ "draft22",
1960 Draft23 @ "draft23",
1961 ] => |_s, draft| draft,
1962 }
1963 }
1964
1965 /// Which stream this subgroup is being written on.
1966 ///
1967 /// Bare rather than an `Option`, unlike [`AnyRequest::stream_id`]: a
1968 /// subgroup stream *is* a stream on every draft, so there is no era for the
1969 /// option to describe.
1970 ///
1971 /// The number is the QUIC stream ID (RFC 9000 Section 2.1): its low two
1972 /// bits say who opened the stream and which way it runs.
1973 pub fn stream_id(&self) -> u64 {
1974 data_stream_dispatch! {
1975 AnySubgroupWriter, self,
1976 [
1977 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
1978 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
1979 Draft13 @ "draft13", Draft14 @ "draft14", Draft15 @ "draft15",
1980 Draft16 @ "draft16", Draft17 @ "draft17", Draft18 @ "draft18",
1981 Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
1982 Draft22 @ "draft22",
1983 Draft23 @ "draft23",
1984 ] => |s, _draft| s.stream_id(),
1985 }
1986 }
1987
1988 /// Append one object carrying `payload` under `object_id`.
1989 ///
1990 /// # What this fills in, and why none of it is an argument
1991 ///
1992 /// The declared payload length is `payload.len()`, always. Drafts 15 and
1993 /// later store it as a field of the object and refuse an object whose
1994 /// field disagrees with the bytes beside it — the length is already on the
1995 /// wire ahead of the payload, so a disagreement produces a frame no reader
1996 /// can parse and no writer can repair. Drafts 07 through 14 overwrite the
1997 /// field for the same reason. A caller has no way to want a third answer.
1998 ///
1999 /// The status is `Normal` when the payload is empty and absent when it is
2000 /// not, which is the only pair the wire can express: the status field and
2001 /// the payload occupy the same position. An object with bytes and a
2002 /// non-Normal status is asking for two framings at once, and the per-draft
2003 /// writers refuse it. Sending a real status — END_OF_GROUP, END_OF_TRACK —
2004 /// is a different call with different rules about where on the track it may
2005 /// appear, and is reached through the draft's own `Connection` rather than
2006 /// invented here.
2007 ///
2008 /// The extension block is empty, and whether one is written at all was
2009 /// settled by the header [`AnyConnection::open_subgroup`] put on the
2010 /// stream. That is not this method's to change: an object that guessed
2011 /// differently from its header would misframe every object after it.
2012 // `length` is read by every family but draft-14's, which derives it, and
2013 // `empty` by draft-14's and later alone, where the status is an `Option`.
2014 // A single-draft build keeps one arm and so leaves one of them unread.
2015 #[allow(unused_variables)]
2016 pub async fn write_object(
2017 &mut self,
2018 object_id: u64,
2019 payload: &[u8],
2020 ) -> Result<(), AnyConnectionError> {
2021 use moqtap_codec::varint::VarInt;
2022 let id = VarInt::from_u64(object_id)
2023 .map_err(|e| AnyConnectionError::facade(format!("object id {object_id}: {e}")))?;
2024 let length = VarInt::from_usize(payload.len());
2025 let empty = payload.is_empty();
2026 // Five shapes across the drafts, and each one is written once
2027 // as a macro taking the draft's module. A macro rather than a shared
2028 // arm because the stream in hand is a different concrete type in every
2029 // variant: `FramedSendStream` names one struct per draft, not one, so a
2030 // body written once and matched against several variants type-checks
2031 // against none of them.
2032 #[allow(unused_macros)]
2033 macro_rules! bare {
2034 // Draft-07 is the one draft with no extension block anywhere, so
2035 // its object header has no field for one.
2036 ($s:ident, $m:ident) => {{
2037 let object = crate::$m::event::SubgroupObject {
2038 header: moqtap_codec::$m::data_stream::ObjectHeader {
2039 object_id: id,
2040 payload_length: length,
2041 object_status: moqtap_codec::$m::types::ObjectStatus::Normal,
2042 },
2043 payload: payload.to_vec(),
2044 };
2045 $s.write_subgroup_object(&object).await.map_err(AnyConnectionError::from)
2046 }};
2047 }
2048 #[allow(unused_macros)]
2049 macro_rules! counted {
2050 // Draft-08 counts its extensions rather than measuring them.
2051 ($s:ident, $m:ident) => {{
2052 let object = crate::$m::event::SubgroupObject {
2053 header: moqtap_codec::$m::data_stream::ObjectHeader {
2054 object_id: id,
2055 extension_count: VarInt::from_usize(0),
2056 extensions: Vec::new(),
2057 payload_length: length,
2058 object_status: moqtap_codec::$m::types::ObjectStatus::Normal,
2059 },
2060 payload: payload.to_vec(),
2061 };
2062 $s.write_subgroup_object(&object).await.map_err(AnyConnectionError::from)
2063 }};
2064 }
2065 #[allow(unused_macros)]
2066 macro_rules! measured {
2067 // Draft-09 changed that count to a byte length, and every draft
2068 // through 13 kept it that way.
2069 ($s:ident, $m:ident) => {{
2070 let object = crate::$m::event::SubgroupObject {
2071 header: moqtap_codec::$m::data_stream::ObjectHeader {
2072 object_id: id,
2073 extension_headers_length: VarInt::from_usize(0),
2074 extensions: Vec::new(),
2075 payload_length: length,
2076 object_status: moqtap_codec::$m::types::ObjectStatus::Normal,
2077 },
2078 payload: payload.to_vec(),
2079 };
2080 $s.write_subgroup_object(&object).await.map_err(AnyConnectionError::from)
2081 }};
2082 }
2083 #[allow(unused_macros)]
2084 macro_rules! derived {
2085 // Draft-14 takes the declared length from the payload and so has no
2086 // field for it, and is the one draft to name the status field
2087 // `status` rather than `object_status`.
2088 ($s:ident, $m:ident) => {{
2089 let object = moqtap_codec::$m::data_stream::SubgroupObject {
2090 object_id: id,
2091 extension_headers: Vec::new(),
2092 status: empty.then_some(moqtap_codec::$m::types::ObjectStatus::Normal),
2093 payload: payload.to_vec(),
2094 };
2095 $s.write_subgroup_object(&object).await.map_err(AnyConnectionError::from)
2096 }};
2097 }
2098 #[allow(unused_macros)]
2099 macro_rules! declared {
2100 // Drafts 15 and later put the length back and refuse an object
2101 // whose field disagrees with the bytes beside it.
2102 ($s:ident, $m:ident) => {{
2103 let object = moqtap_codec::$m::data_stream::SubgroupObject {
2104 object_id: id,
2105 extension_headers: Vec::new(),
2106 payload_length: length,
2107 object_status: empty.then_some(moqtap_codec::$m::types::ObjectStatus::Normal),
2108 payload: payload.to_vec(),
2109 };
2110 $s.write_subgroup_object(&object).await.map_err(AnyConnectionError::from)
2111 }};
2112 }
2113
2114 match self {
2115 #[cfg(feature = "draft07")]
2116 Self::Draft07(s) => bare!(s, draft07),
2117 #[cfg(feature = "draft08")]
2118 Self::Draft08(s) => counted!(s, draft08),
2119 #[cfg(feature = "draft09")]
2120 Self::Draft09(s) => measured!(s, draft09),
2121 #[cfg(feature = "draft10")]
2122 Self::Draft10(s) => measured!(s, draft10),
2123 #[cfg(feature = "draft11")]
2124 Self::Draft11(s) => measured!(s, draft11),
2125 #[cfg(feature = "draft12")]
2126 Self::Draft12(s) => measured!(s, draft12),
2127 #[cfg(feature = "draft13")]
2128 Self::Draft13(s) => measured!(s, draft13),
2129 #[cfg(feature = "draft14")]
2130 Self::Draft14(s) => derived!(s, draft14),
2131 #[cfg(feature = "draft15")]
2132 Self::Draft15(s) => declared!(s, draft15),
2133 #[cfg(feature = "draft16")]
2134 Self::Draft16(s) => declared!(s, draft16),
2135 #[cfg(feature = "draft17")]
2136 Self::Draft17(s) => declared!(s, draft17),
2137 #[cfg(feature = "draft18")]
2138 Self::Draft18(s) => declared!(s, draft18),
2139 #[cfg(feature = "draft19")]
2140 Self::Draft19(s) => declared!(s, draft19),
2141 #[cfg(feature = "draft20")]
2142 Self::Draft20(s) => declared!(s, draft20),
2143 #[cfg(feature = "draft21")]
2144 Self::Draft21(s) => declared!(s, draft21),
2145 #[cfg(feature = "draft22")]
2146 Self::Draft22(s) => declared!(s, draft22),
2147 #[cfg(feature = "draft23")]
2148 Self::Draft23(s) => declared!(s, draft23),
2149 #[allow(unreachable_patterns)]
2150 _ => Err(AnyConnectionError::facade("no draft feature is enabled")),
2151 }
2152 }
2153
2154 /// Close the stream, which ends the subgroup.
2155 ///
2156 /// A subgroup has no terminator message on any draft: the stream ending
2157 /// *is* the end of it. So this is not a courtesy — a reader on the far side
2158 /// is waiting for either another object or the end, and cannot tell which
2159 /// is coming until one of them arrives.
2160 pub async fn finish(&mut self) -> Result<(), AnyConnectionError> {
2161 data_stream_dispatch! {
2162 AnySubgroupWriter, self,
2163 [
2164 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
2165 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
2166 Draft13 @ "draft13", Draft14 @ "draft14", Draft15 @ "draft15",
2167 Draft16 @ "draft16", Draft17 @ "draft17", Draft18 @ "draft18",
2168 Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
2169 Draft22 @ "draft22",
2170 Draft23 @ "draft23",
2171 ] => |s, _draft| {
2172 s.finish().await.map_err(AnyConnectionError::from)
2173 },
2174 }
2175 }
2176}
2177
2178/// A subgroup stream the **peer** opened, that this endpoint is reading.
2179///
2180/// The mirror of [`AnySubgroupWriter`], returned by
2181/// [`AnyConnection::accept_subgroup`] beside the header the peer framed it
2182/// with.
2183#[allow(clippy::large_enum_variant)]
2184pub enum AnySubgroupReader {
2185 /// Draft-07's subgroup stream.
2186 #[cfg(feature = "draft07")]
2187 Draft07(crate::draft07::connection::FramedRecvStream),
2188 /// Draft-08's subgroup stream.
2189 #[cfg(feature = "draft08")]
2190 Draft08(crate::draft08::connection::FramedRecvStream),
2191 /// Draft-09's subgroup stream.
2192 #[cfg(feature = "draft09")]
2193 Draft09(crate::draft09::connection::FramedRecvStream),
2194 /// Draft-10's subgroup stream.
2195 #[cfg(feature = "draft10")]
2196 Draft10(crate::draft10::connection::FramedRecvStream),
2197 /// Draft-11's subgroup stream.
2198 #[cfg(feature = "draft11")]
2199 Draft11(crate::draft11::connection::FramedRecvStream),
2200 /// Draft-12's subgroup stream.
2201 #[cfg(feature = "draft12")]
2202 Draft12(crate::draft12::connection::FramedRecvStream),
2203 /// Draft-13's subgroup stream.
2204 #[cfg(feature = "draft13")]
2205 Draft13(crate::draft13::connection::FramedRecvStream),
2206 /// Draft-14's subgroup stream.
2207 #[cfg(feature = "draft14")]
2208 Draft14(crate::draft14::connection::FramedRecvStream),
2209 /// Draft-15's subgroup stream.
2210 #[cfg(feature = "draft15")]
2211 Draft15(crate::draft15::connection::FramedRecvStream),
2212 /// Draft-16's subgroup stream.
2213 #[cfg(feature = "draft16")]
2214 Draft16(crate::draft16::connection::FramedRecvStream),
2215 /// Draft-17's subgroup stream.
2216 #[cfg(feature = "draft17")]
2217 Draft17(crate::draft17::connection::FramedRecvStream),
2218 /// Draft-18's subgroup stream.
2219 #[cfg(feature = "draft18")]
2220 Draft18(crate::draft18::connection::FramedRecvStream),
2221 /// Draft-19's subgroup stream.
2222 #[cfg(feature = "draft19")]
2223 Draft19(crate::draft19::connection::FramedRecvStream),
2224 /// Draft-20's subgroup stream.
2225 #[cfg(feature = "draft20")]
2226 Draft20(crate::draft20::connection::FramedRecvStream),
2227 /// Draft-21's subgroup stream.
2228 #[cfg(feature = "draft21")]
2229 Draft21(crate::draft21::connection::FramedRecvStream),
2230 /// Draft-22's subgroup stream.
2231 #[cfg(feature = "draft22")]
2232 Draft22(crate::draft22::connection::FramedRecvStream),
2233 /// Draft-23's subgroup stream.
2234 #[cfg(feature = "draft23")]
2235 Draft23(crate::draft23::connection::FramedRecvStream),
2236}
2237
2238impl AnySubgroupReader {
2239 /// The draft framing this stream.
2240 pub fn draft(&self) -> DraftVersion {
2241 data_stream_dispatch! {
2242 AnySubgroupReader, self,
2243 [
2244 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
2245 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
2246 Draft13 @ "draft13", Draft14 @ "draft14", Draft15 @ "draft15",
2247 Draft16 @ "draft16", Draft17 @ "draft17", Draft18 @ "draft18",
2248 Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
2249 Draft22 @ "draft22",
2250 Draft23 @ "draft23",
2251 ] => |_s, draft| draft,
2252 }
2253 }
2254
2255 /// Which stream this subgroup arrived on, on the terms
2256 /// [`AnySubgroupWriter::stream_id`] describes.
2257 pub fn stream_id(&self) -> u64 {
2258 data_stream_dispatch! {
2259 AnySubgroupReader, self,
2260 [
2261 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
2262 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
2263 Draft13 @ "draft13", Draft14 @ "draft14", Draft15 @ "draft15",
2264 Draft16 @ "draft16", Draft17 @ "draft17", Draft18 @ "draft18",
2265 Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
2266 Draft22 @ "draft22",
2267 Draft23 @ "draft23",
2268 ] => |s, _draft| s.stream_id(),
2269 }
2270 }
2271
2272 /// Read the next object on this subgroup.
2273 ///
2274 /// # The stream's end arrives as an error, not as `None`
2275 ///
2276 /// A subgroup ends when its stream ends. Most per-draft readers cannot
2277 /// tell that end from a truncation: both leave the reader wanting bytes
2278 /// that never come, and both surface as `ConnectionError::UnexpectedEnd`.
2279 /// Returning `Option` here would have to invent the distinction, and
2280 /// inventing it means reporting a stream the peer cut off mid-object as a
2281 /// subgroup that finished — which is exactly the case a caller most needs
2282 /// to know about. A reader that can tell them apart reports the truncation
2283 /// as an [`ErrorCause::PeerViolation`].
2284 ///
2285 /// So a caller that expects a known number of objects reads that many, and
2286 /// one that does not treats the error as the end and keeps whatever it read
2287 /// before it. Neither has to guess.
2288 pub async fn read_object(&mut self) -> Result<AnyObject, AnyConnectionError> {
2289 data_stream_dispatch! {
2290 AnySubgroupReader, self,
2291 // Drafts 07 through 13 model the status as a field that is always
2292 // present. The wire does not: it carries one only when the declared
2293 // length is zero, and the per-draft decoder fills in `Normal` for
2294 // every object it read a payload for. Reporting that filled-in value
2295 // as though the peer had sent it would make `AnyObject::status` mean
2296 // something different on these seven drafts than on the other seven.
2297 [
2298 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
2299 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
2300 Draft13 @ "draft13",
2301 ] => |s, _draft| {
2302 let object = s.read_subgroup_object().await
2303 .map_err(AnyConnectionError::from)?;
2304 Ok(AnyObject {
2305 object_id: object.header.object_id.into_inner(),
2306 status: object
2307 .payload
2308 .is_empty()
2309 .then_some(object.header.object_status as u64),
2310 payload: object.payload,
2311 })
2312 },
2313 [Draft14 @ "draft14"] => |s, _draft| {
2314 let object = s.read_subgroup_object().await
2315 .map_err(AnyConnectionError::from)?;
2316 Ok(AnyObject {
2317 object_id: object.object_id.into_inner(),
2318 status: object.status.map(|s| s as u64),
2319 payload: object.payload,
2320 })
2321 },
2322 [
2323 Draft15 @ "draft15", Draft16 @ "draft16", Draft17 @ "draft17",
2324 Draft18 @ "draft18", Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
2325 Draft22 @ "draft22",
2326 Draft23 @ "draft23",
2327 ] => |s, _draft| {
2328 let object = s.read_subgroup_object().await
2329 .map_err(AnyConnectionError::from)?;
2330 Ok(AnyObject {
2331 object_id: object.object_id.into_inner(),
2332 status: object.object_status.map(|s| s as u64),
2333 payload: object.payload,
2334 })
2335 },
2336 }
2337 }
2338}
2339
2340/// One object on a **fetch** stream, in the terms every draft shares.
2341///
2342/// [`AnyObject`]'s twin, and it needs three fields that one does not, because a
2343/// fetch stream is not a subgroup stream with a different header on it. A
2344/// subgroup stream's header names the Group and the Subgroup once and every
2345/// object on it belongs to them; a fetch stream carries a whole range, so each
2346/// object states its own Location — and from draft-15 it states it by
2347/// *inheritance*, leaving fields off the wire that the object before it
2348/// supplies. Every value here is resolved: what a caller gets is where the
2349/// object is, never what the wire happened to write down.
2350#[derive(Debug, Clone, PartialEq, Eq)]
2351pub struct AnyFetchObject {
2352 /// The Group this object belongs to, resolved.
2353 pub group_id: u64,
2354 /// The Subgroup, resolved, and `None` where the draft lets an object on a
2355 /// fetch stream have none.
2356 ///
2357 /// Drafts 07 through 15 always carry it. From draft-16 an object whose
2358 /// Forwarding Preference is Datagram omits it — draft-16 Section 10.2.1 —
2359 /// and so does an end-of-range marker, which is a span rather than an
2360 /// object.
2361 pub subgroup_id: Option<u64>,
2362 /// The Object ID, resolved. Drafts 15 and later may encode it as a step
2363 /// from the object before it on the stream, and the reader has undone that.
2364 pub object_id: u64,
2365 /// The payload. Empty when `status` is `Some`, and empty on every
2366 /// end-of-range marker.
2367 pub payload: Vec<u8>,
2368 /// The Object Status wire code, on the same terms as [`AnyObject::status`]:
2369 /// `Some` exactly when the object carried no payload, because the two
2370 /// occupy the same position on the wire.
2371 ///
2372 /// **Always `None` from draft-16 on**, and that is the drafts talking
2373 /// rather than this type giving up: draft-16 deleted the Object Status
2374 /// field from a fetch object and put [`Self::end_of_range`] in its place.
2375 /// The pair covers one question between them — *is this an object, or a
2376 /// statement about objects that are not here* — and which of the two answers
2377 /// it depends on the draft.
2378 pub status: Option<u64>,
2379 /// The end-of-range marker's own wire code, where this "object" is one.
2380 ///
2381 /// Drafts 16 through 23 end a fetch that cannot serve part of its range
2382 /// with a marker rather than with silence, and the marker covers the whole
2383 /// span from the last serialized object to this Location inclusive:
2384 /// `0x8C` for a span known not to exist, `0x10C` for one whose status is
2385 /// unknown, `0x20C` (drafts 20 through 23) for one the publisher abandoned,
2386 /// and `0x40C` (draft-23) for one that is Gone.
2387 ///
2388 /// Draft-23 also has `0x80` (End of Group) and `0x81` (End of Track),
2389 /// which cover no span: their Location is the prior Object's, which
2390 /// draft-23 Section 11.4.2.2 makes "the last Object in its Group, or in
2391 /// the Track, respectively".
2392 ///
2393 /// A number rather than an enum for the reason [`crate::dispatch`]'s
2394 /// neighbours are: the code is the fingerprint, and a later draft
2395 /// assigning a fourth marker should widen a report rather than fail to
2396 /// parse. `None` on drafts 07 through 15, which have no such marker — there
2397 /// the same news arrives, when it arrives at all, as a [`Self::status`].
2398 pub end_of_range: Option<u64>,
2399}
2400
2401/// Draft-16's fetch stream, carried with the resolver its objects need.
2402///
2403/// The one variant of [`AnyFetchReader`] that is not a bare stream, and the
2404/// reason is a gap one draft wide. All but one of the per-draft
2405/// `FramedRecvStream`s resolve a fetch object's elided fields themselves —
2406/// drafts 07 through 14 have nothing to resolve, and 15, 17, 18, 19 and 20 each
2407/// hold a `FetchObjectReader` on the stream. Draft-16's does not, though
2408/// `moqtap_codec::draft16::data_stream::FetchObjectReader` exists and does
2409/// exactly the job.
2410///
2411/// So the state lives here. It has to live *somewhere* per stream: a draft-16
2412/// object may leave out its Group ID, its Subgroup ID, its Object ID and its
2413/// Priority, and Table 5 gives each absent field a meaning drawn from the
2414/// object before it. A reader without the object before it does not get a worse
2415/// answer, it gets no answer at all.
2416#[cfg(feature = "draft16")]
2417pub struct Draft16FetchStream {
2418 stream: crate::draft16::connection::FramedRecvStream,
2419 reader: moqtap_codec::draft16::data_stream::FetchObjectReader,
2420}
2421
2422#[cfg(feature = "draft16")]
2423impl Draft16FetchStream {
2424 /// Which stream this fetch arrived on, on the terms
2425 /// [`AnySubgroupWriter::stream_id`] describes.
2426 pub fn stream_id(&self) -> u64 {
2427 self.stream.stream_id()
2428 }
2429}
2430
2431/// The Group Order a FETCH_OK named, for handing to
2432/// [`AnyConnection::accept_fetch`].
2433///
2434/// # Why this is a function and not a field lookup at the call site
2435///
2436/// Because the field moved twice and the answer is silent when it is wrong.
2437/// Drafts 07 through 14 put Group Order on FETCH_OK outright. Drafts 15 and 16
2438/// deleted it from the message. Drafts 17 through 20 brought it back as
2439/// **Track Property `0x22`**, `DEFAULT PUBLISHER GROUP ORDER`, which is
2440/// optional — so on the three drafts where the value decides how every Group ID
2441/// after the first is resolved, the commonest FETCH_OK does not carry it at all.
2442///
2443/// [`moqtap_codec::types::GroupOrder::Ascending`] is the answer in that case, and
2444/// it is the draft's answer rather than this function's: draft-20 Section 10.2.8
2445/// makes Ascending the default for an omitted property.
2446///
2447/// A message that is not a FETCH_OK answers `Ascending` too, and that is not a
2448/// claim about it — nothing else here has a Group Order to report, and a caller
2449/// holding the wrong message has a different problem than this can name.
2450///
2451/// Written against [`moqtap_codec::dispatch::AnyControlMessage::fields`] rather
2452/// than as an arm per draft, so a draft that moves the field again is one entry
2453/// here rather than a match that still compiles.
2454pub fn fetch_group_order(
2455 message: &moqtap_codec::dispatch::AnyControlMessage,
2456) -> moqtap_codec::types::GroupOrder {
2457 use moqtap_codec::fields::FieldValue;
2458 use moqtap_codec::types::GroupOrder;
2459
2460 let named = |code: u64| match code {
2461 0x2 => GroupOrder::Descending,
2462 0x1 => GroupOrder::Ascending,
2463 // Including `0x0`, Publisher — which says the publisher decides and so
2464 // states no direction. Every consumer of this value needs one.
2465 _ => GroupOrder::Ascending,
2466 };
2467
2468 let fields = message.fields();
2469 // Drafts 07 through 14, where it is a field of the message.
2470 if let Some(FieldValue::Uint(code)) = fields.get("group_order") {
2471 return named(*code);
2472 }
2473 // Drafts 17 through 20, where it is one entry of a property list that need
2474 // not contain it.
2475 if let Some(FieldValue::Array(properties)) = fields.get("track_properties") {
2476 for property in properties {
2477 let FieldValue::Map(entry) = property else { continue };
2478 if entry.get("name") != Some(&FieldValue::Text("default_publisher_group_order".into()))
2479 {
2480 continue;
2481 }
2482 if let Some(FieldValue::Uint(code)) = entry.get("value") {
2483 return named(*code);
2484 }
2485 }
2486 }
2487 GroupOrder::Ascending
2488}
2489
2490/// A fetch stream the **peer** opened, that this endpoint is reading.
2491///
2492/// [`AnySubgroupReader`]'s twin, returned by [`AnyConnection::accept_fetch`]
2493/// beside the FETCH_HEADER the peer framed it with.
2494///
2495/// The two are separate types for the reason the per-draft connections keep
2496/// `accept_fetch_stream` and `accept_subgroup_stream` separate: the header
2497/// decides how every object after it is framed, so a caller has to know which
2498/// kind it is expecting before the first byte is read. Nothing here can be
2499/// handed a subgroup stream and cope.
2500#[allow(clippy::large_enum_variant)]
2501pub enum AnyFetchReader {
2502 /// Draft-07's fetch stream.
2503 #[cfg(feature = "draft07")]
2504 Draft07(crate::draft07::connection::FramedRecvStream),
2505 /// Draft-08's fetch stream.
2506 #[cfg(feature = "draft08")]
2507 Draft08(crate::draft08::connection::FramedRecvStream),
2508 /// Draft-09's fetch stream.
2509 #[cfg(feature = "draft09")]
2510 Draft09(crate::draft09::connection::FramedRecvStream),
2511 /// Draft-10's fetch stream.
2512 #[cfg(feature = "draft10")]
2513 Draft10(crate::draft10::connection::FramedRecvStream),
2514 /// Draft-11's fetch stream.
2515 #[cfg(feature = "draft11")]
2516 Draft11(crate::draft11::connection::FramedRecvStream),
2517 /// Draft-12's fetch stream.
2518 #[cfg(feature = "draft12")]
2519 Draft12(crate::draft12::connection::FramedRecvStream),
2520 /// Draft-13's fetch stream.
2521 #[cfg(feature = "draft13")]
2522 Draft13(crate::draft13::connection::FramedRecvStream),
2523 /// Draft-14's fetch stream.
2524 #[cfg(feature = "draft14")]
2525 Draft14(crate::draft14::connection::FramedRecvStream),
2526 /// Draft-15's fetch stream.
2527 #[cfg(feature = "draft15")]
2528 Draft15(crate::draft15::connection::FramedRecvStream),
2529 /// Draft-16's fetch stream, and the resolver its objects need. See
2530 /// [`Draft16FetchStream`].
2531 #[cfg(feature = "draft16")]
2532 Draft16(Draft16FetchStream),
2533 /// Draft-17's fetch stream.
2534 #[cfg(feature = "draft17")]
2535 Draft17(crate::draft17::connection::FramedRecvStream),
2536 /// Draft-18's fetch stream.
2537 #[cfg(feature = "draft18")]
2538 Draft18(crate::draft18::connection::FramedRecvStream),
2539 /// Draft-19's fetch stream.
2540 #[cfg(feature = "draft19")]
2541 Draft19(crate::draft19::connection::FramedRecvStream),
2542 /// Draft-20's fetch stream.
2543 #[cfg(feature = "draft20")]
2544 Draft20(crate::draft20::connection::FramedRecvStream),
2545 /// Draft-21's fetch stream.
2546 #[cfg(feature = "draft21")]
2547 Draft21(crate::draft21::connection::FramedRecvStream),
2548 /// Draft-22's fetch stream.
2549 #[cfg(feature = "draft22")]
2550 Draft22(crate::draft22::connection::FramedRecvStream),
2551 /// Draft-23's fetch stream.
2552 #[cfg(feature = "draft23")]
2553 Draft23(crate::draft23::connection::FramedRecvStream),
2554}
2555
2556impl AnyFetchReader {
2557 /// The draft framing this stream.
2558 pub fn draft(&self) -> DraftVersion {
2559 data_stream_dispatch! {
2560 AnyFetchReader, self,
2561 [
2562 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
2563 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
2564 Draft13 @ "draft13", Draft14 @ "draft14", Draft15 @ "draft15",
2565 Draft16 @ "draft16", Draft17 @ "draft17", Draft18 @ "draft18",
2566 Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
2567 Draft22 @ "draft22",
2568 Draft23 @ "draft23",
2569 ] => |_s, draft| draft,
2570 }
2571 }
2572
2573 /// Which stream this fetch arrived on, on the terms
2574 /// [`AnySubgroupWriter::stream_id`] describes.
2575 pub fn stream_id(&self) -> u64 {
2576 data_stream_dispatch! {
2577 AnyFetchReader, self,
2578 [
2579 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
2580 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
2581 Draft13 @ "draft13", Draft14 @ "draft14", Draft15 @ "draft15",
2582 Draft16 @ "draft16", Draft17 @ "draft17", Draft18 @ "draft18",
2583 Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
2584 Draft22 @ "draft22",
2585 Draft23 @ "draft23",
2586 ] => |s, _draft| s.stream_id(),
2587 }
2588 }
2589
2590 /// Read the next object on this fetch stream.
2591 ///
2592 /// # The stream's end arrives as an error, not as `None`
2593 ///
2594 /// For the reason [`AnySubgroupReader::read_object`] gives, and one more of
2595 /// its own: a fetch stream ends when the range it was serving runs out, and
2596 /// the only thing that says so is the FIN. Returning `Option` would have to
2597 /// tell that FIN from a peer that stopped mid-object, which no per-draft
2598 /// reader can.
2599 ///
2600 /// # Errors
2601 ///
2602 /// Everything [`AnySubgroupReader::read_object`] can raise, and on drafts 15
2603 /// through 20 one more class of its own: an object that inherits from an
2604 /// object that does not exist. The first object of a stream may not leave
2605 /// out its Group ID or its Object ID, and a stream whose first object does
2606 /// is refused here rather than resolved against zero — draft-16 Section
2607 /// 10.4.4.1 makes it a PROTOCOL_VIOLATION, and there is no value to produce
2608 /// even where it does not.
2609 ///
2610 /// On drafts 11-23, the Prior gap conditions the stream shows, refused with
2611 /// `CodecError::MalformedTrack` as each draft's `FetchGaps` refuses them,
2612 /// and a property block that does not parse, refused with
2613 /// `CodecError::MalformedObjectProperties`, which drafts 14-23 make a
2614 /// Malformed Track.
2615 ///
2616 /// This reader does not withdraw from the track over either. The caller
2617 /// answers them with the draft's own `Connection::withdraw_for_fetch_stream`
2618 /// (drafts 12-16) or `Connection::requests_to_cancel_for_fetch_stream`
2619 /// (drafts 17-23), passing `ConnectionError::Codec` of the error that
2620 /// [`ErrorCause::Codec`] carries.
2621 pub async fn read_object(&mut self) -> Result<AnyFetchObject, AnyConnectionError> {
2622 data_stream_dispatch! {
2623 AnyFetchReader, self,
2624 // Drafts 07 through 13 model the status as a field that is always
2625 // present, exactly as their subgroup objects do, and the same rule
2626 // applies for the same reason: the wire carries a status only where
2627 // the declared length is zero, so a filled-in `Normal` beside a
2628 // payload is the decoder talking and not the peer.
2629 [
2630 Draft07 @ "draft07", Draft08 @ "draft08", Draft09 @ "draft09",
2631 Draft10 @ "draft10", Draft11 @ "draft11", Draft12 @ "draft12",
2632 Draft13 @ "draft13",
2633 ] => |s, _draft| {
2634 let object = s.read_fetch_object().await
2635 .map_err(AnyConnectionError::from)?;
2636 Ok(AnyFetchObject {
2637 group_id: object.header.group_id.into_inner(),
2638 subgroup_id: Some(object.header.subgroup_id.into_inner()),
2639 object_id: object.header.object_id.into_inner(),
2640 status: object
2641 .payload
2642 .is_empty()
2643 .then_some(object.header.object_status as u64),
2644 payload: object.payload,
2645 end_of_range: None,
2646 })
2647 },
2648 [Draft14 @ "draft14"] => |s, _draft| {
2649 let object = s.read_fetch_object().await
2650 .map_err(AnyConnectionError::from)?;
2651 Ok(AnyFetchObject {
2652 group_id: object.group_id.into_inner(),
2653 subgroup_id: Some(object.subgroup_id.into_inner()),
2654 object_id: object.object_id.into_inner(),
2655 status: object.status.map(|s| s as u64),
2656 payload: object.payload,
2657 end_of_range: None,
2658 })
2659 },
2660 // Draft-15 elides fields too, and its own reader resolves them, so
2661 // what comes back is a header whose every field is a value.
2662 [Draft15 @ "draft15"] => |s, _draft| {
2663 let (header, payload) = s.read_fetch_object().await
2664 .map_err(AnyConnectionError::from)?;
2665 Ok(AnyFetchObject {
2666 group_id: header.group_id.into_inner(),
2667 subgroup_id: Some(header.subgroup_id.into_inner()),
2668 object_id: header.object_id.into_inner(),
2669 status: header.object_status.map(|s| s as u64),
2670 payload,
2671 end_of_range: None,
2672 })
2673 },
2674 // The one arm that resolves rather than reads a resolution. See
2675 // `Draft16FetchStream` for why the state is here and not on the
2676 // stream.
2677 [Draft16 @ "draft16"] => |s, _draft| {
2678 let (header, payload) = s.stream.read_fetch_object().await
2679 .map_err(AnyConnectionError::from)?;
2680 // A Malformed Track is the peer's, and reaches the caller as
2681 // the codec error the draft's withdrawal reads.
2682 let at = s.reader.resolve(&header).map_err(|e| match e {
2683 moqtap_codec::error::CodecError::MalformedTrack { .. }
2684 | moqtap_codec::error::CodecError::MalformedObjectProperties { .. } => {
2685 AnyConnectionError::from(
2686 crate::draft16::connection::ConnectionError::Codec(e),
2687 )
2688 }
2689 e => AnyConnectionError::facade(e.to_string()),
2690 })?;
2691 Ok(AnyFetchObject {
2692 group_id: at.group_id,
2693 subgroup_id: at.subgroup_id,
2694 object_id: at.object_id,
2695 status: None,
2696 payload,
2697 end_of_range: at.end_of_range.map(|marker| marker.as_u64()),
2698 })
2699 },
2700 // Drafts 17 through 20 hand back the resolved Location beside the
2701 // header the wire carried, and the marker is read off the flags
2702 // rather than off the enum: three drafts name two markers and
2703 // draft-20 names three, so a shared body naming them would not
2704 // compile on all four.
2705 [
2706 Draft17 @ "draft17", Draft19 @ "draft19", Draft20 @ "draft20", Draft21 @ "draft21",
2707 Draft22 @ "draft22",
2708 ] => |s, _draft| {
2709 let (object, payload) = s.read_fetch_object().await
2710 .map_err(AnyConnectionError::from)?;
2711 let flags = object.header.serialization_flags.into_inner();
2712 Ok(AnyFetchObject {
2713 group_id: object.group_id,
2714 subgroup_id: object.subgroup_id,
2715 object_id: object.object_id,
2716 status: None,
2717 payload,
2718 end_of_range: object.header.end_of_range().map(|_| flags),
2719 })
2720 },
2721 // draft-23 Section 11.4.2.2: "When Serialization Flags is 0x80 (End
2722 // of Group) or 0x81 (End of Track), no other fields are present.
2723 // The prior Object is the last Object in its Group, or in the
2724 // Track, respectively." Those two are indicators as well as the
2725 // End of Range ones, so `is_indicator` rather than `end_of_range`.
2726 [Draft23 @ "draft23"] => |s, _draft| {
2727 let (object, payload) = s.read_fetch_object().await
2728 .map_err(AnyConnectionError::from)?;
2729 let flags = object.header.serialization_flags.into_inner();
2730 Ok(AnyFetchObject {
2731 group_id: object.group_id,
2732 subgroup_id: object.subgroup_id,
2733 object_id: object.object_id,
2734 status: None,
2735 payload,
2736 end_of_range: object.header.is_indicator().then_some(flags),
2737 })
2738 },
2739 // Identical but for the flags field's type, which draft-18 alone
2740 // holds as a bare `u64`.
2741 [Draft18 @ "draft18"] => |s, _draft| {
2742 let (object, payload) = s.read_fetch_object().await
2743 .map_err(AnyConnectionError::from)?;
2744 let flags = object.header.serialization_flags;
2745 Ok(AnyFetchObject {
2746 group_id: object.group_id,
2747 subgroup_id: object.subgroup_id,
2748 object_id: object.object_id,
2749 status: None,
2750 payload,
2751 end_of_range: object.header.end_of_range().map(|_| flags),
2752 })
2753 },
2754 }
2755 }
2756}
2757
2758/// `forward` as the FORWARD parameter, key 0x10, which is where drafts 15
2759/// through 22 carry the field drafts 12 through 14 write inline in a PUBLISH.
2760/// The codec owns the value's encoding on each draft, so this hands it over as
2761/// a plain varint.
2762#[cfg(any(
2763 feature = "draft15",
2764 feature = "draft16",
2765 feature = "draft17",
2766 feature = "draft18",
2767 feature = "draft19",
2768 feature = "draft20",
2769 feature = "draft21",
2770 feature = "draft22",
2771))]
2772fn forward_parameter(forward: moqtap_codec::types::Forward) -> Vec<KeyValuePair> {
2773 use moqtap_codec::kvp::KvpValue;
2774 use moqtap_codec::varint::VarInt;
2775 vec![KeyValuePair {
2776 key: VarInt::from_u64(0x10).expect("0x10 fits in VarInt"),
2777 value: KvpValue::Varint(VarInt::from_u64(forward as u64).expect("0 or 1 fits in VarInt")),
2778 }]
2779}
2780
2781impl AnyConnection {
2782 /// Connect to a MoQT server using the requested draft. Builds the
2783 /// draft-specific `ClientConfig` from the provided [`AnyClientConfig`]
2784 /// and dispatches to the appropriate `Connection::connect`.
2785 pub async fn connect(addr: &str, config: AnyClientConfig) -> Result<Self, AnyConnectionError> {
2786 // Every arm of the match below is `#[cfg(feature = "draftNN")]`. A build
2787 // with no draft feature enabled keeps only the catch-all, which never
2788 // dials anything, so `addr` is genuinely unread in exactly that build.
2789 #[cfg(not(any(
2790 feature = "draft07",
2791 feature = "draft08",
2792 feature = "draft09",
2793 feature = "draft10",
2794 feature = "draft11",
2795 feature = "draft12",
2796 feature = "draft13",
2797 feature = "draft14",
2798 feature = "draft15",
2799 feature = "draft16",
2800 feature = "draft17",
2801 feature = "draft18",
2802 feature = "draft19",
2803 feature = "draft20",
2804 feature = "draft21",
2805 feature = "draft22",
2806 feature = "draft23"
2807 )))]
2808 let _ = addr;
2809 match config.draft {
2810 #[cfg(feature = "draft07")]
2811 DraftVersion::Draft07 => {
2812 use crate::draft07::connection::{ClientConfig, Connection, TransportType};
2813 let transport = match config.transport {
2814 AnyTransportType::Quic => TransportType::Quic,
2815 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2816 };
2817 let inner = ClientConfig {
2818 additional_versions: config.additional_versions,
2819 transport,
2820 skip_cert_verification: config.skip_cert_verification,
2821 ca_certs: config.ca_certs,
2822 setup_parameters: config.setup_parameters,
2823 };
2824 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2825 Ok(AnyConnection::Draft07(c))
2826 }
2827 #[cfg(feature = "draft08")]
2828 DraftVersion::Draft08 => {
2829 use crate::draft08::connection::{ClientConfig, Connection, TransportType};
2830 let transport = match config.transport {
2831 AnyTransportType::Quic => TransportType::Quic,
2832 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2833 };
2834 let inner = ClientConfig {
2835 additional_versions: config.additional_versions,
2836 transport,
2837 skip_cert_verification: config.skip_cert_verification,
2838 ca_certs: config.ca_certs,
2839 setup_parameters: config.setup_parameters,
2840 };
2841 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2842 Ok(AnyConnection::Draft08(c))
2843 }
2844 #[cfg(feature = "draft09")]
2845 DraftVersion::Draft09 => {
2846 use crate::draft09::connection::{ClientConfig, Connection, TransportType};
2847 let transport = match config.transport {
2848 AnyTransportType::Quic => TransportType::Quic,
2849 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2850 };
2851 let inner = ClientConfig {
2852 additional_versions: config.additional_versions,
2853 transport,
2854 skip_cert_verification: config.skip_cert_verification,
2855 ca_certs: config.ca_certs,
2856 setup_parameters: config.setup_parameters,
2857 };
2858 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2859 Ok(AnyConnection::Draft09(c))
2860 }
2861 #[cfg(feature = "draft10")]
2862 DraftVersion::Draft10 => {
2863 use crate::draft10::connection::{ClientConfig, Connection, TransportType};
2864 let transport = match config.transport {
2865 AnyTransportType::Quic => TransportType::Quic,
2866 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2867 };
2868 let inner = ClientConfig {
2869 additional_versions: config.additional_versions,
2870 transport,
2871 skip_cert_verification: config.skip_cert_verification,
2872 ca_certs: config.ca_certs,
2873 setup_parameters: config.setup_parameters,
2874 };
2875 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2876 Ok(AnyConnection::Draft10(c))
2877 }
2878 #[cfg(feature = "draft11")]
2879 DraftVersion::Draft11 => {
2880 use crate::draft11::connection::{ClientConfig, Connection, TransportType};
2881 let transport = match config.transport {
2882 AnyTransportType::Quic => TransportType::Quic,
2883 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2884 };
2885 let inner = ClientConfig {
2886 additional_versions: config.additional_versions,
2887 transport,
2888 skip_cert_verification: config.skip_cert_verification,
2889 ca_certs: config.ca_certs,
2890 setup_parameters: config.setup_parameters,
2891 };
2892 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2893 Ok(AnyConnection::Draft11(c))
2894 }
2895 #[cfg(feature = "draft12")]
2896 DraftVersion::Draft12 => {
2897 use crate::draft12::connection::{ClientConfig, Connection, TransportType};
2898 let transport = match config.transport {
2899 AnyTransportType::Quic => TransportType::Quic,
2900 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2901 };
2902 let inner = ClientConfig {
2903 additional_versions: config.additional_versions,
2904 transport,
2905 skip_cert_verification: config.skip_cert_verification,
2906 ca_certs: config.ca_certs,
2907 setup_parameters: config.setup_parameters,
2908 };
2909 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2910 Ok(AnyConnection::Draft12(c))
2911 }
2912 #[cfg(feature = "draft13")]
2913 DraftVersion::Draft13 => {
2914 use crate::draft13::connection::{ClientConfig, Connection, TransportType};
2915 let transport = match config.transport {
2916 AnyTransportType::Quic => TransportType::Quic,
2917 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2918 };
2919 let inner = ClientConfig {
2920 additional_versions: config.additional_versions,
2921 transport,
2922 skip_cert_verification: config.skip_cert_verification,
2923 ca_certs: config.ca_certs,
2924 setup_parameters: config.setup_parameters,
2925 };
2926 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2927 Ok(AnyConnection::Draft13(c))
2928 }
2929 #[cfg(feature = "draft14")]
2930 DraftVersion::Draft14 => {
2931 use crate::draft14::connection::{ClientConfig, Connection, TransportType};
2932 let transport = match config.transport {
2933 AnyTransportType::Quic => TransportType::Quic,
2934 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2935 };
2936 let inner = ClientConfig {
2937 draft: config.draft,
2938 additional_versions: config.additional_versions,
2939 transport,
2940 skip_cert_verification: config.skip_cert_verification,
2941 ca_certs: config.ca_certs,
2942 setup_parameters: config.setup_parameters,
2943 };
2944 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2945 Ok(AnyConnection::Draft14(c))
2946 }
2947 #[cfg(feature = "draft15")]
2948 DraftVersion::Draft15 => {
2949 use crate::draft15::connection::{ClientConfig, Connection, TransportType};
2950 let transport = match config.transport {
2951 AnyTransportType::Quic => TransportType::Quic,
2952 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2953 };
2954 let inner = ClientConfig {
2955 draft: config.draft,
2956 transport,
2957 skip_cert_verification: config.skip_cert_verification,
2958 ca_certs: config.ca_certs,
2959 setup_parameters: config.setup_parameters,
2960 };
2961 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2962 Ok(AnyConnection::Draft15(c))
2963 }
2964 #[cfg(feature = "draft16")]
2965 DraftVersion::Draft16 => {
2966 use crate::draft16::connection::{ClientConfig, Connection, TransportType};
2967 let transport = match config.transport {
2968 AnyTransportType::Quic => TransportType::Quic,
2969 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2970 };
2971 let inner = ClientConfig {
2972 draft: config.draft,
2973 transport,
2974 skip_cert_verification: config.skip_cert_verification,
2975 ca_certs: config.ca_certs,
2976 setup_parameters: config.setup_parameters,
2977 };
2978 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2979 Ok(AnyConnection::Draft16(c))
2980 }
2981 #[cfg(feature = "draft17")]
2982 DraftVersion::Draft17 => {
2983 use crate::draft17::connection::{ClientConfig, Connection, TransportType};
2984 let transport = match config.transport {
2985 AnyTransportType::Quic => TransportType::Quic,
2986 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
2987 };
2988 let inner = ClientConfig {
2989 draft: config.draft,
2990 transport,
2991 skip_cert_verification: config.skip_cert_verification,
2992 ca_certs: config.ca_certs,
2993 setup_parameters: config.setup_parameters,
2994 };
2995 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
2996 Ok(AnyConnection::Draft17(c))
2997 }
2998 #[cfg(feature = "draft18")]
2999 DraftVersion::Draft18 => {
3000 use crate::draft18::connection::{ClientConfig, Connection, TransportType};
3001 let transport = match config.transport {
3002 AnyTransportType::Quic => TransportType::Quic,
3003 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
3004 };
3005 let inner = ClientConfig {
3006 draft: config.draft,
3007 transport,
3008 skip_cert_verification: config.skip_cert_verification,
3009 ca_certs: config.ca_certs,
3010 setup_parameters: config.setup_parameters,
3011 };
3012 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
3013 Ok(AnyConnection::Draft18(c))
3014 }
3015 #[cfg(feature = "draft19")]
3016 DraftVersion::Draft19 => {
3017 use crate::draft19::connection::{ClientConfig, Connection, TransportType};
3018 let transport = match config.transport {
3019 AnyTransportType::Quic => TransportType::Quic,
3020 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
3021 };
3022 let inner = ClientConfig {
3023 draft: config.draft,
3024 transport,
3025 skip_cert_verification: config.skip_cert_verification,
3026 ca_certs: config.ca_certs,
3027 setup_parameters: config.setup_parameters,
3028 };
3029 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
3030 Ok(AnyConnection::Draft19(c))
3031 }
3032 #[cfg(feature = "draft20")]
3033 DraftVersion::Draft20 => {
3034 use crate::draft20::connection::{ClientConfig, Connection, TransportType};
3035 let transport = match config.transport {
3036 AnyTransportType::Quic => TransportType::Quic,
3037 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
3038 };
3039 let inner = ClientConfig {
3040 draft: config.draft,
3041 transport,
3042 skip_cert_verification: config.skip_cert_verification,
3043 ca_certs: config.ca_certs,
3044 setup_parameters: config.setup_parameters,
3045 };
3046 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
3047 Ok(AnyConnection::Draft20(c))
3048 }
3049 #[cfg(feature = "draft21")]
3050 DraftVersion::Draft21 => {
3051 use crate::draft21::connection::{ClientConfig, Connection, TransportType};
3052 let transport = match config.transport {
3053 AnyTransportType::Quic => TransportType::Quic,
3054 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
3055 };
3056 let inner = ClientConfig {
3057 draft: config.draft,
3058 transport,
3059 skip_cert_verification: config.skip_cert_verification,
3060 ca_certs: config.ca_certs,
3061 setup_parameters: config.setup_parameters,
3062 };
3063 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
3064 Ok(AnyConnection::Draft21(c))
3065 }
3066 #[cfg(feature = "draft22")]
3067 DraftVersion::Draft22 => {
3068 use crate::draft22::connection::{ClientConfig, Connection, TransportType};
3069 let transport = match config.transport {
3070 AnyTransportType::Quic => TransportType::Quic,
3071 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
3072 };
3073 let inner = ClientConfig {
3074 draft: config.draft,
3075 transport,
3076 skip_cert_verification: config.skip_cert_verification,
3077 ca_certs: config.ca_certs,
3078 setup_parameters: config.setup_parameters,
3079 };
3080 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
3081 Ok(AnyConnection::Draft22(c))
3082 }
3083 #[cfg(feature = "draft23")]
3084 DraftVersion::Draft23 => {
3085 use crate::draft23::connection::{ClientConfig, Connection, TransportType};
3086 let transport = match config.transport {
3087 AnyTransportType::Quic => TransportType::Quic,
3088 AnyTransportType::WebTransport { url } => TransportType::WebTransport { url },
3089 };
3090 let inner = ClientConfig {
3091 draft: config.draft,
3092 transport,
3093 skip_cert_verification: config.skip_cert_verification,
3094 ca_certs: config.ca_certs,
3095 setup_parameters: config.setup_parameters,
3096 };
3097 let c = Connection::connect(addr, inner).await.map_err(AnyConnectionError::from)?;
3098 Ok(AnyConnection::Draft23(c))
3099 }
3100 #[allow(unreachable_patterns)]
3101 other => Err(AnyConnectionError::facade(format!(
3102 "draft {other:?} not enabled in this build",
3103 ))),
3104 }
3105 }
3106
3107 /// Run the MoQT setup handshake over a transport somebody else established.
3108 ///
3109 /// [`connect`](Self::connect) dials its own socket, which is right for a
3110 /// caller that wants a connection and wrong for one that wants to *measure*
3111 /// how a peer behaves: choosing the address, the SNI, the ALPN offer or the
3112 /// certificate policy all mean dialling first and adopting after. Every
3113 /// draft module has `Connection::adopt` for exactly this, and this wrapper
3114 /// is how a caller reaches it without re-implementing the per-draft
3115 /// match over `AnyConnection` for itself.
3116 ///
3117 /// `config.draft` selects the module. Nothing here re-checks it against the
3118 /// ALPN the transport was negotiated with: adopting a transport under a
3119 /// draft the peer did not agree to is a legitimate probe, and refusing it
3120 /// would remove the ability to ask.
3121 pub async fn adopt(
3122 transport: crate::transport::Transport,
3123 config: AnyClientConfig,
3124 ) -> Result<Self, AnyConnectionError> {
3125 Self::adopt_offering(transport, config, None).await
3126 }
3127
3128 /// [`Self::adopt`], offering exactly `versions` in CLIENT_SETUP.
3129 ///
3130 /// `None` offers what `config` implies. `Some` replaces the list outright,
3131 /// and takes raw varints rather than [`DraftVersion`]s because the reason
3132 /// to reach for this is to offer a version no draft assigns — which an enum
3133 /// of drafts cannot name.
3134 ///
3135 /// Drafts 07-14 only. Drafts 15-23 settle the version by ALPN and put no
3136 /// version list on the wire, so there is nothing there to offer and a
3137 /// `Some` on one of them is refused rather than quietly ignored: silently
3138 /// sending the ordinary handshake would answer a question that was never
3139 /// asked.
3140 // Both `transport` and `versions` are read only by the per-draft arms, so
3141 // the `<zero drafts>` build reaches none of them. Allowed rather than
3142 // underscore-prefixed because the names are part of the public signature
3143 // and appear in the docs above; renaming them to satisfy a build that
3144 // enables no draft would make the documentation wrong for every build that
3145 // enables one.
3146 #[allow(unused_variables)]
3147 pub async fn adopt_offering(
3148 transport: crate::transport::Transport,
3149 config: AnyClientConfig,
3150 versions: Option<Vec<moqtap_codec::varint::VarInt>>,
3151 ) -> Result<Self, AnyConnectionError> {
3152 // Three shapes, not one, because `ClientConfig` is not the same struct
3153 // in every era and a single arm cannot spell all three:
3154 //
3155 // 07-13 no `draft` field (the module is the draft), offers a
3156 // version list through `additional_versions`
3157 // 14 both — the last draft that can offer several versions at once
3158 // 15-23 `draft` only. One connection offers exactly one version,
3159 // which is why enumerating these costs a connection each.
3160 macro_rules! adopt_dispatch {
3161 (
3162 legacy: [ $( ($lf:literal, $lv:ident, $lm:ident) ),* $(,)? ],
3163 versioned: [ $( ($vf:literal, $vv:ident, $vm:ident) ),* $(,)? ],
3164 single: [ $( ($sf:literal, $sv:ident, $sm:ident) ),* $(,)? ],
3165 ) => {
3166 match config.draft {
3167 $(
3168 #[cfg(feature = $lf)]
3169 DraftVersion::$lv => {
3170 use crate::$lm::connection::{ClientConfig, Connection, TransportType};
3171 let inner = ClientConfig {
3172 additional_versions: config.additional_versions,
3173 transport: match config.transport {
3174 AnyTransportType::Quic => TransportType::Quic,
3175 AnyTransportType::WebTransport { url } => {
3176 TransportType::WebTransport { url }
3177 }
3178 },
3179 skip_cert_verification: config.skip_cert_verification,
3180 ca_certs: config.ca_certs,
3181 setup_parameters: config.setup_parameters,
3182 };
3183 let c = Connection::adopt_offering(transport, inner, versions)
3184 .await
3185 .map_err(AnyConnectionError::from)?;
3186 Ok(AnyConnection::$lv(c))
3187 }
3188 )*
3189 $(
3190 #[cfg(feature = $vf)]
3191 DraftVersion::$vv => {
3192 use crate::$vm::connection::{ClientConfig, Connection, TransportType};
3193 let inner = ClientConfig {
3194 draft: config.draft,
3195 additional_versions: config.additional_versions,
3196 transport: match config.transport {
3197 AnyTransportType::Quic => TransportType::Quic,
3198 AnyTransportType::WebTransport { url } => {
3199 TransportType::WebTransport { url }
3200 }
3201 },
3202 skip_cert_verification: config.skip_cert_verification,
3203 ca_certs: config.ca_certs,
3204 setup_parameters: config.setup_parameters,
3205 };
3206 let c = Connection::adopt_offering(transport, inner, versions)
3207 .await
3208 .map_err(AnyConnectionError::from)?;
3209 Ok(AnyConnection::$vv(c))
3210 }
3211 )*
3212 $(
3213 #[cfg(feature = $sf)]
3214 DraftVersion::$sv => {
3215 use crate::$sm::connection::{ClientConfig, Connection, TransportType};
3216 if versions.is_some() {
3217 return Err(AnyConnectionError::facade(format!(
3218 "draft {:?} settles its version by ALPN and puts no version \
3219 list on the wire, so there is nothing to offer",
3220 config.draft
3221 )));
3222 }
3223 let inner = ClientConfig {
3224 draft: config.draft,
3225 transport: match config.transport {
3226 AnyTransportType::Quic => TransportType::Quic,
3227 AnyTransportType::WebTransport { url } => {
3228 TransportType::WebTransport { url }
3229 }
3230 },
3231 skip_cert_verification: config.skip_cert_verification,
3232 ca_certs: config.ca_certs,
3233 setup_parameters: config.setup_parameters,
3234 };
3235 let c = Connection::adopt(transport, inner)
3236 .await
3237 .map_err(AnyConnectionError::from)?;
3238 Ok(AnyConnection::$sv(c))
3239 }
3240 )*
3241 #[allow(unreachable_patterns)]
3242 other => Err(AnyConnectionError::facade(format!(
3243 "draft {other:?} not enabled in this build"
3244 ))),
3245 }
3246 };
3247 }
3248
3249 adopt_dispatch! {
3250 legacy: [
3251 ("draft07", Draft07, draft07),
3252 ("draft08", Draft08, draft08),
3253 ("draft09", Draft09, draft09),
3254 ("draft10", Draft10, draft10),
3255 ("draft11", Draft11, draft11),
3256 ("draft12", Draft12, draft12),
3257 ("draft13", Draft13, draft13),
3258 ],
3259 versioned: [
3260 ("draft14", Draft14, draft14),
3261 ],
3262 single: [
3263 ("draft15", Draft15, draft15),
3264 ("draft16", Draft16, draft16),
3265 ("draft17", Draft17, draft17),
3266 ("draft18", Draft18, draft18),
3267 ("draft19", Draft19, draft19),
3268 ("draft20", Draft20, draft20),
3269 ("draft21", Draft21, draft21),
3270 ("draft22", Draft22, draft22),
3271 ("draft23", Draft23, draft23),
3272 ],
3273 }
3274 }
3275
3276 /// The version SERVER_SETUP selected, when the draft has a version list to
3277 /// select from.
3278 ///
3279 /// Not the same question as [`draft`](Self::draft), and the difference is
3280 /// the whole point for a caller enumerating what a relay supports:
3281 /// `draft()` reports the module the connection is running, which is the one
3282 /// the *client* chose, while this reports what the *server* picked out of
3283 /// the versions offered. Offer 11 through 14 and a server that settles on
3284 /// 12 leaves `draft()` saying 14 and this saying 12.
3285 ///
3286 /// `None` for drafts 15-23, which is a fact about those drafts rather than
3287 /// a gap here: they carry no `additional_versions`, so a connection offers
3288 /// exactly one version and there is nothing for a server to choose between.
3289 /// What a peer supports there is discovered by ALPN instead.
3290 pub fn negotiated_version(&self) -> Option<moqtap_codec::varint::VarInt> {
3291 match self {
3292 #[cfg(feature = "draft07")]
3293 Self::Draft07(c) => c.negotiated_version(),
3294 #[cfg(feature = "draft08")]
3295 Self::Draft08(c) => c.negotiated_version(),
3296 #[cfg(feature = "draft09")]
3297 Self::Draft09(c) => c.negotiated_version(),
3298 #[cfg(feature = "draft10")]
3299 Self::Draft10(c) => c.negotiated_version(),
3300 #[cfg(feature = "draft11")]
3301 Self::Draft11(c) => c.negotiated_version(),
3302 #[cfg(feature = "draft12")]
3303 Self::Draft12(c) => c.negotiated_version(),
3304 #[cfg(feature = "draft13")]
3305 Self::Draft13(c) => c.negotiated_version(),
3306 #[cfg(feature = "draft14")]
3307 Self::Draft14(c) => c.negotiated_version(),
3308 #[allow(unreachable_patterns)]
3309 _ => None,
3310 }
3311 }
3312
3313 /// Read and dispatch one control message on the active draft. Draft-specific
3314 /// control-message return values are discarded because event delivery goes
3315 /// through the attached observer; callers only care about success/failure.
3316 pub async fn recv_and_dispatch(&mut self) -> Result<(), AnyConnectionError> {
3317 match self {
3318 #[cfg(feature = "draft07")]
3319 Self::Draft07(c) => {
3320 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3321 }
3322 #[cfg(feature = "draft08")]
3323 Self::Draft08(c) => {
3324 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3325 }
3326 #[cfg(feature = "draft09")]
3327 Self::Draft09(c) => {
3328 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3329 }
3330 #[cfg(feature = "draft10")]
3331 Self::Draft10(c) => {
3332 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3333 }
3334 #[cfg(feature = "draft11")]
3335 Self::Draft11(c) => {
3336 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3337 }
3338 #[cfg(feature = "draft12")]
3339 Self::Draft12(c) => {
3340 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3341 }
3342 #[cfg(feature = "draft13")]
3343 Self::Draft13(c) => {
3344 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3345 }
3346 #[cfg(feature = "draft14")]
3347 Self::Draft14(c) => {
3348 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3349 }
3350 #[cfg(feature = "draft15")]
3351 Self::Draft15(c) => {
3352 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3353 }
3354 #[cfg(feature = "draft16")]
3355 Self::Draft16(c) => {
3356 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3357 }
3358 #[cfg(feature = "draft17")]
3359 Self::Draft17(c) => {
3360 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3361 }
3362 #[cfg(feature = "draft18")]
3363 Self::Draft18(c) => {
3364 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3365 }
3366 #[cfg(feature = "draft19")]
3367 Self::Draft19(c) => {
3368 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3369 }
3370 #[cfg(feature = "draft20")]
3371 Self::Draft20(c) => {
3372 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3373 }
3374 #[cfg(feature = "draft21")]
3375 Self::Draft21(c) => {
3376 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3377 }
3378 #[cfg(feature = "draft22")]
3379 Self::Draft22(c) => {
3380 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3381 }
3382 #[cfg(feature = "draft23")]
3383 Self::Draft23(c) => {
3384 c.recv_and_dispatch().await.map(|_| ()).map_err(AnyConnectionError::from)
3385 }
3386 #[allow(unreachable_patterns)]
3387 _ => Err(AnyConnectionError::facade("AnyConnection has no enabled variants")),
3388 }
3389 }
3390
3391 /// Read the next control message that could answer `request`, wherever the
3392 /// negotiated draft carries it.
3393 ///
3394 /// This is the read half of the split [`AnyRequest`] describes, and without
3395 /// it the facade can send a request on drafts 17-23 and then has no way to
3396 /// hear the answer: those drafts put every response on the request's own
3397 /// bidirectional stream, which [`recv_and_dispatch`](Self::recv_and_dispatch)
3398 /// — a control-stream read — never touches.
3399 ///
3400 /// # What "could answer" means, and why it is not "does answer"
3401 ///
3402 /// Which of the two happens depends on the draft, and the difference is the
3403 /// protocol's, not this method's:
3404 ///
3405 /// - **Drafts 07-16, [`AnyRequest::ControlPlane`].** Every request shares
3406 /// the one control stream, so this returns *the next control message on
3407 /// it*, which is the answer only if nothing else was in flight. A peer is
3408 /// free to send MAX_REQUEST_ID or a PUBLISH_NAMESPACE of its own first.
3409 /// Correlate on the `request_id` the response carries against
3410 /// [`AnyRequest::request_id`], and read again if it is not this one.
3411 /// - **Draft-16 namespace subscriptions and drafts 17-23.** The read is on
3412 /// the request's own stream, so nothing else can arrive on it. Those
3413 /// drafts' responses carry no request id at all — the stream *is* the
3414 /// correlation — which is exactly why the read has to be addressed by the
3415 /// handle rather than by the connection.
3416 ///
3417 /// # `Ok(None)`
3418 ///
3419 /// The peer ended the stream cleanly without a message on it. Reachable
3420 /// today only on a draft-16 namespace stream, whose reader distinguishes a
3421 /// FIN from a message; drafts 17-23 report the same event as an error. A
3422 /// distinct value rather than an error because a peer that answered nothing
3423 /// and closed is a different fact from a read that failed, and a caller
3424 /// that has to tell them apart should not be reading either one out of a
3425 /// message string. Neither phrase carries quotation marks and neither may:
3426 /// they are this crate naming two outcomes, and the marks would hand them
3427 /// to the drafts the line above names.
3428 ///
3429 /// # Errors
3430 ///
3431 /// Whatever the underlying read or dispatch produced, flattened to a
3432 /// string like every other error here — a transport failure, a peer reset,
3433 /// or an endpoint refusing a message that does not fit the request's
3434 /// state. A message the endpoint refuses has still been emitted to any
3435 /// attached observer by the time this returns, so an observer is the way to
3436 /// see *what arrived* when the return value only says that something did
3437 /// not fit.
3438 ///
3439 /// A `request` from a different draft than this connection is refused
3440 /// rather than silently read on the wrong stream.
3441 #[allow(unused_variables)]
3442 pub async fn recv_response(
3443 &mut self,
3444 request: &mut AnyRequest,
3445 ) -> Result<Option<moqtap_codec::dispatch::AnyControlMessage>, AnyConnectionError> {
3446 // Used only by the per-draft arms below, so a build with no draft
3447 // feature enabled — the `<zero drafts>` row of `just draft-matrix`,
3448 // which exists to prove the crate still compiles with nothing
3449 // selected — compiles this import and reaches no arm that reads it.
3450 #[allow(unused_imports)]
3451 use moqtap_codec::dispatch::AnyControlMessage;
3452
3453 match (self, request) {
3454 #[cfg(feature = "draft07")]
3455 (Self::Draft07(c), AnyRequest::ControlPlane { .. }) => c
3456 .recv_and_dispatch()
3457 .await
3458 .map(|m| Some(AnyControlMessage::Draft07(m)))
3459 .map_err(AnyConnectionError::from),
3460 #[cfg(feature = "draft08")]
3461 (Self::Draft08(c), AnyRequest::ControlPlane { .. }) => c
3462 .recv_and_dispatch()
3463 .await
3464 .map(|m| Some(AnyControlMessage::Draft08(m)))
3465 .map_err(AnyConnectionError::from),
3466 #[cfg(feature = "draft09")]
3467 (Self::Draft09(c), AnyRequest::ControlPlane { .. }) => c
3468 .recv_and_dispatch()
3469 .await
3470 .map(|m| Some(AnyControlMessage::Draft09(m)))
3471 .map_err(AnyConnectionError::from),
3472 #[cfg(feature = "draft10")]
3473 (Self::Draft10(c), AnyRequest::ControlPlane { .. }) => c
3474 .recv_and_dispatch()
3475 .await
3476 .map(|m| Some(AnyControlMessage::Draft10(m)))
3477 .map_err(AnyConnectionError::from),
3478 #[cfg(feature = "draft11")]
3479 (Self::Draft11(c), AnyRequest::ControlPlane { .. }) => c
3480 .recv_and_dispatch()
3481 .await
3482 .map(|m| Some(AnyControlMessage::Draft11(m)))
3483 .map_err(AnyConnectionError::from),
3484 #[cfg(feature = "draft12")]
3485 (Self::Draft12(c), AnyRequest::ControlPlane { .. }) => c
3486 .recv_and_dispatch()
3487 .await
3488 .map(|m| Some(AnyControlMessage::Draft12(m)))
3489 .map_err(AnyConnectionError::from),
3490 #[cfg(feature = "draft13")]
3491 (Self::Draft13(c), AnyRequest::ControlPlane { .. }) => c
3492 .recv_and_dispatch()
3493 .await
3494 .map(|m| Some(AnyControlMessage::Draft13(m)))
3495 .map_err(AnyConnectionError::from),
3496 #[cfg(feature = "draft14")]
3497 (Self::Draft14(c), AnyRequest::ControlPlane { .. }) => c
3498 .recv_and_dispatch()
3499 .await
3500 .map(|m| Some(AnyControlMessage::Draft14(m)))
3501 .map_err(AnyConnectionError::from),
3502 #[cfg(feature = "draft15")]
3503 (Self::Draft15(c), AnyRequest::ControlPlane { .. }) => c
3504 .recv_and_dispatch()
3505 .await
3506 .map(|m| Some(AnyControlMessage::Draft15(m)))
3507 .map_err(AnyConnectionError::from),
3508 #[cfg(feature = "draft16")]
3509 (Self::Draft16(c), AnyRequest::ControlPlane { .. }) => c
3510 .recv_and_dispatch()
3511 .await
3512 .map(|m| Some(AnyControlMessage::Draft16(m)))
3513 .map_err(AnyConnectionError::from),
3514 // The one request draft-16 moved off the control stream, and so the
3515 // one place on that draft where this reads a stream instead.
3516 #[cfg(feature = "draft16")]
3517 (Self::Draft16(c), AnyRequest::Draft16(s)) => c
3518 .recv_on_namespace_stream(s)
3519 .await
3520 .map(|m| m.map(AnyControlMessage::Draft16))
3521 .map_err(AnyConnectionError::from),
3522 #[cfg(feature = "draft17")]
3523 (Self::Draft17(c), AnyRequest::Draft17(s)) => c
3524 .recv_on_request_stream(s)
3525 .await
3526 .map(|m| Some(AnyControlMessage::Draft17(m)))
3527 .map_err(AnyConnectionError::from),
3528 #[cfg(feature = "draft18")]
3529 (Self::Draft18(c), AnyRequest::Draft18(s)) => c
3530 .recv_on_request_stream(s)
3531 .await
3532 .map(|m| Some(AnyControlMessage::Draft18(m)))
3533 .map_err(AnyConnectionError::from),
3534 #[cfg(feature = "draft19")]
3535 (Self::Draft19(c), AnyRequest::Draft19(s)) => c
3536 .recv_on_request_stream(s)
3537 .await
3538 .map(|m| Some(AnyControlMessage::Draft19(m)))
3539 .map_err(AnyConnectionError::from),
3540 #[cfg(feature = "draft20")]
3541 (Self::Draft20(c), AnyRequest::Draft20(s)) => c
3542 .recv_on_request_stream(s)
3543 .await
3544 .map(|m| Some(AnyControlMessage::Draft20(m)))
3545 .map_err(AnyConnectionError::from),
3546 #[cfg(feature = "draft21")]
3547 (Self::Draft21(c), AnyRequest::Draft21(s)) => c
3548 .recv_on_request_stream(s)
3549 .await
3550 .map(|m| Some(AnyControlMessage::Draft21(m)))
3551 .map_err(AnyConnectionError::from),
3552 #[cfg(feature = "draft22")]
3553 (Self::Draft22(c), AnyRequest::Draft22(s)) => c
3554 .recv_on_request_stream(s)
3555 .await
3556 .map(|m| Some(AnyControlMessage::Draft22(m)))
3557 .map_err(AnyConnectionError::from),
3558 #[cfg(feature = "draft23")]
3559 (Self::Draft23(c), AnyRequest::Draft23(s)) => c
3560 .recv_on_request_stream(s)
3561 .await
3562 .map(|m| Some(AnyControlMessage::Draft23(m)))
3563 .map_err(AnyConnectionError::from),
3564 // A handle from another draft, or a build with no drafts enabled.
3565 // Refused rather than read on whatever stream happens to be at
3566 // hand: the two kinds of handle address different streams, so
3567 // guessing here would read the control stream for a request that is
3568 // waiting on its own.
3569 #[allow(unreachable_patterns)]
3570 (connection, request) => Err(AnyConnectionError::facade(format!(
3571 "recv_response: a draft {:?} request cannot be read on a draft {:?} connection",
3572 request.draft(),
3573 connection.draft(),
3574 ))),
3575 }
3576 }
3577
3578 // ── Unified control-message helpers ──────────────────────────────────
3579 //
3580 // Draft-agnostic shorthands. Each dispatches to the active variant and
3581 // defaults fields not expressible in the unified shape; drafts that
3582 // lack the operation return an `AnyConnectionError`. Match on the
3583 // variant directly when full per-draft control is needed.
3584
3585 /// Send an UNSUBSCRIBE for the given request ID. Drafts 07 through 16.
3586 ///
3587 /// # Drafts 17 through 23 have no such message
3588 ///
3589 /// Draft-17 deleted UNSUBSCRIBE, and every later draft keeps it deleted:
3590 /// a subscriber ends a subscription by **resetting its request stream**,
3591 /// which is [`AnyRequest::cancel`], or waits for PUBLISH_DONE. So the error
3592 /// those drafts return is not a gap to be filled later — there is nothing to
3593 /// wire — and a caller reaching for it on one of them wants `cancel` on the
3594 /// handle `subscribe` returned.
3595 #[allow(unused_variables)]
3596 pub async fn unsubscribe(
3597 &mut self,
3598 request_id: moqtap_codec::varint::VarInt,
3599 ) -> Result<(), AnyConnectionError> {
3600 match self {
3601 #[cfg(feature = "draft07")]
3602 Self::Draft07(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3603 #[cfg(feature = "draft08")]
3604 Self::Draft08(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3605 #[cfg(feature = "draft09")]
3606 Self::Draft09(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3607 #[cfg(feature = "draft10")]
3608 Self::Draft10(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3609 #[cfg(feature = "draft11")]
3610 Self::Draft11(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3611 #[cfg(feature = "draft12")]
3612 Self::Draft12(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3613 #[cfg(feature = "draft13")]
3614 Self::Draft13(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3615 #[cfg(feature = "draft14")]
3616 Self::Draft14(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3617 #[cfg(feature = "draft15")]
3618 Self::Draft15(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3619 #[cfg(feature = "draft16")]
3620 Self::Draft16(c) => c.unsubscribe(request_id).await.map_err(AnyConnectionError::from),
3621 // The text names the protocol reason and an alternative rather than
3622 // reporting the draft as unimplemented: there is nothing here left
3623 // to wire, and an error that contradicts this method's own docs is
3624 // worse than no error text at all, because the next reader believes
3625 // the error.
3626 #[allow(unreachable_patterns)]
3627 other => Err(AnyConnectionError::facade(format!(
3628 "unsubscribe: draft-{:02} has no UNSUBSCRIBE — draft-17 deleted it and the drafts \
3629 after it keep it deleted. End the subscription with AnyRequest::cancel on the \
3630 handle subscribe returned, or wait for PUBLISH_DONE",
3631 other.draft().number()
3632 ))),
3633 }
3634 }
3635
3636 /// Send a SUBSCRIBE with the given filter, priority, and group order.
3637 /// Supported on **every draft this build carries**. Drafts 15 onward
3638 /// carry priority/order/filter as parameters rather than fields;
3639 /// this helper passes an empty parameter list, so on those drafts all three
3640 /// take the protocol default and the three arguments here are ignored.
3641 ///
3642 /// # The Track Alias, and why it is not an argument
3643 ///
3644 /// Drafts 07 through 11 carry a Track Alias on SUBSCRIBE and make it the
3645 /// **subscriber's** to choose; draft-12 moved the field to SUBSCRIBE_OK and
3646 /// made it the publisher's. An argument here would therefore do nothing on
3647 /// nine of the drafts, and a fixed value would collide the moment
3648 /// a caller subscribed to a second track.
3649 ///
3650 /// So the value is read off the endpoint —
3651 /// `Connection::next_free_track_alias`, the lowest alias no live binding
3652 /// holds — rather than asked of the caller. That table is the same one the
3653 /// endpoint checks before writing, so a caller mixing these calls with a
3654 /// draft's own `Connection::subscribe` and aliases of its own is correct by
3655 /// construction rather than by convention, and a genuine duplicate is still
3656 /// refused with `EndpointError::TrackAliasInUse` before anything reaches
3657 /// the wire.
3658 ///
3659 /// There is no range to get wrong *here*: a subscription with no filter is
3660 /// one that starts where the draft says it starts, and nothing is
3661 /// converted. The two filters that name a Start Location —
3662 /// `AbsoluteStart` and `AbsoluteRange` — are refused by this call on every
3663 /// draft that takes a Filter Type as an argument, because it has no start
3664 /// location to put beside them; [`AnyConnection::subscribe_range`] is the
3665 /// entry point that takes one.
3666 /// A draft-20 caller that wants a Location filter, a fill, or anything else
3667 /// from Section 10.2 reaches
3668 /// [`draft20::connection::Connection::subscribe`](crate::draft20::connection::Connection::subscribe)
3669 /// through the variant with the parameters it wants.
3670 ///
3671 /// The returned [`AnyRequest`] must be held while the request is live: on
3672 /// drafts 17-23 it owns the bidirectional stream the request went out on
3673 /// and dropping it cancels the subscription. See [`AnyRequest`] for how
3674 /// the two kinds of handle differ.
3675 #[allow(unused_variables)]
3676 pub async fn subscribe(
3677 &mut self,
3678 namespace: moqtap_codec::types::TrackNamespace,
3679 track_name: Vec<u8>,
3680 subscriber_priority: u8,
3681 group_order: moqtap_codec::types::GroupOrder,
3682 filter_type: moqtap_codec::types::FilterType,
3683 ) -> Result<AnyRequest, AnyConnectionError> {
3684 // Only the draft-11 and draft-12 arms below convert `filter_type` by
3685 // hand: those two draw that field as a variable-length integer, where
3686 // the drafts on either side of them take the typed value straight
3687 // through. So the import is dead outside those builds.
3688 #[cfg(any(feature = "draft11", feature = "draft12"))]
3689 use moqtap_codec::varint::VarInt;
3690 let draft = self.draft();
3691 match self {
3692 // Drafts 07 through 11 put the Track Alias on SUBSCRIBE and leave
3693 // its choice to the subscriber, and have no parameter block. The
3694 // alias comes from the endpoint's own table of live bindings; see
3695 // this method's docs for why it is not an argument.
3696 #[cfg(feature = "draft07")]
3697 Self::Draft07(c) => {
3698 let alias = c.next_free_track_alias();
3699 c.subscribe(
3700 alias,
3701 namespace,
3702 track_name,
3703 subscriber_priority,
3704 group_order,
3705 filter_type,
3706 )
3707 .await
3708 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3709 .map_err(AnyConnectionError::from)
3710 }
3711 #[cfg(feature = "draft08")]
3712 Self::Draft08(c) => {
3713 let alias = c.next_free_track_alias();
3714 c.subscribe(
3715 alias,
3716 namespace,
3717 track_name,
3718 subscriber_priority,
3719 group_order,
3720 filter_type,
3721 )
3722 .await
3723 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3724 .map_err(AnyConnectionError::from)
3725 }
3726 #[cfg(feature = "draft09")]
3727 Self::Draft09(c) => {
3728 let alias = c.next_free_track_alias();
3729 c.subscribe(
3730 alias,
3731 namespace,
3732 track_name,
3733 subscriber_priority,
3734 group_order,
3735 filter_type,
3736 )
3737 .await
3738 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3739 .map_err(AnyConnectionError::from)
3740 }
3741 #[cfg(feature = "draft10")]
3742 Self::Draft10(c) => {
3743 let alias = c.next_free_track_alias();
3744 c.subscribe(
3745 alias,
3746 namespace,
3747 track_name,
3748 subscriber_priority,
3749 group_order,
3750 filter_type,
3751 )
3752 .await
3753 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3754 .map_err(AnyConnectionError::from)
3755 }
3756 #[cfg(feature = "draft11")]
3757 Self::Draft11(c) => {
3758 let ft = VarInt::from_u64(filter_type as u64)
3759 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
3760 let alias = c.next_free_track_alias();
3761 c.subscribe(alias, namespace, track_name, subscriber_priority, group_order, ft)
3762 .await
3763 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3764 .map_err(AnyConnectionError::from)
3765 }
3766 #[cfg(feature = "draft12")]
3767 Self::Draft12(c) => {
3768 let ft = VarInt::from_u64(filter_type as u64)
3769 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
3770 c.subscribe(namespace, track_name, subscriber_priority, group_order, ft, Vec::new())
3771 .await
3772 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3773 .map_err(AnyConnectionError::from)
3774 }
3775 #[cfg(feature = "draft13")]
3776 Self::Draft13(c) => c
3777 .subscribe(
3778 namespace,
3779 track_name,
3780 subscriber_priority,
3781 group_order,
3782 filter_type,
3783 Vec::new(),
3784 )
3785 .await
3786 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3787 .map_err(AnyConnectionError::from),
3788 #[cfg(feature = "draft14")]
3789 Self::Draft14(c) => c
3790 .subscribe(
3791 namespace,
3792 track_name,
3793 subscriber_priority,
3794 group_order,
3795 filter_type,
3796 Vec::new(),
3797 )
3798 .await
3799 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3800 .map_err(AnyConnectionError::from),
3801 #[cfg(feature = "draft15")]
3802 Self::Draft15(c) => c
3803 .subscribe(namespace, track_name, Vec::new())
3804 .await
3805 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3806 .map_err(AnyConnectionError::from),
3807 #[cfg(feature = "draft16")]
3808 Self::Draft16(c) => c
3809 .subscribe(namespace, track_name, Vec::new())
3810 .await
3811 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3812 .map_err(AnyConnectionError::from),
3813 #[cfg(feature = "draft17")]
3814 Self::Draft17(c) => c
3815 .subscribe(namespace, track_name, Vec::new())
3816 .await
3817 .map(AnyRequest::Draft17)
3818 .map_err(AnyConnectionError::from),
3819 #[cfg(feature = "draft18")]
3820 Self::Draft18(c) => c
3821 .subscribe(namespace, track_name, Vec::new())
3822 .await
3823 .map(AnyRequest::Draft18)
3824 .map_err(AnyConnectionError::from),
3825 #[cfg(feature = "draft19")]
3826 Self::Draft19(c) => c
3827 .subscribe(namespace, track_name, Vec::new())
3828 .await
3829 .map(AnyRequest::Draft19)
3830 .map_err(AnyConnectionError::from),
3831 #[cfg(feature = "draft20")]
3832 Self::Draft20(c) => c
3833 .subscribe(namespace, track_name, Vec::new())
3834 .await
3835 .map(AnyRequest::Draft20)
3836 .map_err(AnyConnectionError::from),
3837 #[cfg(feature = "draft21")]
3838 Self::Draft21(c) => c
3839 .subscribe(namespace, track_name, Vec::new())
3840 .await
3841 .map(AnyRequest::Draft21)
3842 .map_err(AnyConnectionError::from),
3843 #[cfg(feature = "draft22")]
3844 Self::Draft22(c) => c
3845 .subscribe(namespace, track_name, Vec::new())
3846 .await
3847 .map(AnyRequest::Draft22)
3848 .map_err(AnyConnectionError::from),
3849 #[cfg(feature = "draft23")]
3850 Self::Draft23(c) => c
3851 .subscribe(namespace, track_name, Vec::new())
3852 .await
3853 .map(AnyRequest::Draft23)
3854 .map_err(AnyConnectionError::from),
3855 #[allow(unreachable_patterns)]
3856 other => Err(AnyConnectionError::facade(format!(
3857 "subscribe: not yet wired up for draft {:?} via AnyConnection",
3858 other.draft()
3859 ))),
3860 }
3861 }
3862
3863 /// Send a SUBSCRIBE that names where the subscription starts, and
3864 /// optionally where it stops. Wired on **every draft this build carries**.
3865 ///
3866 /// This is the half of SUBSCRIBE [`AnyConnection::subscribe`] cannot reach.
3867 /// The two filters that ask a relay for anything it has **already carried**
3868 /// — AbsoluteStart and AbsoluteRange — both put a Start Location on the
3869 /// wire, and a call taking the Filter Type beside the other arguments has
3870 /// none to give. So `subscribe` refuses them and this takes a
3871 /// [`SubscribeRange`] instead, from which the Filter Type is **derived**:
3872 /// the message cannot name a filter whose fields it does not carry, because
3873 /// nothing here gets to name one.
3874 ///
3875 /// Without it there is no way to ask whether a relay holds a cache at all,
3876 /// which is a question about relays and not about ranges.
3877 ///
3878 /// # What each draft is handed
3879 ///
3880 /// Five wire shapes for one range, and the conversions are
3881 /// [`SubscribeRange`]'s rather than each arm's:
3882 ///
3883 /// * **draft-07** — a Start Location and an End *Location*, whose Object is
3884 /// the last one plus 1 with `0` for the whole Group, exactly as FETCH
3885 /// words it. [`SubscribeRange::inline_end_location`].
3886 /// * **drafts 08 through 14** — a Start Location and an absolute End
3887 /// *Group*, the End Object having been deleted in draft-08.
3888 /// * **drafts 15 and 16** — the same two, moved into the
3889 /// `SUBSCRIPTION_FILTER` parameter draft-15 introduced.
3890 /// * **drafts 17 through 19** — the same parameter, with the End Group
3891 /// written as a delta from the start.
3892 /// * **drafts 20 and 21** — `LOCATION_FILTER`, whose shape comes from its
3893 /// field count rather than from a Filter Type, and whose ranges are
3894 /// inclusive. [`SubscribeRange::location_filter_draft20`], which is also
3895 /// where the one value this facade refuses on those drafts is documented.
3896 /// * **draft-22** — the same parameter, opening with a `Location Filter
3897 /// Type` that names the shape, so `{0, 0}` is the beginning of the track
3898 /// and nothing is refused for it.
3899 /// [`SubscribeRange::location_filter_draft22`].
3900 ///
3901 /// # What this does not carry
3902 ///
3903 /// An empty parameter list on every draft that has one beside the filter,
3904 /// and the same priority and group order defaults
3905 /// [`AnyConnection::subscribe`] passes: those are fields of SUBSCRIBE on
3906 /// drafts 07 through 14 and parameters on drafts 15 and later, so they are
3907 /// taken here for the drafts that have the fields and ignored by the six
3908 /// that do not — which is the arrangement `subscribe` already documents.
3909 ///
3910 /// # Errors
3911 ///
3912 /// A range the negotiated draft cannot express, before anything is written.
3913 /// There are three, and each is a genuine difference between the drafts
3914 /// rather than a limitation of this call:
3915 /// [`SubscribeEnd::ThroughObject`] on drafts 08 through 19; an `end_group`
3916 /// below `start_group` on drafts 17 through 23; and a `{0, 0}`
3917 /// [`SubscribeEnd::Open`] on drafts 20 and 21, which read it as the live
3918 /// edge, where every other draft reads the beginning of the track.
3919 ///
3920 /// The returned [`AnyRequest`] must be held while the request is live: on
3921 /// drafts 17-23 it owns the bidirectional stream the request went out on
3922 /// and dropping it cancels the subscription.
3923 #[allow(unused_variables)]
3924 pub async fn subscribe_range(
3925 &mut self,
3926 namespace: moqtap_codec::types::TrackNamespace,
3927 track_name: Vec<u8>,
3928 subscriber_priority: u8,
3929 group_order: moqtap_codec::types::GroupOrder,
3930 range: SubscribeRange,
3931 ) -> Result<AnyRequest, AnyConnectionError> {
3932 // The Start Location every arm below 15 puts on the wire directly, and
3933 // every arm from 15 puts inside a parameter. Built once because it is
3934 // the one field none of the drafts disagree about.
3935 let start = range.start_location();
3936 let draft = self.draft();
3937 match self {
3938 // Draft-07 alone carries an End Location rather than an End Group,
3939 // and carries it with FETCH's plus-one convention. The arithmetic
3940 // is `inline_end_location`'s, in one place, so it cannot reach the
3941 // drafts that must not have it.
3942 #[cfg(feature = "draft07")]
3943 Self::Draft07(c) => {
3944 let end = range.inline_end_location()?;
3945 let alias = c.next_free_track_alias();
3946 c.subscribe_range(
3947 alias,
3948 namespace,
3949 track_name,
3950 subscriber_priority,
3951 group_order,
3952 start,
3953 end,
3954 )
3955 .await
3956 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3957 .map_err(AnyConnectionError::from)
3958 }
3959 // Drafts 08 through 11 put the Track Alias on SUBSCRIBE and make it
3960 // the subscriber's to choose, and have no parameter block; see
3961 // `subscribe` for why the alias is read off the endpoint rather
3962 // than taken as an argument.
3963 #[cfg(feature = "draft08")]
3964 Self::Draft08(c) => {
3965 let end =
3966 range.group_only_end(draft)?.map(moqtap_codec::varint::VarInt::from_u64_moqt);
3967 let alias = c.next_free_track_alias();
3968 c.subscribe_range(
3969 alias,
3970 namespace,
3971 track_name,
3972 subscriber_priority,
3973 group_order,
3974 start,
3975 end,
3976 )
3977 .await
3978 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3979 .map_err(AnyConnectionError::from)
3980 }
3981 #[cfg(feature = "draft09")]
3982 Self::Draft09(c) => {
3983 let end =
3984 range.group_only_end(draft)?.map(moqtap_codec::varint::VarInt::from_u64_moqt);
3985 let alias = c.next_free_track_alias();
3986 c.subscribe_range(
3987 alias,
3988 namespace,
3989 track_name,
3990 subscriber_priority,
3991 group_order,
3992 start,
3993 end,
3994 )
3995 .await
3996 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
3997 .map_err(AnyConnectionError::from)
3998 }
3999 #[cfg(feature = "draft10")]
4000 Self::Draft10(c) => {
4001 let end =
4002 range.group_only_end(draft)?.map(moqtap_codec::varint::VarInt::from_u64_moqt);
4003 let alias = c.next_free_track_alias();
4004 c.subscribe_range(
4005 alias,
4006 namespace,
4007 track_name,
4008 subscriber_priority,
4009 group_order,
4010 start,
4011 end,
4012 )
4013 .await
4014 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4015 .map_err(AnyConnectionError::from)
4016 }
4017 #[cfg(feature = "draft11")]
4018 Self::Draft11(c) => {
4019 let end =
4020 range.group_only_end(draft)?.map(moqtap_codec::varint::VarInt::from_u64_moqt);
4021 let alias = c.next_free_track_alias();
4022 c.subscribe_range(
4023 alias,
4024 namespace,
4025 track_name,
4026 subscriber_priority,
4027 group_order,
4028 start,
4029 end,
4030 )
4031 .await
4032 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4033 .map_err(AnyConnectionError::from)
4034 }
4035 // Drafts 12 through 14: the alias moved to SUBSCRIBE_OK and a
4036 // parameter block arrived, and the range is still two fields on
4037 // SUBSCRIBE itself.
4038 #[cfg(feature = "draft12")]
4039 Self::Draft12(c) => {
4040 let end =
4041 range.group_only_end(draft)?.map(moqtap_codec::varint::VarInt::from_u64_moqt);
4042 c.subscribe_range(
4043 namespace,
4044 track_name,
4045 subscriber_priority,
4046 group_order,
4047 start,
4048 end,
4049 Vec::new(),
4050 )
4051 .await
4052 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4053 .map_err(AnyConnectionError::from)
4054 }
4055 #[cfg(feature = "draft13")]
4056 Self::Draft13(c) => {
4057 let end =
4058 range.group_only_end(draft)?.map(moqtap_codec::varint::VarInt::from_u64_moqt);
4059 c.subscribe_range(
4060 namespace,
4061 track_name,
4062 subscriber_priority,
4063 group_order,
4064 start,
4065 end,
4066 Vec::new(),
4067 )
4068 .await
4069 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4070 .map_err(AnyConnectionError::from)
4071 }
4072 #[cfg(feature = "draft14")]
4073 Self::Draft14(c) => {
4074 let end =
4075 range.group_only_end(draft)?.map(moqtap_codec::varint::VarInt::from_u64_moqt);
4076 c.subscribe_range(
4077 namespace,
4078 track_name,
4079 subscriber_priority,
4080 group_order,
4081 start,
4082 end,
4083 Vec::new(),
4084 )
4085 .await
4086 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4087 .map_err(AnyConnectionError::from)
4088 }
4089 // From draft-15 the whole filter is a parameter value, so the
4090 // range does not reach the endpoint as arguments at all and there
4091 // is no `subscribe_range` under here to call. `group_only_end`
4092 // still runs, inside `subscription_filter`, because these drafts
4093 // deleted the End Object along with drafts 08 through 14.
4094 #[cfg(feature = "draft15")]
4095 Self::Draft15(c) => {
4096 let filter = range
4097 .subscription_filter(draft, false)?
4098 .parameter()
4099 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4100 c.subscribe(namespace, track_name, vec![filter])
4101 .await
4102 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4103 .map_err(AnyConnectionError::from)
4104 }
4105 #[cfg(feature = "draft16")]
4106 Self::Draft16(c) => {
4107 let filter = range
4108 .subscription_filter(draft, false)?
4109 .parameter()
4110 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4111 c.subscribe(namespace, track_name, vec![filter])
4112 .await
4113 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4114 .map_err(AnyConnectionError::from)
4115 }
4116 // Draft-17 introduced the End Group Delta and its own integer
4117 // encoding, in which the 7-byte length is an invalid code point.
4118 // Draft-18 restored that length, which is why the profile differs
4119 // between this arm and the two below it.
4120 #[cfg(feature = "draft17")]
4121 Self::Draft17(c) => {
4122 let filter = range
4123 .subscription_filter(draft, true)?
4124 .parameter_moqt::<moqtap_codec::varint::Moqt17>()
4125 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4126 c.subscribe(namespace, track_name, vec![filter])
4127 .await
4128 .map(AnyRequest::Draft17)
4129 .map_err(AnyConnectionError::from)
4130 }
4131 #[cfg(feature = "draft18")]
4132 Self::Draft18(c) => {
4133 let filter = range
4134 .subscription_filter(draft, true)?
4135 .parameter_moqt::<moqtap_codec::varint::Moqt18>()
4136 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4137 c.subscribe(namespace, track_name, vec![filter])
4138 .await
4139 .map(AnyRequest::Draft18)
4140 .map_err(AnyConnectionError::from)
4141 }
4142 #[cfg(feature = "draft19")]
4143 Self::Draft19(c) => {
4144 let filter = range
4145 .subscription_filter(draft, true)?
4146 .parameter_moqt::<moqtap_codec::varint::Moqt18>()
4147 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4148 c.subscribe(namespace, track_name, vec![filter])
4149 .await
4150 .map(AnyRequest::Draft19)
4151 .map_err(AnyConnectionError::from)
4152 }
4153 // Draft-20 deleted the Filter Type enum and reads the shape off the
4154 // field count, so this arm builds a different value from the same
4155 // range rather than the same value with a different integer
4156 // encoding.
4157 #[cfg(feature = "draft20")]
4158 Self::Draft20(c) => {
4159 let filter = range
4160 .location_filter_draft20()?
4161 .parameter()
4162 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4163 c.subscribe(namespace, track_name, vec![filter])
4164 .await
4165 .map(AnyRequest::Draft20)
4166 .map_err(AnyConnectionError::from)
4167 }
4168 #[cfg(feature = "draft21")]
4169 Self::Draft21(c) => {
4170 let filter = range
4171 .location_filter_draft21()?
4172 .parameter()
4173 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4174 c.subscribe(namespace, track_name, vec![filter])
4175 .await
4176 .map(AnyRequest::Draft21)
4177 .map_err(AnyConnectionError::from)
4178 }
4179 #[cfg(feature = "draft22")]
4180 Self::Draft22(c) => {
4181 let filter = range
4182 .location_filter_draft22()?
4183 .parameter()
4184 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4185 c.subscribe(namespace, track_name, vec![filter])
4186 .await
4187 .map(AnyRequest::Draft22)
4188 .map_err(AnyConnectionError::from)
4189 }
4190 #[cfg(feature = "draft23")]
4191 Self::Draft23(c) => {
4192 let filter = range
4193 .location_filter_draft23()?
4194 .parameter()
4195 .map_err(|e| AnyConnectionError::facade(e.to_string()))?;
4196 c.subscribe(namespace, track_name, vec![filter])
4197 .await
4198 .map(AnyRequest::Draft23)
4199 .map_err(AnyConnectionError::from)
4200 }
4201 #[allow(unreachable_patterns)]
4202 other => Err(AnyConnectionError::facade(format!(
4203 "subscribe_range: not yet wired up for draft {:?} via AnyConnection",
4204 other.draft()
4205 ))),
4206 }
4207 }
4208
4209 /// Send a standalone FETCH for `range`. Wired on **every draft this build
4210 /// carries**.
4211 ///
4212 /// # What the range means here
4213 ///
4214 /// **[`FetchEnd::Object`] is the last Object the fetch covers, and the
4215 /// range holds it. [`FetchEnd::EntireGroup`] covers the whole end Group.
4216 /// `end_group` is absolute.** That is the whole contract, and it is stated
4217 /// in [`FetchEnd`] as well because it is the one thing about this call a
4218 /// caller can get wrong without being told.
4219 ///
4220 /// The drafts do not agree, which is why the argument is a
4221 /// [`FetchRange`] rather than four numbers. Drafts 07 through 19 carry the
4222 /// start and the end inline in FETCH and all thirteen word the end alike —
4223 /// draft-19 Section 10.12.1: "The end Location, plus 1. A Location.Object
4224 /// value of 0 means the entire group is requested." The section number
4225 /// moves between drafts and the fields are regrouped into a `Location`
4226 /// along the way; the sentence does not change, which is why
4227 /// `moqtap_codec::types::check_location_range` is one shared function
4228 /// rather than thirteen.
4229 /// Draft-20 Section 10.13 deleted both fields and
4230 /// moved the range into the `LOCATION_FILTER` parameter, whose ranges
4231 /// Section 5.1.2 calls **inclusive** — the `+ 1` and the `0`-means-whole-
4232 /// group convention are both gone, and neither deletion is in the draft's
4233 /// own change log. A single `end_object: u64` at this boundary would have
4234 /// meant one of those two things and looked like the other.
4235 ///
4236 /// The conversion is [`FetchRange::inline_end_object`] for the first group
4237 /// and [`FetchRange::location_filter_draft20`] for draft-20. **The `+ 1` exists in
4238 /// exactly one place**, the first of those, so it cannot reach draft-20 by
4239 /// being ported.
4240 ///
4241 /// # What this does not carry
4242 ///
4243 /// An empty parameter list, on every draft that has one — drafts 07 through
4244 /// 11 have no parameter field on FETCH at all. Drafts 07 through 14 carry
4245 /// subscriber priority and group order as fields of FETCH and no later
4246 /// draft does, so those arms send the defaults rather than widening an
4247 /// entry point shared with six drafts that have no such fields. Reach a
4248 /// draft's own `Connection::fetch` through the variant for anything past a
4249 /// plain range.
4250 ///
4251 /// # Errors
4252 ///
4253 /// A range the negotiated draft cannot express, before anything is written:
4254 /// `FetchEnd::Object(u64::MAX)` on drafts 14 through 19, and an `end_group`
4255 /// below `start_group` on draft-20. See the two conversions for why each is
4256 /// inexpressible rather than merely unusual.
4257 ///
4258 /// The returned [`AnyRequest`] must be held while the request is live: on
4259 /// drafts 17-23 it owns the bidirectional stream the request went out on
4260 /// and dropping it cancels the fetch.
4261 #[allow(unused_variables)]
4262 pub async fn fetch(
4263 &mut self,
4264 namespace: moqtap_codec::types::TrackNamespace,
4265 track_name: Vec<u8>,
4266 range: FetchRange,
4267 ) -> Result<AnyRequest, AnyConnectionError> {
4268 // The three fields drafts 14 through 19 carry unchanged, built once for
4269 // whichever arm runs. `from_u64_moqt` rather than `from_u64` because
4270 // MoQT's varint reaches the full 64-bit range (Section 1.4.1) and the
4271 // newtype *is* the value: a `VarInt` a caller could have handed the old
4272 // four-argument form is the same `VarInt` this makes, so every draft
4273 // that was wired before writes the bytes it wrote before. The fourth
4274 // field is the one that differs, and it is built inside each arm from
4275 // `inline_end_object`.
4276 let start_group = moqtap_codec::varint::VarInt::from_u64_moqt(range.start_group);
4277 let start_object = moqtap_codec::varint::VarInt::from_u64_moqt(range.start_object);
4278 let end_group = moqtap_codec::varint::VarInt::from_u64_moqt(range.end_group);
4279 // The fourth field, which is the one that differs: draft-19 Section
4280 // 10.13 makes `End Location.Object` the last Object plus 1, or 0 for
4281 // the whole Group. Built here so the six arms that need it stay
4282 // identical, and lazily so that draft-20 — which can express a range
4283 // those six cannot — is not refused on their behalf.
4284 let inline_end_object = || -> Result<moqtap_codec::varint::VarInt, AnyConnectionError> {
4285 Ok(moqtap_codec::varint::VarInt::from_u64_moqt(range.inline_end_object()?))
4286 };
4287 let draft = self.draft();
4288 match self {
4289 // Drafts 07 through 11 take the four location fields and nothing
4290 // else: FETCH gained its parameter block in draft-12. The
4291 // priority and the order are fields of FETCH on every draft up to
4292 // and including 14 and of no later draft's, so these arms send the
4293 // defaults rather than widening an entry point shared with six
4294 // drafts that have no such fields.
4295 #[cfg(feature = "draft07")]
4296 Self::Draft07(c) => c
4297 .fetch(
4298 namespace,
4299 track_name,
4300 128,
4301 moqtap_codec::types::GroupOrder::Ascending,
4302 start_group,
4303 start_object,
4304 end_group,
4305 inline_end_object()?,
4306 )
4307 .await
4308 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4309 .map_err(AnyConnectionError::from),
4310 #[cfg(feature = "draft08")]
4311 Self::Draft08(c) => c
4312 .fetch(
4313 namespace,
4314 track_name,
4315 128,
4316 moqtap_codec::types::GroupOrder::Ascending,
4317 start_group,
4318 start_object,
4319 end_group,
4320 inline_end_object()?,
4321 )
4322 .await
4323 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4324 .map_err(AnyConnectionError::from),
4325 #[cfg(feature = "draft09")]
4326 Self::Draft09(c) => c
4327 .fetch(
4328 namespace,
4329 track_name,
4330 128,
4331 moqtap_codec::types::GroupOrder::Ascending,
4332 start_group,
4333 start_object,
4334 end_group,
4335 inline_end_object()?,
4336 )
4337 .await
4338 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4339 .map_err(AnyConnectionError::from),
4340 #[cfg(feature = "draft10")]
4341 Self::Draft10(c) => c
4342 .fetch(
4343 namespace,
4344 track_name,
4345 128,
4346 moqtap_codec::types::GroupOrder::Ascending,
4347 start_group,
4348 start_object,
4349 end_group,
4350 inline_end_object()?,
4351 )
4352 .await
4353 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4354 .map_err(AnyConnectionError::from),
4355 #[cfg(feature = "draft11")]
4356 Self::Draft11(c) => c
4357 .fetch(
4358 namespace,
4359 track_name,
4360 128,
4361 moqtap_codec::types::GroupOrder::Ascending,
4362 start_group,
4363 start_object,
4364 end_group,
4365 inline_end_object()?,
4366 )
4367 .await
4368 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4369 .map_err(AnyConnectionError::from),
4370 #[cfg(feature = "draft12")]
4371 Self::Draft12(c) => c
4372 .fetch(
4373 namespace,
4374 track_name,
4375 128,
4376 moqtap_codec::types::GroupOrder::Ascending,
4377 start_group,
4378 start_object,
4379 end_group,
4380 inline_end_object()?,
4381 Vec::new(),
4382 )
4383 .await
4384 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4385 .map_err(AnyConnectionError::from),
4386 #[cfg(feature = "draft13")]
4387 Self::Draft13(c) => c
4388 .fetch(
4389 namespace,
4390 track_name,
4391 128,
4392 moqtap_codec::types::GroupOrder::Ascending,
4393 start_group,
4394 start_object,
4395 end_group,
4396 inline_end_object()?,
4397 Vec::new(),
4398 )
4399 .await
4400 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4401 .map_err(AnyConnectionError::from),
4402 #[cfg(feature = "draft14")]
4403 Self::Draft14(c) => c
4404 .fetch(
4405 namespace,
4406 track_name,
4407 // The last draft that carries these two on FETCH; see the
4408 // comment above the draft-07 arm.
4409 128,
4410 moqtap_codec::types::GroupOrder::Ascending,
4411 start_group,
4412 start_object,
4413 end_group,
4414 inline_end_object()?,
4415 Vec::new(),
4416 )
4417 .await
4418 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4419 .map_err(AnyConnectionError::from),
4420 #[cfg(feature = "draft15")]
4421 Self::Draft15(c) => c
4422 .fetch(
4423 namespace,
4424 track_name,
4425 start_group,
4426 start_object,
4427 end_group,
4428 inline_end_object()?,
4429 Vec::new(),
4430 )
4431 .await
4432 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4433 .map_err(AnyConnectionError::from),
4434 #[cfg(feature = "draft16")]
4435 Self::Draft16(c) => c
4436 .fetch(
4437 namespace,
4438 track_name,
4439 start_group,
4440 start_object,
4441 end_group,
4442 inline_end_object()?,
4443 Vec::new(),
4444 )
4445 .await
4446 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4447 .map_err(AnyConnectionError::from),
4448 #[cfg(feature = "draft17")]
4449 Self::Draft17(c) => c
4450 .fetch(
4451 namespace,
4452 track_name,
4453 start_group,
4454 start_object,
4455 end_group,
4456 inline_end_object()?,
4457 Vec::new(),
4458 )
4459 .await
4460 .map(AnyRequest::Draft17)
4461 .map_err(AnyConnectionError::from),
4462 #[cfg(feature = "draft18")]
4463 Self::Draft18(c) => c
4464 .fetch(
4465 namespace,
4466 track_name,
4467 start_group,
4468 start_object,
4469 end_group,
4470 inline_end_object()?,
4471 Vec::new(),
4472 )
4473 .await
4474 .map(AnyRequest::Draft18)
4475 .map_err(AnyConnectionError::from),
4476 #[cfg(feature = "draft19")]
4477 Self::Draft19(c) => c
4478 .fetch(
4479 namespace,
4480 track_name,
4481 start_group,
4482 start_object,
4483 end_group,
4484 inline_end_object()?,
4485 Vec::new(),
4486 )
4487 .await
4488 .map(AnyRequest::Draft19)
4489 .map_err(AnyConnectionError::from),
4490 #[cfg(feature = "draft20")]
4491 Self::Draft20(c) => {
4492 // Draft-20 Section 10.13 has no location fields to fill in:
4493 // `fetch_range` puts the whole range in the `LOCATION_FILTER`
4494 // parameter, at the position ascending Parameter Type order
4495 // requires. Nothing here adds one to the end —
4496 // draft-20 Section 5.1.2 makes the filter's range inclusive,
4497 // and the `+ 1` the six arms above apply lives in
4498 // `inline_end_object` alone.
4499 let filter = range.location_filter_draft20()?;
4500 c.fetch_range(namespace, track_name, &filter, Vec::new())
4501 .await
4502 .map(AnyRequest::Draft20)
4503 .map_err(AnyConnectionError::from)
4504 }
4505 #[cfg(feature = "draft21")]
4506 Self::Draft21(c) => {
4507 // Draft-21 Section 9.11 has no location fields to fill in:
4508 // `fetch_range` puts the whole range in the `LOCATION_FILTER`
4509 // parameter, at the position ascending Parameter Type order
4510 // requires. Nothing here adds one to the end —
4511 // draft-21 Section 3.3.1 makes the filter's range inclusive,
4512 // and the `+ 1` the six arms above apply lives in
4513 // `inline_end_object` alone.
4514 let filter = range.location_filter_draft21()?;
4515 c.fetch_range(namespace, track_name, &filter, Vec::new())
4516 .await
4517 .map(AnyRequest::Draft21)
4518 .map_err(AnyConnectionError::from)
4519 }
4520 #[cfg(feature = "draft22")]
4521 Self::Draft22(c) => {
4522 // Draft-22 Section 9.11 has no location fields to fill in:
4523 // `fetch_range` puts the whole range in the `LOCATION_FILTER`
4524 // parameter, at the position ascending Parameter Type order
4525 // requires. Nothing here adds one to the end —
4526 // draft-22 Section 3.3.1 makes the filter's range inclusive,
4527 // and the `+ 1` the six arms above apply lives in
4528 // `inline_end_object` alone.
4529 let filter = range.location_filter_draft22()?;
4530 c.fetch_range(namespace, track_name, &filter, Vec::new())
4531 .await
4532 .map(AnyRequest::Draft22)
4533 .map_err(AnyConnectionError::from)
4534 }
4535 #[cfg(feature = "draft23")]
4536 Self::Draft23(c) => {
4537 // Draft-23 Section 9.11 has no location fields to fill in:
4538 // `fetch_range` puts the whole range in the `LOCATION_FILTER`
4539 // parameter, at the position ascending Parameter Type order
4540 // requires. Nothing here adds one to the end —
4541 // draft-23 Section 3.3.1 makes the filter's range inclusive,
4542 // and the `+ 1` the six arms above apply lives in
4543 // `inline_end_object` alone.
4544 let filter = range.location_filter_draft23()?;
4545 c.fetch_range(namespace, track_name, &filter, Vec::new())
4546 .await
4547 .map(AnyRequest::Draft23)
4548 .map_err(AnyConnectionError::from)
4549 }
4550 #[allow(unreachable_patterns)]
4551 other => Err(AnyConnectionError::facade(format!(
4552 "fetch: not yet wired up for draft {:?} via AnyConnection",
4553 other.draft()
4554 ))),
4555 }
4556 }
4557
4558 /// Send a **Joining** FETCH against a subscription this session already
4559 /// holds. Drafts 08 through 19.
4560 ///
4561 /// [`AnyConnection::fetch`]'s other half, and a different question rather
4562 /// than a shorthand for the same one. A standalone FETCH names a track and
4563 /// a range and asks a relay's store for it. A Joining FETCH names **a
4564 /// subscription** and asks for the part of that subscription's track that
4565 /// precedes it, and the publisher fills in the namespace, the name and the
4566 /// end from the subscription itself. So a subscriber that wants "what I am
4567 /// watching, plus the run-up to it" has one request for the pair instead of
4568 /// a subscription and a fetch whose range it had to compute — and, on a
4569 /// live track, could not compute, because the run-up ends wherever the
4570 /// subscription happened to start.
4571 ///
4572 /// `joining_request_id` is the Request ID of that subscription, which is
4573 /// [`AnyRequest::request_id`] on the handle [`AnyConnection::subscribe`]
4574 /// returned. Nothing here checks that it names one: draft-19 Section
4575 /// 10.12.2 puts that check at the publisher — "it MUST respond with a Fetch
4576 /// Error with code Invalid Joining Request ID" — and this side is the
4577 /// subscriber. The subscription must still be live when the FETCH
4578 /// *arrives*, which is a thing about ordering rather than about this call.
4579 ///
4580 /// `start` is [`JoiningStart`], and it carries the Fetch Type as well as
4581 /// the number for the reason that type documents.
4582 ///
4583 /// # The two drafts at the ends, and what each of them deleted
4584 ///
4585 /// **Draft-07 has no Joining Fetch at all.** Its FETCH has no Fetch Type
4586 /// field, so there is no bit in the message that could ask for one; the
4587 /// field and the second Fetch Type both arrive in draft-08.
4588 ///
4589 /// **Draft-20 deleted the whole mechanism** — Section 10.13 removed the
4590 /// Fetch Type field, both payload structures and the Fetch Type registry
4591 /// together, and promoted the namespace and the name to fields of FETCH
4592 /// itself. There is no joining form to fall back to and no parameter that
4593 /// restores one, so this refuses on draft-20 rather than sending something
4594 /// adjacent. **This is the entry point that makes "one suite run per draft"
4595 /// concrete**: a relay speaking both 14 and 20 answers this question on one
4596 /// of them and cannot be asked it on the other, and a probe that tested only
4597 /// the newest draft would never learn that the relay implements it.
4598 ///
4599 /// Drafts 08 through 10 carry only the relative form, so
4600 /// [`JoiningStart::Group`] is refused there — see that variant.
4601 ///
4602 /// # What this does not carry
4603 ///
4604 /// The same three things [`AnyConnection::fetch`] leaves out, for the same
4605 /// reasons: an empty parameter list on every draft that has one (drafts 08
4606 /// through 11 have no parameter field on FETCH), and the subscriber
4607 /// priority and group order as defaults on drafts 08 through 14, which are
4608 /// the only drafts carrying them as fields of FETCH.
4609 ///
4610 /// The returned [`AnyRequest`] must be held while the fetch is live: on
4611 /// drafts 17 through 19 it owns the bidirectional stream the request went
4612 /// out on and dropping it cancels the fetch.
4613 #[allow(unused_variables)]
4614 pub async fn fetch_joining(
4615 &mut self,
4616 joining_request_id: moqtap_codec::varint::VarInt,
4617 start: JoiningStart,
4618 ) -> Result<AnyRequest, AnyConnectionError> {
4619 let draft = self.draft();
4620 let joining_start = moqtap_codec::varint::VarInt::from_u64_moqt(match start {
4621 JoiningStart::GroupsBefore(n) => n,
4622 JoiningStart::Group(g) => g,
4623 });
4624 // Refused once, here, rather than in three arms: drafts 08 through 10
4625 // define Fetch Types 0x1 and 0x2 and nothing else, and sending the
4626 // relative form with an absolute number in it would ask for the last
4627 // `g` Groups of the track when the caller asked for everything from
4628 // Group `g`.
4629 // The return type is spelled out because a single-draft build can
4630 // compile this closure without ever calling it: with only draft-20
4631 // enabled every arm that would have pinned `T` is gone, and inference
4632 // has nothing left to work from. Annotating it keeps `just draft-matrix`
4633 // green on all single-draft rows.
4634 let absolute_unavailable = || -> Result<AnyRequest, AnyConnectionError> {
4635 Err(AnyConnectionError::facade(format!(
4636 "fetch_joining: draft {draft:?} has no Absolute Joining Fetch — its FETCH offers \
4637 Fetch Types 0x1 and 0x2, and 0x3 arrives in draft-11. Ask relatively, or reach \
4638 the draft's own Connection::joining_fetch"
4639 )))
4640 };
4641 // The two fields drafts 08 through 14 carry on FETCH and no later draft
4642 // does. Sent as defaults for `fetch`'s reason: widening an entry point
4643 // shared with five drafts that have no such fields would give a caller
4644 // two arguments that are silently dropped on more than a third of them.
4645 //
4646 // Both are read only by the draft-08 through draft-14 arms, so a build
4647 // that enables none of those seven — `--features draft07,draft22` is
4648 // the pair `just draft-pairs` picks — compiles them and never reaches
4649 // them. Allowing the dead code is preferred to a seven-feature `cfg`
4650 // here: the `cfg` would have to be repeated and kept in step with the
4651 // arms below, and the day a draft joined or left that range the two
4652 // would drift apart silently, where an unused constant costs nothing
4653 // and cannot be wrong.
4654 #[allow(dead_code)]
4655 const PRIORITY: u8 = 128;
4656 #[allow(unused_variables)]
4657 let order = moqtap_codec::types::GroupOrder::Ascending;
4658 match self {
4659 // Draft-07's FETCH has no Fetch Type field, so this is not a gap to
4660 // be filled later — there is nothing in the message to set.
4661 #[cfg(feature = "draft07")]
4662 Self::Draft07(_) => Err(AnyConnectionError::facade(
4663 "fetch_joining: draft-07's FETCH has no Fetch Type field and no Joining Fetch; \
4664 draft-08 is where both arrive",
4665 )),
4666 #[cfg(feature = "draft08")]
4667 Self::Draft08(c) => match start {
4668 JoiningStart::Group(_) => absolute_unavailable(),
4669 JoiningStart::GroupsBefore(_) => c
4670 .joining_fetch(PRIORITY, order, joining_request_id, joining_start)
4671 .await
4672 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4673 .map_err(AnyConnectionError::from),
4674 },
4675 #[cfg(feature = "draft09")]
4676 Self::Draft09(c) => match start {
4677 JoiningStart::Group(_) => absolute_unavailable(),
4678 JoiningStart::GroupsBefore(_) => c
4679 .joining_fetch(PRIORITY, order, joining_request_id, joining_start)
4680 .await
4681 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4682 .map_err(AnyConnectionError::from),
4683 },
4684 #[cfg(feature = "draft10")]
4685 Self::Draft10(c) => match start {
4686 JoiningStart::Group(_) => absolute_unavailable(),
4687 JoiningStart::GroupsBefore(_) => c
4688 .joining_fetch(PRIORITY, order, joining_request_id, joining_start)
4689 .await
4690 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4691 .map_err(AnyConnectionError::from),
4692 },
4693 // Draft-11 splits Joining into the relative and absolute pair and
4694 // has no parameter block on FETCH; draft-12 adds the block.
4695 #[cfg(feature = "draft11")]
4696 Self::Draft11(c) => match start {
4697 JoiningStart::GroupsBefore(_) => {
4698 c.joining_fetch(PRIORITY, order, joining_request_id, joining_start).await
4699 }
4700 JoiningStart::Group(_) => {
4701 c.absolute_joining_fetch(PRIORITY, order, joining_request_id, joining_start)
4702 .await
4703 }
4704 }
4705 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4706 .map_err(AnyConnectionError::from),
4707 #[cfg(feature = "draft12")]
4708 Self::Draft12(c) => match start {
4709 JoiningStart::GroupsBefore(_) => {
4710 c.joining_fetch(PRIORITY, order, joining_request_id, joining_start, Vec::new())
4711 .await
4712 }
4713 JoiningStart::Group(_) => {
4714 c.absolute_joining_fetch(
4715 PRIORITY,
4716 order,
4717 joining_request_id,
4718 joining_start,
4719 Vec::new(),
4720 )
4721 .await
4722 }
4723 }
4724 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4725 .map_err(AnyConnectionError::from),
4726 #[cfg(feature = "draft13")]
4727 Self::Draft13(c) => match start {
4728 JoiningStart::GroupsBefore(_) => {
4729 c.joining_fetch(PRIORITY, order, joining_request_id, joining_start, Vec::new())
4730 .await
4731 }
4732 JoiningStart::Group(_) => {
4733 c.absolute_joining_fetch(
4734 PRIORITY,
4735 order,
4736 joining_request_id,
4737 joining_start,
4738 Vec::new(),
4739 )
4740 .await
4741 }
4742 }
4743 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4744 .map_err(AnyConnectionError::from),
4745 // The last draft carrying the priority and the order on FETCH; see
4746 // the constants above.
4747 #[cfg(feature = "draft14")]
4748 Self::Draft14(c) => match start {
4749 JoiningStart::GroupsBefore(_) => {
4750 c.joining_fetch(PRIORITY, order, joining_request_id, joining_start, Vec::new())
4751 .await
4752 }
4753 JoiningStart::Group(_) => {
4754 c.absolute_joining_fetch(
4755 PRIORITY,
4756 order,
4757 joining_request_id,
4758 joining_start,
4759 Vec::new(),
4760 )
4761 .await
4762 }
4763 }
4764 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4765 .map_err(AnyConnectionError::from),
4766 // From draft-15 the two fields are gone from FETCH and the request
4767 // is three varints and a parameter block.
4768 #[cfg(feature = "draft15")]
4769 Self::Draft15(c) => match start {
4770 JoiningStart::GroupsBefore(_) => {
4771 c.joining_fetch(joining_request_id, joining_start, Vec::new()).await
4772 }
4773 JoiningStart::Group(_) => {
4774 c.absolute_joining_fetch(joining_request_id, joining_start, Vec::new()).await
4775 }
4776 }
4777 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4778 .map_err(AnyConnectionError::from),
4779 #[cfg(feature = "draft16")]
4780 Self::Draft16(c) => match start {
4781 JoiningStart::GroupsBefore(_) => {
4782 c.joining_fetch(joining_request_id, joining_start, Vec::new()).await
4783 }
4784 JoiningStart::Group(_) => {
4785 c.absolute_joining_fetch(joining_request_id, joining_start, Vec::new()).await
4786 }
4787 }
4788 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4789 .map_err(AnyConnectionError::from),
4790 // From draft-17 the request travels on a bidirectional stream of its
4791 // own, so the handle owns that stream rather than naming an id on a
4792 // shared one.
4793 #[cfg(feature = "draft17")]
4794 Self::Draft17(c) => match start {
4795 JoiningStart::GroupsBefore(_) => {
4796 c.joining_fetch(joining_request_id, joining_start, Vec::new()).await
4797 }
4798 JoiningStart::Group(_) => {
4799 c.absolute_joining_fetch(joining_request_id, joining_start, Vec::new()).await
4800 }
4801 }
4802 .map(AnyRequest::Draft17)
4803 .map_err(AnyConnectionError::from),
4804 #[cfg(feature = "draft18")]
4805 Self::Draft18(c) => match start {
4806 JoiningStart::GroupsBefore(_) => {
4807 c.joining_fetch(joining_request_id, joining_start, Vec::new()).await
4808 }
4809 JoiningStart::Group(_) => {
4810 c.absolute_joining_fetch(joining_request_id, joining_start, Vec::new()).await
4811 }
4812 }
4813 .map(AnyRequest::Draft18)
4814 .map_err(AnyConnectionError::from),
4815 #[cfg(feature = "draft19")]
4816 Self::Draft19(c) => match start {
4817 JoiningStart::GroupsBefore(_) => {
4818 c.joining_fetch(joining_request_id, joining_start, Vec::new()).await
4819 }
4820 JoiningStart::Group(_) => {
4821 c.absolute_joining_fetch(joining_request_id, joining_start, Vec::new()).await
4822 }
4823 }
4824 .map(AnyRequest::Draft19)
4825 .map_err(AnyConnectionError::from),
4826 // Not a gap either. Draft-20 Section 10.13 deleted the Fetch Type
4827 // field, both Fetch payload structures and the Fetch Type registry
4828 // in one rewrite, and its change log does not mention the joining
4829 // mechanism at all.
4830 #[cfg(feature = "draft20")]
4831 Self::Draft20(_) => Err(AnyConnectionError::facade(
4832 "fetch_joining: draft-20 deleted the Fetch Type field and the whole Joining Fetch \
4833 mechanism (Section 10.13); a fetch there names its own track and range, which is \
4834 AnyConnection::fetch",
4835 )),
4836 #[cfg(feature = "draft21")]
4837 Self::Draft21(_) => Err(AnyConnectionError::facade(
4838 "fetch_joining: draft-20 deleted the Fetch Type field and the whole Joining Fetch \
4839 mechanism and draft-21 keeps it gone (Section 9.11); a fetch there names its own \
4840 track and range, which is AnyConnection::fetch",
4841 )),
4842 #[cfg(feature = "draft22")]
4843 Self::Draft22(_) => Err(AnyConnectionError::facade(
4844 "fetch_joining: draft-20 deleted the Fetch Type field and the whole Joining Fetch \
4845 mechanism and draft-22 keeps it gone (Section 9.11); a fetch there names its own \
4846 track and range, which is AnyConnection::fetch",
4847 )),
4848 #[cfg(feature = "draft23")]
4849 Self::Draft23(_) => Err(AnyConnectionError::facade(
4850 "fetch_joining: draft-23 has no Joining Fetch (Section 9.11); a fetch there names \
4851 its own track and range, which is AnyConnection::fetch",
4852 )),
4853 #[allow(unreachable_patterns)]
4854 other => Err(AnyConnectionError::facade(format!(
4855 "fetch_joining: not yet wired up for draft {:?} via AnyConnection",
4856 other.draft()
4857 ))),
4858 }
4859 }
4860
4861 /// Send a TRACK_STATUS query for the given track. Supported on drafts 11
4862 /// through 22. From draft-15 on, passes an empty parameter list.
4863 ///
4864 /// # Which message this sends, and why the name moved
4865 ///
4866 /// Up to draft-12 the query is TRACK_STATUS_REQUEST and `track_status` is
4867 /// the *response* to it; draft-13 renamed the request to TRACK_STATUS and
4868 /// gave the answer its own TRACK_STATUS_OK. So this method reaches
4869 /// `Connection::track_status_request` on drafts 11 and 12 and
4870 /// `Connection::track_status` from draft-13 on — **the same question under
4871 /// two names, not the same name for two things.** Calling the
4872 /// same-named method on a draft-11 `Connection` would send a reply to a
4873 /// question nobody asked, which is why the naming test
4874 /// `an_unsent_request_is_not_named_from_the_drafts_table` exists.
4875 ///
4876 /// # Drafts 07 through 10
4877 ///
4878 /// Not supported here, and not for want of a match arm. Their
4879 /// TRACK_STATUS_REQUEST carries **no Request ID** — `Connection::
4880 /// track_status_request` returns `()` on those drafts, because the answer
4881 /// is matched by track namespace and name rather than by an identifier.
4882 /// [`AnyRequest`] is a handle to a request the endpoint numbered, so there
4883 /// is nothing for this method to return, and fabricating an ID would put a
4884 /// number in [`AnyRequest::request_id`] that was never on the wire. Send
4885 /// the query through the variant's own `Connection` and read the reply with
4886 /// [`AnyConnection::recv_response`] against any other outstanding request,
4887 /// or match it by name off `recv_and_dispatch`.
4888 ///
4889 /// The returned [`AnyRequest`] must be held until the answer arrives: on
4890 /// drafts 17-23 it owns the bidirectional stream the query went out on and
4891 /// dropping it cancels the query.
4892 #[allow(unused_variables)]
4893 pub async fn track_status(
4894 &mut self,
4895 namespace: moqtap_codec::types::TrackNamespace,
4896 track_name: Vec<u8>,
4897 ) -> Result<AnyRequest, AnyConnectionError> {
4898 let draft = self.draft();
4899 match self {
4900 // Drafts 11 and 12: TRACK_STATUS_REQUEST, which is this question's
4901 // name before draft-13 renames it. Both drafts carry a parameter
4902 // block — it arrives with draft-11 — and the two arms differ only
4903 // because this build's draft-11 entry point takes no parameter list
4904 // of its own and sends an empty one.
4905 #[cfg(feature = "draft11")]
4906 Self::Draft11(c) => c
4907 .track_status_request(namespace, track_name)
4908 .await
4909 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4910 .map_err(AnyConnectionError::from),
4911 #[cfg(feature = "draft12")]
4912 Self::Draft12(c) => c
4913 .track_status_request(namespace, track_name, Vec::new())
4914 .await
4915 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4916 .map_err(AnyConnectionError::from),
4917 // Draft-13 is where the request takes the name and the four
4918 // SUBSCRIBE-shaped fields, exactly as draft-14 carries them.
4919 #[cfg(feature = "draft13")]
4920 Self::Draft13(c) => c
4921 .track_status(
4922 namespace,
4923 track_name,
4924 128,
4925 moqtap_codec::types::GroupOrder::Ascending,
4926 moqtap_codec::types::Forward::Forward,
4927 moqtap_codec::types::FilterType::LargestObject,
4928 Vec::new(),
4929 )
4930 .await
4931 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4932 .map_err(AnyConnectionError::from),
4933 #[cfg(feature = "draft14")]
4934 Self::Draft14(c) => c
4935 .track_status(
4936 namespace,
4937 track_name,
4938 // Drafts 13 and 14 word TRACK_STATUS like a SUBSCRIBE and
4939 // no later draft does, so these four are sent as they
4940 // always were rather than widening an entry point shared
4941 // with six drafts that have no such fields.
4942 128,
4943 moqtap_codec::types::GroupOrder::Ascending,
4944 moqtap_codec::types::Forward::Forward,
4945 moqtap_codec::types::FilterType::LargestObject,
4946 Vec::new(),
4947 )
4948 .await
4949 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4950 .map_err(AnyConnectionError::from),
4951 #[cfg(feature = "draft15")]
4952 Self::Draft15(c) => c
4953 .track_status(namespace, track_name, Vec::new())
4954 .await
4955 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4956 .map_err(AnyConnectionError::from),
4957 #[cfg(feature = "draft16")]
4958 Self::Draft16(c) => c
4959 .track_status(namespace, track_name, Vec::new())
4960 .await
4961 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
4962 .map_err(AnyConnectionError::from),
4963 #[cfg(feature = "draft17")]
4964 Self::Draft17(c) => c
4965 .track_status(namespace, track_name, Vec::new())
4966 .await
4967 .map(AnyRequest::Draft17)
4968 .map_err(AnyConnectionError::from),
4969 #[cfg(feature = "draft18")]
4970 Self::Draft18(c) => c
4971 .track_status(namespace, track_name, Vec::new())
4972 .await
4973 .map(AnyRequest::Draft18)
4974 .map_err(AnyConnectionError::from),
4975 #[cfg(feature = "draft19")]
4976 Self::Draft19(c) => c
4977 .track_status(namespace, track_name, Vec::new())
4978 .await
4979 .map(AnyRequest::Draft19)
4980 .map_err(AnyConnectionError::from),
4981 #[cfg(feature = "draft20")]
4982 Self::Draft20(c) => c
4983 .track_status(namespace, track_name, Vec::new())
4984 .await
4985 .map(AnyRequest::Draft20)
4986 .map_err(AnyConnectionError::from),
4987 #[cfg(feature = "draft21")]
4988 Self::Draft21(c) => c
4989 .track_status(namespace, track_name, Vec::new())
4990 .await
4991 .map(AnyRequest::Draft21)
4992 .map_err(AnyConnectionError::from),
4993 #[cfg(feature = "draft22")]
4994 Self::Draft22(c) => c
4995 .track_status(namespace, track_name, Vec::new())
4996 .await
4997 .map(AnyRequest::Draft22)
4998 .map_err(AnyConnectionError::from),
4999 #[cfg(feature = "draft23")]
5000 Self::Draft23(c) => c
5001 .track_status(namespace, track_name, Vec::new())
5002 .await
5003 .map(AnyRequest::Draft23)
5004 .map_err(AnyConnectionError::from),
5005 #[allow(unreachable_patterns)]
5006 other => Err(AnyConnectionError::facade(format!(
5007 "track_status: not yet wired up for draft {:?} via AnyConnection",
5008 other.draft()
5009 ))),
5010 }
5011 }
5012
5013 /// Send a SUBSCRIBE_NAMESPACE (or SUBSCRIBE_ANNOUNCES on drafts 11–12).
5014 /// Supported on drafts 11 through 23. Drafts 16 and 17 pass default
5015 /// subscribe options; every draft from 12 on passes an empty parameter
5016 /// list.
5017 ///
5018 /// From draft-18 this is the renumbered SUBSCRIBE_NAMESPACE (0x50), which
5019 /// asks for NAMESPACE and NAMESPACE_DONE only. SUBSCRIBE_TRACKS, the other
5020 /// half of the draft-18 split, is [`AnyConnection::subscribe_tracks`].
5021 ///
5022 /// The returned [`AnyRequest`] must be held while the request is live: on
5023 /// drafts 17-23 it owns the bidirectional stream the request went out on
5024 /// and dropping it cancels the namespace subscription.
5025 #[allow(unused_variables)]
5026 pub async fn subscribe_namespace(
5027 &mut self,
5028 namespace_prefix: moqtap_codec::types::TrackNamespace,
5029 ) -> Result<AnyRequest, AnyConnectionError> {
5030 // Only the draft-16 and draft-17 arms below build a subscribe-options
5031 // varint; the other drafts' wrappers take no such argument, so the
5032 // import is dead outside those two builds.
5033 #[cfg(any(feature = "draft16", feature = "draft17"))]
5034 use moqtap_codec::varint::VarInt;
5035 let draft = self.draft();
5036 match self {
5037 #[cfg(feature = "draft11")]
5038 Self::Draft11(c) => c
5039 .subscribe_announces(namespace_prefix)
5040 .await
5041 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5042 .map_err(AnyConnectionError::from),
5043 #[cfg(feature = "draft12")]
5044 Self::Draft12(c) => c
5045 .subscribe_announces(namespace_prefix, Vec::new())
5046 .await
5047 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5048 .map_err(AnyConnectionError::from),
5049 #[cfg(feature = "draft13")]
5050 Self::Draft13(c) => c
5051 .subscribe_namespace(namespace_prefix, Vec::new())
5052 .await
5053 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5054 .map_err(AnyConnectionError::from),
5055 #[cfg(feature = "draft14")]
5056 Self::Draft14(c) => c
5057 .subscribe_namespace(namespace_prefix, Vec::new())
5058 .await
5059 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5060 .map_err(AnyConnectionError::from),
5061 #[cfg(feature = "draft15")]
5062 Self::Draft15(c) => c
5063 .subscribe_namespace(namespace_prefix, Vec::new())
5064 .await
5065 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5066 .map_err(AnyConnectionError::from),
5067 #[cfg(feature = "draft16")]
5068 Self::Draft16(c) => {
5069 let opts = VarInt::from_u64(0).expect("0 fits in VarInt");
5070 c.subscribe_namespace(namespace_prefix, opts, Vec::new())
5071 .await
5072 .map(AnyRequest::Draft16)
5073 .map_err(AnyConnectionError::from)
5074 }
5075 #[cfg(feature = "draft17")]
5076 Self::Draft17(c) => {
5077 let opts = VarInt::from_u64(0).expect("0 fits in VarInt");
5078 c.subscribe_namespace(namespace_prefix, opts, Vec::new())
5079 .await
5080 .map(AnyRequest::Draft17)
5081 .map_err(AnyConnectionError::from)
5082 }
5083 #[cfg(feature = "draft18")]
5084 Self::Draft18(c) => c
5085 .subscribe_namespace(namespace_prefix, Vec::new())
5086 .await
5087 .map(AnyRequest::Draft18)
5088 .map_err(AnyConnectionError::from),
5089 #[cfg(feature = "draft19")]
5090 Self::Draft19(c) => c
5091 .subscribe_namespace(namespace_prefix, Vec::new())
5092 .await
5093 .map(AnyRequest::Draft19)
5094 .map_err(AnyConnectionError::from),
5095 #[cfg(feature = "draft20")]
5096 Self::Draft20(c) => c
5097 .subscribe_namespace(namespace_prefix, Vec::new())
5098 .await
5099 .map(AnyRequest::Draft20)
5100 .map_err(AnyConnectionError::from),
5101 #[cfg(feature = "draft21")]
5102 Self::Draft21(c) => c
5103 .subscribe_namespace(namespace_prefix, Vec::new())
5104 .await
5105 .map(AnyRequest::Draft21)
5106 .map_err(AnyConnectionError::from),
5107 #[cfg(feature = "draft22")]
5108 Self::Draft22(c) => c
5109 .subscribe_namespace(namespace_prefix, Vec::new())
5110 .await
5111 .map(AnyRequest::Draft22)
5112 .map_err(AnyConnectionError::from),
5113 #[cfg(feature = "draft23")]
5114 Self::Draft23(c) => c
5115 .subscribe_namespace(namespace_prefix, Vec::new())
5116 .await
5117 .map(AnyRequest::Draft23)
5118 .map_err(AnyConnectionError::from),
5119 #[allow(unreachable_patterns)]
5120 other => Err(AnyConnectionError::facade(format!(
5121 "subscribe_namespace: not yet wired up for draft {:?} via AnyConnection",
5122 other.draft()
5123 ))),
5124 }
5125 }
5126
5127 /// Offer a namespace to the peer: PUBLISH_NAMESPACE, or ANNOUNCE on the
5128 /// drafts that called it that. Supported on drafts 11 through 23. From
5129 /// draft-12 on, passes an empty parameter list.
5130 ///
5131 /// This is the first entry point here that asks the peer to **hold state**
5132 /// rather than to answer a question and forget it. A relay that accepts it
5133 /// records this session as the publisher for that namespace and will route
5134 /// matching subscriptions back down this connection, so a caller that
5135 /// announces owes the peer either a withdrawal — see
5136 /// [`AnyConnection::publish_namespace_done`] — or a closed session.
5137 ///
5138 /// # One request under two names
5139 ///
5140 /// Drafts 07 through 13 call it ANNOUNCE; draft-14 renamed it
5141 /// PUBLISH_NAMESPACE and every later draft keeps that name. The rename is
5142 /// the whole of the difference — the same namespace goes out and the same
5143 /// acceptance comes back — so this is one method rather than two, on the
5144 /// same reasoning as [`AnyConnection::track_status`], where the *request*
5145 /// changed names in the other direction.
5146 ///
5147 /// # Drafts 07 through 10
5148 ///
5149 /// Not supported here, and for the reason drafts 07 through 10 have no
5150 /// [`AnyConnection::track_status`] either: their ANNOUNCE carries **no
5151 /// Request ID**. `Connection::announce` returns `()` on those four drafts
5152 /// because ANNOUNCE_OK is matched by track namespace rather than by an
5153 /// identifier, so there is nothing for this method to hand back and a
5154 /// fabricated ID would put a number in [`AnyRequest::request_id`] that was
5155 /// never on the wire. Draft-11 is where the request ID arrives, and where
5156 /// this method starts. Send the announcement through the variant's own
5157 /// `Connection` if an older draft is the target.
5158 ///
5159 /// The returned [`AnyRequest`] must be held while the announcement is live:
5160 /// on drafts 17-23 it owns the bidirectional stream the request went out on
5161 /// and dropping it withdraws the namespace, which on those drafts is the
5162 /// only way to withdraw one.
5163 #[allow(unused_variables)]
5164 pub async fn publish_namespace(
5165 &mut self,
5166 namespace: moqtap_codec::types::TrackNamespace,
5167 ) -> Result<AnyRequest, AnyConnectionError> {
5168 let draft = self.draft();
5169 match self {
5170 // Draft-11 is the first draft to number the request, and the last
5171 // to take no parameter block.
5172 #[cfg(feature = "draft11")]
5173 Self::Draft11(c) => c
5174 .announce(namespace)
5175 .await
5176 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5177 .map_err(AnyConnectionError::from),
5178 #[cfg(feature = "draft12")]
5179 Self::Draft12(c) => c
5180 .announce(namespace, Vec::new())
5181 .await
5182 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5183 .map_err(AnyConnectionError::from),
5184 #[cfg(feature = "draft13")]
5185 Self::Draft13(c) => c
5186 .announce(namespace, Vec::new())
5187 .await
5188 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5189 .map_err(AnyConnectionError::from),
5190 // Draft-14 is the rename. Nothing else about the request moved.
5191 #[cfg(feature = "draft14")]
5192 Self::Draft14(c) => c
5193 .publish_namespace(namespace, Vec::new())
5194 .await
5195 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5196 .map_err(AnyConnectionError::from),
5197 #[cfg(feature = "draft15")]
5198 Self::Draft15(c) => c
5199 .publish_namespace(namespace, Vec::new())
5200 .await
5201 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5202 .map_err(AnyConnectionError::from),
5203 // Draft-16 gave a bidirectional stream to namespace *subscriptions*
5204 // and to nothing else, so this one stays on the control stream —
5205 // which is why the handle is a `ControlPlane` here and a stream one
5206 // line down.
5207 #[cfg(feature = "draft16")]
5208 Self::Draft16(c) => c
5209 .publish_namespace(namespace, Vec::new())
5210 .await
5211 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5212 .map_err(AnyConnectionError::from),
5213 #[cfg(feature = "draft17")]
5214 Self::Draft17(c) => c
5215 .publish_namespace(namespace, Vec::new())
5216 .await
5217 .map(AnyRequest::Draft17)
5218 .map_err(AnyConnectionError::from),
5219 #[cfg(feature = "draft18")]
5220 Self::Draft18(c) => c
5221 .publish_namespace(namespace, Vec::new())
5222 .await
5223 .map(AnyRequest::Draft18)
5224 .map_err(AnyConnectionError::from),
5225 #[cfg(feature = "draft19")]
5226 Self::Draft19(c) => c
5227 .publish_namespace(namespace, Vec::new())
5228 .await
5229 .map(AnyRequest::Draft19)
5230 .map_err(AnyConnectionError::from),
5231 #[cfg(feature = "draft20")]
5232 Self::Draft20(c) => c
5233 .publish_namespace(namespace, Vec::new())
5234 .await
5235 .map(AnyRequest::Draft20)
5236 .map_err(AnyConnectionError::from),
5237 #[cfg(feature = "draft21")]
5238 Self::Draft21(c) => c
5239 .publish_namespace(namespace, Vec::new())
5240 .await
5241 .map(AnyRequest::Draft21)
5242 .map_err(AnyConnectionError::from),
5243 #[cfg(feature = "draft22")]
5244 Self::Draft22(c) => c
5245 .publish_namespace(namespace, Vec::new())
5246 .await
5247 .map(AnyRequest::Draft22)
5248 .map_err(AnyConnectionError::from),
5249 #[cfg(feature = "draft23")]
5250 Self::Draft23(c) => c
5251 .publish_namespace(namespace, Vec::new())
5252 .await
5253 .map(AnyRequest::Draft23)
5254 .map_err(AnyConnectionError::from),
5255 #[allow(unreachable_patterns)]
5256 other => Err(AnyConnectionError::facade(format!(
5257 "publish_namespace: not yet wired up for draft {:?} via AnyConnection",
5258 other.draft()
5259 ))),
5260 }
5261 }
5262
5263 /// Withdraw a namespace this session announced: PUBLISH_NAMESPACE_DONE, or
5264 /// UNANNOUNCE on the drafts that called it that. Drafts 11 through 16.
5265 ///
5266 /// # Why this takes both the handle and the namespace
5267 ///
5268 /// Because the drafts disagree about which of the two identifies the
5269 /// announcement being withdrawn, and they disagree twice:
5270 ///
5271 /// - **Drafts 11 through 15** name the namespace. UNANNOUNCE and, from
5272 /// draft-14, PUBLISH_NAMESPACE_DONE carry the tuple itself.
5273 /// - **Draft-16** names the request ID, having moved the whole message onto
5274 /// the identifier the announcement was allocated.
5275 /// - **Drafts 17 through 23** name neither, because there is no message: the
5276 /// announcement lives exactly as long as its bidirectional stream, so
5277 /// withdrawing one is [`AnyRequest::cancel`] or simply dropping the
5278 /// handle. Those drafts return an error here rather than silently doing
5279 /// nothing, on the same terms as [`AnyConnection::unsubscribe`] — it is
5280 /// not a gap waiting to be wired, there is nothing to send.
5281 ///
5282 /// A signature taking only the namespace would be wrong on draft-16 and one
5283 /// taking only the handle would be wrong on the five drafts before it.
5284 /// Taking both keeps the caller from having to know which era it is in,
5285 /// which is the entire point of this facade.
5286 ///
5287 /// `request` is borrowed rather than consumed: on drafts 11 through 16 it
5288 /// carries no stream, so the caller keeps a handle that is still good for
5289 /// [`AnyRequest::request_id`] afterwards.
5290 #[allow(unused_variables)]
5291 pub async fn publish_namespace_done(
5292 &mut self,
5293 request: &AnyRequest,
5294 namespace: moqtap_codec::types::TrackNamespace,
5295 ) -> Result<(), AnyConnectionError> {
5296 match self {
5297 #[cfg(feature = "draft11")]
5298 Self::Draft11(c) => c.unannounce(namespace).await.map_err(AnyConnectionError::from),
5299 #[cfg(feature = "draft12")]
5300 Self::Draft12(c) => c.unannounce(namespace).await.map_err(AnyConnectionError::from),
5301 #[cfg(feature = "draft13")]
5302 Self::Draft13(c) => c.unannounce(namespace).await.map_err(AnyConnectionError::from),
5303 // Draft-14 renamed UNANNOUNCE to PUBLISH_NAMESPACE_DONE and kept
5304 // the namespace on it; draft-16 swapped the namespace for the
5305 // request ID.
5306 #[cfg(feature = "draft14")]
5307 Self::Draft14(c) => {
5308 c.publish_namespace_done(namespace).await.map_err(AnyConnectionError::from)
5309 }
5310 #[cfg(feature = "draft15")]
5311 Self::Draft15(c) => {
5312 c.publish_namespace_done(namespace).await.map_err(AnyConnectionError::from)
5313 }
5314 #[cfg(feature = "draft16")]
5315 Self::Draft16(c) => c
5316 .publish_namespace_done(request.request_id())
5317 .await
5318 .map_err(AnyConnectionError::from),
5319 #[allow(unreachable_patterns)]
5320 other => Err(AnyConnectionError::facade(format!(
5321 "publish_namespace_done: draft {:?} withdraws a namespace by ending the \
5322 announcement's own stream, so there is no message to send; cancel or drop \
5323 the request handle instead",
5324 other.draft()
5325 ))),
5326 }
5327 }
5328
5329 /// Offer the peer a track: PUBLISH. Drafts 12 through 23.
5330 ///
5331 /// Like [`AnyConnection::publish_namespace`], this asks the peer to hold
5332 /// state: a relay that accepts records this session as the track's
5333 /// publisher and forwards the offer to whoever subscribed to a matching
5334 /// prefix. A caller whose offer was accepted owes the peer
5335 /// [`AnyConnection::publish_done`] or a closed session.
5336 ///
5337 /// # Why `forward` is the one choice offered
5338 ///
5339 /// It is the only property of the offer every draft can express — a field
5340 /// on drafts 12 through 14, the FORWARD parameter (key 0x10) from draft-15
5341 /// on — and it decides whether the peer should expect objects at all.
5342 /// [`Forward::DontForward`](moqtap_codec::types::Forward::DontForward)
5343 /// offers a track without promising any. Everything else is sent as the
5344 /// emptiest form its draft allows: ascending group order and no Largest
5345 /// Location on drafts 12 through 14, and no track extensions or
5346 /// properties after that.
5347 ///
5348 /// # What comes back, and where
5349 ///
5350 /// Drafts 12 through 16 carry the offer on the control stream and the
5351 /// handle is a [`AnyRequest::ControlPlane`]; the answer is PUBLISH_OK or
5352 /// PUBLISH_ERROR on 12 through 14 and PUBLISH_OK or REQUEST_ERROR on 15 and
5353 /// 16, correlated by Request ID. From draft-17 the offer owns a
5354 /// bidirectional stream, and from draft-18 the acceptance on it is
5355 /// REQUEST_OK. Read either with [`AnyConnection::recv_response`].
5356 ///
5357 /// Drafts 07 through 11 have no PUBLISH and are refused.
5358 ///
5359 /// The returned [`AnyRequest`] must be held while the offer is live: on
5360 /// drafts 17-23 dropping it resets the stream, which cancels the offer.
5361 #[allow(unused_variables)]
5362 pub async fn publish(
5363 &mut self,
5364 namespace: moqtap_codec::types::TrackNamespace,
5365 track_name: Vec<u8>,
5366 track_alias: moqtap_codec::varint::VarInt,
5367 forward: moqtap_codec::types::Forward,
5368 ) -> Result<AnyRequest, AnyConnectionError> {
5369 let draft = self.draft();
5370 match self {
5371 // Drafts 12 through 14 carry Forward inline. `None` for the Largest
5372 // Location is what writes Content Exists as 0.
5373 #[cfg(feature = "draft12")]
5374 Self::Draft12(c) => c
5375 .publish(
5376 namespace,
5377 track_name,
5378 track_alias,
5379 moqtap_codec::types::GroupOrder::Ascending,
5380 None,
5381 forward,
5382 Vec::new(),
5383 )
5384 .await
5385 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5386 .map_err(AnyConnectionError::from),
5387 #[cfg(feature = "draft13")]
5388 Self::Draft13(c) => c
5389 .publish(
5390 namespace,
5391 track_name,
5392 track_alias,
5393 moqtap_codec::types::GroupOrder::Ascending,
5394 None,
5395 forward,
5396 Vec::new(),
5397 )
5398 .await
5399 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5400 .map_err(AnyConnectionError::from),
5401 #[cfg(feature = "draft14")]
5402 Self::Draft14(c) => c
5403 .publish(
5404 namespace,
5405 track_name,
5406 track_alias,
5407 moqtap_codec::types::GroupOrder::Ascending,
5408 None,
5409 forward,
5410 Vec::new(),
5411 )
5412 .await
5413 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5414 .map_err(AnyConnectionError::from),
5415 #[cfg(feature = "draft15")]
5416 Self::Draft15(c) => c
5417 .publish(namespace, track_name, track_alias, forward_parameter(forward))
5418 .await
5419 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5420 .map_err(AnyConnectionError::from),
5421 // Draft-16's helper takes the track extensions *before* the
5422 // parameters, the reverse of the order drafts 17 on take their
5423 // properties in.
5424 #[cfg(feature = "draft16")]
5425 Self::Draft16(c) => c
5426 .publish(namespace, track_name, track_alias, Vec::new(), forward_parameter(forward))
5427 .await
5428 .map(|request_id| AnyRequest::ControlPlane { request_id, draft })
5429 .map_err(AnyConnectionError::from),
5430 #[cfg(feature = "draft17")]
5431 Self::Draft17(c) => c
5432 .publish(namespace, track_name, track_alias, forward_parameter(forward), Vec::new())
5433 .await
5434 .map(AnyRequest::Draft17)
5435 .map_err(AnyConnectionError::from),
5436 #[cfg(feature = "draft18")]
5437 Self::Draft18(c) => c
5438 .publish(namespace, track_name, track_alias, forward_parameter(forward), Vec::new())
5439 .await
5440 .map(AnyRequest::Draft18)
5441 .map_err(AnyConnectionError::from),
5442 #[cfg(feature = "draft19")]
5443 Self::Draft19(c) => c
5444 .publish(namespace, track_name, track_alias, forward_parameter(forward), Vec::new())
5445 .await
5446 .map(AnyRequest::Draft19)
5447 .map_err(AnyConnectionError::from),
5448 #[cfg(feature = "draft20")]
5449 Self::Draft20(c) => c
5450 .publish(namespace, track_name, track_alias, forward_parameter(forward), Vec::new())
5451 .await
5452 .map(AnyRequest::Draft20)
5453 .map_err(AnyConnectionError::from),
5454 #[cfg(feature = "draft21")]
5455 Self::Draft21(c) => c
5456 .publish(namespace, track_name, track_alias, forward_parameter(forward), Vec::new())
5457 .await
5458 .map(AnyRequest::Draft21)
5459 .map_err(AnyConnectionError::from),
5460 #[cfg(feature = "draft22")]
5461 Self::Draft22(c) => c
5462 .publish(namespace, track_name, track_alias, forward_parameter(forward), Vec::new())
5463 .await
5464 .map(AnyRequest::Draft22)
5465 .map_err(AnyConnectionError::from),
5466 #[cfg(feature = "draft23")]
5467 Self::Draft23(c) => c
5468 .publish(
5469 namespace,
5470 track_name,
5471 track_alias,
5472 crate::draft23::publish::paused_parameter(forward),
5473 Vec::new(),
5474 )
5475 .await
5476 .map(AnyRequest::Draft23)
5477 .map_err(AnyConnectionError::from),
5478 #[allow(unreachable_patterns)]
5479 other => Err(AnyConnectionError::facade(format!(
5480 "publish: PUBLISH first appears in draft-12; draft {:?} has no such message",
5481 other.draft()
5482 ))),
5483 }
5484 }
5485
5486 /// End a publication this session offered with
5487 /// [`AnyConnection::publish`]: PUBLISH_DONE, or SUBSCRIBE_DONE on drafts
5488 /// 12 and 13, which end a PUBLISH-established subscription with the same
5489 /// message as any other. Drafts 12 through 23.
5490 ///
5491 /// Only after the peer accepted. Every draft's endpoint refuses the done
5492 /// for an offer that was rejected or never answered, because there is no
5493 /// subscription to end: a rejected offer owes nothing, and an unanswered
5494 /// one ends with the session on drafts 12 through 16 and with the handle's
5495 /// reset from draft-17.
5496 ///
5497 /// `stream_count` is the number of data streams the publication opened,
5498 /// which is `0` for an offer that sent no objects. **Draft-14 ignores it**:
5499 /// that draft's endpoint writes 0 itself, which is right for a publication
5500 /// that opened no streams and wrong for one that did.
5501 ///
5502 /// # The FIN after it
5503 ///
5504 /// From draft-17 the done is the last message on the offer's own stream,
5505 /// and draft-18 Section 10.11 asks for the FIN straight after it: "A sender
5506 /// SHOULD send FIN on the subscription's bidi stream immediately after
5507 /// sending PUBLISH_DONE." Drafts 19 through 23 send it inside their own
5508 /// helper. The draft-17 and draft-18 helpers leave the send half open, so
5509 /// this finishes it for them; a finished handle does nothing on drop, so
5510 /// the publication is not then cancelled by the drop that follows.
5511 ///
5512 /// `request` is borrowed: on drafts 12 through 16 the handle carries no
5513 /// stream, and from draft-17 the receive half stays open after the FIN.
5514 #[allow(unused_variables)]
5515 pub async fn publish_done(
5516 &mut self,
5517 request: &mut AnyRequest,
5518 status_code: moqtap_codec::varint::VarInt,
5519 stream_count: moqtap_codec::varint::VarInt,
5520 reason: Vec<u8>,
5521 ) -> Result<(), AnyConnectionError> {
5522 match (self, request) {
5523 #[cfg(feature = "draft12")]
5524 (Self::Draft12(c), AnyRequest::ControlPlane { request_id, .. }) => c
5525 .subscribe_done(*request_id, status_code, stream_count, reason)
5526 .await
5527 .map_err(AnyConnectionError::from),
5528 #[cfg(feature = "draft13")]
5529 (Self::Draft13(c), AnyRequest::ControlPlane { request_id, .. }) => c
5530 .subscribe_done(*request_id, status_code, stream_count, reason)
5531 .await
5532 .map_err(AnyConnectionError::from),
5533 #[cfg(feature = "draft14")]
5534 (Self::Draft14(c), AnyRequest::ControlPlane { request_id, .. }) => c
5535 .publish_done(*request_id, status_code, reason)
5536 .await
5537 .map_err(AnyConnectionError::from),
5538 #[cfg(feature = "draft15")]
5539 (Self::Draft15(c), AnyRequest::ControlPlane { request_id, .. }) => c
5540 .publish_done(*request_id, status_code, stream_count, reason)
5541 .await
5542 .map_err(AnyConnectionError::from),
5543 #[cfg(feature = "draft16")]
5544 (Self::Draft16(c), AnyRequest::ControlPlane { request_id, .. }) => c
5545 .publish_done(*request_id, status_code, stream_count, reason)
5546 .await
5547 .map_err(AnyConnectionError::from),
5548 #[cfg(feature = "draft17")]
5549 (Self::Draft17(c), AnyRequest::Draft17(s)) => {
5550 c.publish_done(s, status_code, stream_count, reason)
5551 .await
5552 .map_err(AnyConnectionError::from)?;
5553 s.finish().await.map_err(AnyConnectionError::from)
5554 }
5555 #[cfg(feature = "draft18")]
5556 (Self::Draft18(c), AnyRequest::Draft18(s)) => {
5557 c.publish_done(s, status_code, stream_count, reason)
5558 .await
5559 .map_err(AnyConnectionError::from)?;
5560 s.finish().await.map_err(AnyConnectionError::from)
5561 }
5562 #[cfg(feature = "draft19")]
5563 (Self::Draft19(c), AnyRequest::Draft19(s)) => c
5564 .publish_done(s, status_code, stream_count, reason)
5565 .await
5566 .map_err(AnyConnectionError::from),
5567 #[cfg(feature = "draft20")]
5568 (Self::Draft20(c), AnyRequest::Draft20(s)) => c
5569 .publish_done(s, status_code, stream_count, reason)
5570 .await
5571 .map_err(AnyConnectionError::from),
5572 #[cfg(feature = "draft21")]
5573 (Self::Draft21(c), AnyRequest::Draft21(s)) => c
5574 .publish_done(s, status_code, stream_count, reason)
5575 .await
5576 .map_err(AnyConnectionError::from),
5577 #[cfg(feature = "draft22")]
5578 (Self::Draft22(c), AnyRequest::Draft22(s)) => c
5579 .publish_done(s, status_code, stream_count, reason)
5580 .await
5581 .map_err(AnyConnectionError::from),
5582 #[cfg(feature = "draft23")]
5583 (Self::Draft23(c), AnyRequest::Draft23(s)) => c
5584 .publish_done(s, status_code, stream_count, reason)
5585 .await
5586 .map_err(AnyConnectionError::from),
5587 // A handle from another draft, one `publish` never returns (a
5588 // draft-16 namespace subscription), a draft with no PUBLISH, or a
5589 // build with no drafts enabled.
5590 #[allow(unreachable_patterns)]
5591 (connection, request) => {
5592 let (on, of) = (connection.draft(), request.draft());
5593 Err(AnyConnectionError::facade(if on == of {
5594 format!(
5595 "publish_done: this draft {on:?} handle is not one `publish` returns; \
5596 PUBLISH first appears in draft-12"
5597 )
5598 } else {
5599 format!(
5600 "publish_done: a draft {of:?} request cannot be ended on a draft {on:?} \
5601 connection"
5602 )
5603 }))
5604 }
5605 }
5606 }
5607
5608 /// Ask the peer to offer every track under a namespace prefix:
5609 /// SUBSCRIBE_TRACKS. Drafts 18 through 23.
5610 ///
5611 /// The other half of the draft-18 split described on
5612 /// [`AnyConnection::subscribe_namespace`]: that one asks for namespaces,
5613 /// this one for the tracks inside them. A peer that accepts answers
5614 /// REQUEST_OK on the request's own stream and then sends a PUBLISH, each on
5615 /// a new bidirectional stream, for every matching track — and this side
5616 /// owes each of those an answer. A caller that wants the answer and not
5617 /// the tracks names a prefix nothing matches.
5618 ///
5619 /// `forward` goes out as the FORWARD parameter, and the PUBLISH messages
5620 /// the subscription brings carry it.
5621 ///
5622 /// # Ending it
5623 ///
5624 /// Hold the handle for as long as the subscription should stand, and drop
5625 /// it or [`AnyRequest::cancel`] it to end the subscription: both reset the
5626 /// stream. What a FIN does depends on the draft. Draft-18 counts it as a
5627 /// cancellation: "A SUBSCRIBE_NAMESPACE or SUBSCRIBE_TRACKS can be
5628 /// cancelled by closing the stream with either a FIN or RESET_STREAM."
5629 /// Drafts 19 through 23 do not — draft-23 Section 6.4.2.2: "A FIN only
5630 /// indicates that an endpoint will send no further messages in that
5631 /// direction; it is not a request cancellation."
5632 ///
5633 /// # Drafts 12 through 17
5634 ///
5635 /// Refused here, though the question exists: those drafts ask it with the
5636 /// namespace subscription [`AnyConnection::subscribe_namespace`] sends. On
5637 /// drafts 12 through 15 every SUBSCRIBE_ANNOUNCES or SUBSCRIBE_NAMESPACE
5638 /// brings the matching PUBLISH messages, with no way to decline them; on
5639 /// drafts 16 and 17 Subscribe Options value 0x00 requests them, and that is
5640 /// the value the facade sends. Drafts before 12 have no PUBLISH.
5641 #[allow(unused_variables)]
5642 pub async fn subscribe_tracks(
5643 &mut self,
5644 namespace_prefix: moqtap_codec::types::TrackNamespace,
5645 forward: moqtap_codec::types::Forward,
5646 ) -> Result<AnyRequest, AnyConnectionError> {
5647 match self {
5648 #[cfg(feature = "draft18")]
5649 Self::Draft18(c) => c
5650 .subscribe_tracks(namespace_prefix, forward_parameter(forward))
5651 .await
5652 .map(AnyRequest::Draft18)
5653 .map_err(AnyConnectionError::from),
5654 #[cfg(feature = "draft19")]
5655 Self::Draft19(c) => c
5656 .subscribe_tracks(namespace_prefix, forward_parameter(forward))
5657 .await
5658 .map(AnyRequest::Draft19)
5659 .map_err(AnyConnectionError::from),
5660 #[cfg(feature = "draft20")]
5661 Self::Draft20(c) => c
5662 .subscribe_tracks(namespace_prefix, forward_parameter(forward))
5663 .await
5664 .map(AnyRequest::Draft20)
5665 .map_err(AnyConnectionError::from),
5666 #[cfg(feature = "draft21")]
5667 Self::Draft21(c) => c
5668 .subscribe_tracks(namespace_prefix, forward_parameter(forward))
5669 .await
5670 .map(AnyRequest::Draft21)
5671 .map_err(AnyConnectionError::from),
5672 #[cfg(feature = "draft22")]
5673 Self::Draft22(c) => c
5674 .subscribe_tracks(namespace_prefix, forward_parameter(forward))
5675 .await
5676 .map(AnyRequest::Draft22)
5677 .map_err(AnyConnectionError::from),
5678 #[cfg(feature = "draft23")]
5679 Self::Draft23(c) => c
5680 .subscribe_tracks(
5681 namespace_prefix,
5682 crate::draft23::publish::paused_parameter(forward),
5683 )
5684 .await
5685 .map(AnyRequest::Draft23)
5686 .map_err(AnyConnectionError::from),
5687 #[allow(unreachable_patterns)]
5688 other => Err(AnyConnectionError::facade(format!(
5689 "subscribe_tracks: SUBSCRIBE_TRACKS first appears in draft-18; draft {:?} has \
5690 no such message (drafts 12 through 17 bring tracks with \
5691 subscribe_namespace: always on 12 to 15, through Subscribe Options on 16 \
5692 and 17)",
5693 other.draft()
5694 ))),
5695 }
5696 }
5697
5698 /// Wait for the peer to send something, and say whether it is a request
5699 /// this facade can answer. Supported on **every draft this build carries**.
5700 ///
5701 /// This is the direction [`AnyConnection::recv_response`] does not cover: a
5702 /// relay that accepted a namespace from [`AnyConnection::publish_namespace`]
5703 /// will send a SUBSCRIBE *down* this connection when somebody asks for a
5704 /// track under it, and that message answers nothing this side asked for.
5705 ///
5706 /// # What "wait" means on either side of draft-17
5707 ///
5708 /// Two different waits for one question. Drafts 07 through 16 read the
5709 /// shared control stream, so every message the peer sends — its requests
5710 /// and its answers to this side's — arrives through here and is returned,
5711 /// tagged. From draft-17 this waits on a **new bidirectional stream**, so
5712 /// only the peer's own requests arrive and an answer to something this side
5713 /// asked is read with [`AnyConnection::recv_response`] instead.
5714 ///
5715 /// That is why [`AnyArrival::Other`] exists rather than a filter. On the
5716 /// older drafts a reader that dropped what it was not looking for would
5717 /// swallow a SUBSCRIBE_OK somebody was waiting on; see that variant for
5718 /// what it costs on the newer ones.
5719 ///
5720 /// Every message is dispatched into the endpoint before it is returned, so
5721 /// the session's state is correct whether the caller inspects it or not.
5722 #[allow(unused_variables)]
5723 pub async fn recv_inbound(&mut self) -> Result<AnyArrival, AnyConnectionError> {
5724 // Used only by the per-draft arms below, so a build with no draft
5725 // feature enabled — the `<zero drafts>` row of `just draft-matrix`,
5726 // which exists to prove the crate still compiles with nothing
5727 // selected — compiles this import and reaches no arm that reads it.
5728 #[allow(unused_imports)]
5729 use moqtap_codec::dispatch::AnyControlMessage;
5730
5731 let draft = self.draft();
5732
5733 // The shared-control-stream drafts, whose only difference here is what
5734 // their SUBSCRIBE calls the field holding the peer's Request ID:
5735 // `subscribe_id` up to draft-10 and `request_id` from draft-11. The id
5736 // is read out before the message is wrapped, because wrapping moves it.
5737 #[allow(unused_macros)]
5738 macro_rules! control_plane {
5739 ($conn:ident, $version:ident, $module:ident, $id_field:ident) => {{
5740 let message = $conn.recv_and_dispatch().await.map_err(AnyConnectionError::from)?;
5741 let request_id = match &message {
5742 moqtap_codec::$module::message::ControlMessage::Subscribe(s) => {
5743 Some(s.$id_field)
5744 }
5745 _ => None,
5746 };
5747 let message = AnyControlMessage::$version(message);
5748 Ok(match request_id {
5749 Some(request_id) => AnyArrival::Subscribe {
5750 message,
5751 request: AnyInboundRequest::ControlPlane { request_id, draft },
5752 },
5753 None => AnyArrival::Other(message),
5754 })
5755 }};
5756 }
5757
5758 // The request-stream drafts. A request this facade cannot answer is
5759 // refused with REQUEST_ERROR NOT_SUPPORTED and its stream finished —
5760 // see [`AnyArrival::Other`]. `$extra` fills the fields REQUEST_ERROR
5761 // gained after draft-17.
5762 #[allow(unused_macros)]
5763 macro_rules! request_stream {
5764 ($conn:ident, $version:ident, $module:ident $(, $extra:ident: $value:expr)*) => {{
5765 let (message, mut stream) =
5766 $conn.accept_request_stream().await.map_err(AnyConnectionError::from)?;
5767 let subscribe =
5768 matches!(message, moqtap_codec::$module::message::ControlMessage::Subscribe(_));
5769 let message = AnyControlMessage::$version(message);
5770 if subscribe {
5771 return Ok(AnyArrival::Subscribe {
5772 message,
5773 request: AnyInboundRequest::$version(stream),
5774 });
5775 }
5776 let refusal = moqtap_codec::$module::message::RequestError {
5777 error_code: moqtap_codec::varint::VarInt::from_u64(
5778 moqtap_codec::$module::error_codes::RequestErrorCode::NotSupported as u64,
5779 )
5780 .expect("a request error code fits a varint"),
5781 retry_interval: moqtap_codec::varint::VarInt::from_u64(0).expect("zero fits a varint"),
5782 reason_phrase: Vec::new(),
5783 $($extra: $value,)*
5784 };
5785 // A failed answer falls back to the reset the handle's drop
5786 // sends, so it is not this call's error.
5787 let _ = $conn.respond_error(&mut stream, refusal).await;
5788 Ok(AnyArrival::Other(message))
5789 }};
5790 }
5791
5792 match self {
5793 #[cfg(feature = "draft07")]
5794 Self::Draft07(c) => control_plane!(c, Draft07, draft07, subscribe_id),
5795 #[cfg(feature = "draft08")]
5796 Self::Draft08(c) => control_plane!(c, Draft08, draft08, subscribe_id),
5797 #[cfg(feature = "draft09")]
5798 Self::Draft09(c) => control_plane!(c, Draft09, draft09, subscribe_id),
5799 #[cfg(feature = "draft10")]
5800 Self::Draft10(c) => control_plane!(c, Draft10, draft10, subscribe_id),
5801 #[cfg(feature = "draft11")]
5802 Self::Draft11(c) => control_plane!(c, Draft11, draft11, request_id),
5803 #[cfg(feature = "draft12")]
5804 Self::Draft12(c) => control_plane!(c, Draft12, draft12, request_id),
5805 #[cfg(feature = "draft13")]
5806 Self::Draft13(c) => control_plane!(c, Draft13, draft13, request_id),
5807 #[cfg(feature = "draft14")]
5808 Self::Draft14(c) => control_plane!(c, Draft14, draft14, request_id),
5809 #[cfg(feature = "draft15")]
5810 Self::Draft15(c) => control_plane!(c, Draft15, draft15, request_id),
5811 // Draft-16 gave a stream to namespace subscriptions and to nothing
5812 // else, so an inbound SUBSCRIBE is still a control-stream message
5813 // here and the handle is still a `ControlPlane`.
5814 #[cfg(feature = "draft16")]
5815 Self::Draft16(c) => control_plane!(c, Draft16, draft16, request_id),
5816 #[cfg(feature = "draft17")]
5817 Self::Draft17(c) => request_stream!(c, Draft17, draft17),
5818 #[cfg(feature = "draft18")]
5819 Self::Draft18(c) => request_stream!(c, Draft18, draft18, redirect: None),
5820 #[cfg(feature = "draft19")]
5821 Self::Draft19(c) => request_stream!(c, Draft19, draft19, redirect: None),
5822 #[cfg(feature = "draft20")]
5823 Self::Draft20(c) => request_stream!(c, Draft20, draft20, redirect: None),
5824 #[cfg(feature = "draft21")]
5825 Self::Draft21(c) => request_stream!(c, Draft21, draft21, redirect: None),
5826 #[cfg(feature = "draft22")]
5827 Self::Draft22(c) => request_stream!(c, Draft22, draft22, redirect: None),
5828 #[cfg(feature = "draft23")]
5829 Self::Draft23(c) => request_stream!(c, Draft23, draft23, error_details: Vec::new()),
5830 #[allow(unreachable_patterns)]
5831 other => Err(AnyConnectionError::facade(format!(
5832 "recv_inbound: not yet wired up for draft {:?} via AnyConnection",
5833 other.draft()
5834 ))),
5835 }
5836 }
5837
5838 /// Answer a peer's SUBSCRIBE with SUBSCRIBE_OK. Supported on **every draft
5839 /// this build carries**.
5840 ///
5841 /// `request` is the handle from [`AnyArrival::Subscribe`]. What the unified
5842 /// shape cannot express is defaulted: no expiry, ascending group order, and
5843 /// no parameters, properties or track extensions — every draft's own
5844 /// `Connection::subscribe_ok` is reachable through the variant for a
5845 /// responder that wants any of them.
5846 ///
5847 /// # The Track Alias, and why it *is* an argument here
5848 ///
5849 /// The opposite of [`AnyConnection::subscribe`], and not by inconsistency.
5850 /// Sending a SUBSCRIBE, the alias is bookkeeping the caller has no opinion
5851 /// about — any free one names the track locally — so it is read off the
5852 /// endpoint. Answering one, the alias is what **this side's objects will
5853 /// carry on the wire**, and draft-14 Section 9.8 and draft-14 Section 9.13
5854 /// require one alias to name one track: two subscriptions to the same track
5855 /// must be answered with the *same* alias, and a fresh one per subscription
5856 /// would be wrong. The endpoint cannot know which track a caller considers
5857 /// this to be, so the choice is the caller's and cannot be inferred.
5858 ///
5859 /// **Drafts 07 through 11 ignore it.** Those drafts put the Track Alias on
5860 /// SUBSCRIBE and make it the subscriber's, so their SUBSCRIBE_OK has no
5861 /// such field and the value that counts already arrived — it is on the
5862 /// message in [`AnyArrival::Subscribe`]. The argument is accepted and
5863 /// dropped there rather than the signature splitting in two.
5864 ///
5865 /// # Drafts 07 through 10 have an arm here and no way to be reached
5866 ///
5867 /// Not a gap in this call. A request is legal only below the ceiling its
5868 /// recipient granted, and a ceiling nobody granted is zero, which forbids
5869 /// every request. Drafts 11 and later let this side grant one in
5870 /// CLIENT_SETUP, as Setup Parameter `0x02`; drafts 07 through 10 grant it
5871 /// with the MAX_SUBSCRIBE_ID **message**, type `0x15`, and no `Connection`
5872 /// in this crate has an entry point for sending one. So a peer's SUBSCRIBE
5873 /// on those four drafts is refused for exceeding a ceiling of zero long
5874 /// before it reaches here.
5875 ///
5876 /// The arms stay because they are right and become reachable the day that
5877 /// message can be sent, without being touched.
5878 #[allow(unused_variables)]
5879 pub async fn accept_subscribe(
5880 &mut self,
5881 request: &mut AnyInboundRequest,
5882 track_alias: moqtap_codec::varint::VarInt,
5883 ) -> Result<(), AnyConnectionError> {
5884 use moqtap_codec::varint::VarInt;
5885
5886 // No expiry, which every draft carrying the field spells as zero.
5887 let never_expires = VarInt::from_u64(0).expect("0 fits in VarInt");
5888
5889 match (self, request) {
5890 // Drafts 07 through 11: the subscriber chose the alias, so
5891 // SUBSCRIBE_OK has nowhere to put one.
5892 #[cfg(feature = "draft07")]
5893 (Self::Draft07(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5894 .subscribe_ok(
5895 *request_id,
5896 never_expires,
5897 moqtap_codec::types::GroupOrder::Ascending,
5898 Vec::new(),
5899 )
5900 .await
5901 .map_err(AnyConnectionError::from),
5902 #[cfg(feature = "draft08")]
5903 (Self::Draft08(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5904 .subscribe_ok(
5905 *request_id,
5906 never_expires,
5907 moqtap_codec::types::GroupOrder::Ascending,
5908 Vec::new(),
5909 )
5910 .await
5911 .map_err(AnyConnectionError::from),
5912 #[cfg(feature = "draft09")]
5913 (Self::Draft09(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5914 .subscribe_ok(
5915 *request_id,
5916 never_expires,
5917 moqtap_codec::types::GroupOrder::Ascending,
5918 Vec::new(),
5919 )
5920 .await
5921 .map_err(AnyConnectionError::from),
5922 #[cfg(feature = "draft10")]
5923 (Self::Draft10(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5924 .subscribe_ok(
5925 *request_id,
5926 never_expires,
5927 moqtap_codec::types::GroupOrder::Ascending,
5928 Vec::new(),
5929 )
5930 .await
5931 .map_err(AnyConnectionError::from),
5932 #[cfg(feature = "draft11")]
5933 (Self::Draft11(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5934 .subscribe_ok(
5935 *request_id,
5936 never_expires,
5937 moqtap_codec::types::GroupOrder::Ascending,
5938 Vec::new(),
5939 )
5940 .await
5941 .map_err(AnyConnectionError::from),
5942 // Draft-12 moved the alias here, and it stays until draft-17 moves
5943 // the whole response onto the request's own stream.
5944 #[cfg(feature = "draft12")]
5945 (Self::Draft12(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5946 .subscribe_ok(
5947 *request_id,
5948 track_alias,
5949 never_expires,
5950 moqtap_codec::types::GroupOrder::Ascending,
5951 Vec::new(),
5952 )
5953 .await
5954 .map_err(AnyConnectionError::from),
5955 #[cfg(feature = "draft13")]
5956 (Self::Draft13(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5957 .subscribe_ok(
5958 *request_id,
5959 track_alias,
5960 never_expires,
5961 moqtap_codec::types::GroupOrder::Ascending,
5962 Vec::new(),
5963 )
5964 .await
5965 .map_err(AnyConnectionError::from),
5966 #[cfg(feature = "draft14")]
5967 (Self::Draft14(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5968 .subscribe_ok(
5969 *request_id,
5970 track_alias,
5971 never_expires,
5972 moqtap_codec::types::GroupOrder::Ascending,
5973 Vec::new(),
5974 )
5975 .await
5976 .map_err(AnyConnectionError::from),
5977 // Draft-15 turned expiry and group order into parameters, so the
5978 // two arguments above become an empty list rather than defaults.
5979 #[cfg(feature = "draft15")]
5980 (Self::Draft15(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5981 .subscribe_ok(*request_id, track_alias, Vec::new())
5982 .await
5983 .map_err(AnyConnectionError::from),
5984 // Draft-16 added a Track Extensions list beside the parameters.
5985 #[cfg(feature = "draft16")]
5986 (Self::Draft16(c), AnyInboundRequest::ControlPlane { request_id, .. }) => c
5987 .subscribe_ok(*request_id, track_alias, Vec::new(), Vec::new())
5988 .await
5989 .map_err(AnyConnectionError::from),
5990 // Drafts 17 through 20: the response goes back on the request's own
5991 // stream, which is why it carries no Request ID — the stream is the
5992 // correlation.
5993 #[cfg(feature = "draft17")]
5994 (Self::Draft17(c), AnyInboundRequest::Draft17(s)) => {
5995 use moqtap_codec::draft17::message::SubscribeOk;
5996 c.respond_subscribe_ok(
5997 s,
5998 SubscribeOk {
5999 track_alias,
6000 parameters: Vec::new(),
6001 track_properties: Vec::new(),
6002 },
6003 )
6004 .await
6005 .map_err(AnyConnectionError::from)
6006 }
6007 #[cfg(feature = "draft18")]
6008 (Self::Draft18(c), AnyInboundRequest::Draft18(s)) => {
6009 use moqtap_codec::draft18::message::SubscribeOk;
6010 c.respond_subscribe_ok(
6011 s,
6012 SubscribeOk {
6013 track_alias,
6014 parameters: Vec::new(),
6015 track_properties: Vec::new(),
6016 },
6017 )
6018 .await
6019 .map_err(AnyConnectionError::from)
6020 }
6021 #[cfg(feature = "draft19")]
6022 (Self::Draft19(c), AnyInboundRequest::Draft19(s)) => {
6023 use moqtap_codec::draft19::message::SubscribeOk;
6024 c.respond_subscribe_ok(
6025 s,
6026 SubscribeOk {
6027 track_alias,
6028 parameters: Vec::new(),
6029 track_properties: Vec::new(),
6030 },
6031 )
6032 .await
6033 .map_err(AnyConnectionError::from)
6034 }
6035 #[cfg(feature = "draft20")]
6036 (Self::Draft20(c), AnyInboundRequest::Draft20(s)) => {
6037 use moqtap_codec::draft20::message::SubscribeOk;
6038 c.respond_subscribe_ok(
6039 s,
6040 SubscribeOk {
6041 track_alias,
6042 parameters: Vec::new(),
6043 track_properties: Vec::new(),
6044 },
6045 )
6046 .await
6047 .map_err(AnyConnectionError::from)
6048 }
6049 #[cfg(feature = "draft21")]
6050 (Self::Draft21(c), AnyInboundRequest::Draft21(s)) => {
6051 use moqtap_codec::draft21::message::SubscribeOk;
6052 c.respond_subscribe_ok(
6053 s,
6054 SubscribeOk {
6055 track_alias,
6056 parameters: Vec::new(),
6057 track_properties: Vec::new(),
6058 },
6059 )
6060 .await
6061 .map_err(AnyConnectionError::from)
6062 }
6063 #[cfg(feature = "draft22")]
6064 (Self::Draft22(c), AnyInboundRequest::Draft22(s)) => {
6065 use moqtap_codec::draft22::message::SubscribeOk;
6066 c.respond_subscribe_ok(
6067 s,
6068 SubscribeOk {
6069 track_alias,
6070 parameters: Vec::new(),
6071 track_properties: Vec::new(),
6072 },
6073 )
6074 .await
6075 .map_err(AnyConnectionError::from)
6076 }
6077 #[cfg(feature = "draft23")]
6078 (Self::Draft23(c), AnyInboundRequest::Draft23(s)) => {
6079 use moqtap_codec::draft23::message::SubscribeOk;
6080 c.respond_subscribe_ok(
6081 s,
6082 SubscribeOk {
6083 track_alias,
6084 parameters: Vec::new(),
6085 track_properties: Vec::new(),
6086 },
6087 )
6088 .await
6089 .map_err(AnyConnectionError::from)
6090 }
6091 // A handle from another draft, refused rather than answered on
6092 // whatever stream happens to be at hand — the same rule
6093 // [`AnyConnection::recv_response`] follows, for the same reason.
6094 #[allow(unreachable_patterns)]
6095 (connection, request) => Err(AnyConnectionError::facade(format!(
6096 "accept_subscribe: a draft {:?} request cannot be answered on a draft {:?} \
6097 connection",
6098 request.draft(),
6099 connection.draft(),
6100 ))),
6101 }
6102 }
6103
6104 /// Open a subgroup stream and frame it, ready for objects.
6105 ///
6106 /// The first thing a relay can be asked that is not a question about
6107 /// control messages. Answering SUBSCRIBE is not delivering a track, and
6108 /// until something puts an object on the wire there is no way to tell the
6109 /// two apart from outside.
6110 ///
6111 /// # The header this writes, and the four shapes it takes
6112 ///
6113 /// An explicit Subgroup ID, no extension block where a draft can say so,
6114 /// and a publisher priority on the wire — the plainest conforming subgroup
6115 /// stream each draft can carry. The struct that says so is different four
6116 /// times over:
6117 ///
6118 /// * **Drafts 07 through 10** put the stream type outside the header
6119 /// entirely, so the header is four fields and nothing selects a layout.
6120 /// * **Drafts 11 through 13** name a `StreamType` variant per layout, and
6121 /// `SubgroupExplicit` is the one that carries an ID and no extensions.
6122 /// * **Draft-14** folds the type into the header as a flag word, and the
6123 /// Subgroup ID becomes an `Option` that must agree with it — a type
6124 /// saying a field follows, with no field, is refused.
6125 /// * **Drafts 15 through 20** replace the flags with a type byte:
6126 /// `0x14` is the subgroup base `0x10`, plus `0x04` for an explicit
6127 /// Subgroup ID, with the extension bit `0x01` clear, the end-of-group
6128 /// bit `0x08` clear, and the no-priority bit `0x20` clear. Drafts 15 and
6129 /// 16 read those two ID bits as a pair of table columns and drafts 17
6130 /// through 20 call them a SUBGROUP_ID_MODE field, which is a difference
6131 /// in wording and not in bytes.
6132 ///
6133 /// # Why the extension block is not an argument
6134 ///
6135 /// It is not a property of any object, so there is nothing an object-level
6136 /// caller could be asked. The header settles it for the whole stream, and
6137 /// an object that disagreed with its header would misframe every object
6138 /// after it — the missing length is read out of the next field along. A
6139 /// caller that needs to put extensions on a stream needs to say so when the
6140 /// stream opens, which is a second entry point rather than an argument
6141 /// here.
6142 ///
6143 /// Three drafts do not offer the choice at all, and in opposite
6144 /// directions. Draft-07 has no extension block anywhere, so its streams
6145 /// carry none because there is none to carry. Drafts 08 through 10 have one
6146 /// on every object and **no header field that could say otherwise**: their
6147 /// subgroup header is four values with no type byte, so every object writes
6148 /// a length whether it has anything to put after it or not. On those three
6149 /// [`moqtap_codec::dispatch::AnySubgroupHeader::carries_extension_block`]
6150 /// answers without consulting the header, and a caller asking for a stream
6151 /// without one is asking for a stream the draft does not define.
6152 ///
6153 /// Nothing checks that `track_alias` names a track this session agreed to
6154 /// publish, because a probe measuring a relay may want to send objects for
6155 /// one it did not.
6156 #[allow(unused_variables)]
6157 pub async fn open_subgroup(
6158 &self,
6159 track_alias: u64,
6160 group_id: u64,
6161 subgroup_id: u64,
6162 publisher_priority: u8,
6163 ) -> Result<AnySubgroupWriter, AnyConnectionError> {
6164 // Used only by the per-draft arms below, so a build with no draft
6165 // feature enabled — the `<zero drafts>` row of `just draft-matrix`,
6166 // which exists to prove the crate still compiles with nothing
6167 // selected — compiles this import and reaches no arm that reads it.
6168 #[allow(unused_imports)]
6169 use moqtap_codec::dispatch::AnySubgroupHeader;
6170 use moqtap_codec::varint::VarInt;
6171 let alias = VarInt::from_u64(track_alias)
6172 .map_err(|e| AnyConnectionError::facade(format!("track alias {track_alias}: {e}")))?;
6173 let group = VarInt::from_u64(group_id)
6174 .map_err(|e| AnyConnectionError::facade(format!("group id {group_id}: {e}")))?;
6175 let subgroup = VarInt::from_u64(subgroup_id)
6176 .map_err(|e| AnyConnectionError::facade(format!("subgroup id {subgroup_id}: {e}")))?;
6177
6178 #[allow(unused_macros)]
6179 macro_rules! open {
6180 ($c:ident, $variant:ident, $header:expr) => {{
6181 let header = AnySubgroupHeader::$variant($header);
6182 $c.open_subgroup_stream(&header)
6183 .await
6184 .map(AnySubgroupWriter::$variant)
6185 .map_err(AnyConnectionError::from)
6186 }};
6187 }
6188
6189 match self {
6190 #[cfg(feature = "draft07")]
6191 Self::Draft07(c) => open!(
6192 c,
6193 Draft07,
6194 moqtap_codec::draft07::data_stream::SubgroupHeader {
6195 track_alias: alias,
6196 group_id: group,
6197 subgroup_id: subgroup,
6198 publisher_priority,
6199 }
6200 ),
6201 #[cfg(feature = "draft08")]
6202 Self::Draft08(c) => open!(
6203 c,
6204 Draft08,
6205 moqtap_codec::draft08::data_stream::SubgroupHeader {
6206 track_alias: alias,
6207 group_id: group,
6208 subgroup_id: subgroup,
6209 publisher_priority,
6210 }
6211 ),
6212 #[cfg(feature = "draft09")]
6213 Self::Draft09(c) => open!(
6214 c,
6215 Draft09,
6216 moqtap_codec::draft09::data_stream::SubgroupHeader {
6217 track_alias: alias,
6218 group_id: group,
6219 subgroup_id: subgroup,
6220 publisher_priority,
6221 }
6222 ),
6223 #[cfg(feature = "draft10")]
6224 Self::Draft10(c) => open!(
6225 c,
6226 Draft10,
6227 moqtap_codec::draft10::data_stream::SubgroupHeader {
6228 track_alias: alias,
6229 group_id: group,
6230 subgroup_id: subgroup,
6231 publisher_priority,
6232 }
6233 ),
6234 #[cfg(feature = "draft11")]
6235 Self::Draft11(c) => open!(
6236 c,
6237 Draft11,
6238 moqtap_codec::draft11::data_stream::SubgroupHeader {
6239 stream_type: moqtap_codec::draft11::data_stream::StreamType::SubgroupExplicit,
6240 track_alias: alias,
6241 group_id: group,
6242 subgroup_id: subgroup,
6243 publisher_priority,
6244 }
6245 ),
6246 #[cfg(feature = "draft12")]
6247 Self::Draft12(c) => open!(
6248 c,
6249 Draft12,
6250 moqtap_codec::draft12::data_stream::SubgroupHeader {
6251 stream_type: moqtap_codec::draft12::data_stream::StreamType::SubgroupExplicit,
6252 track_alias: alias,
6253 group_id: group,
6254 subgroup_id: subgroup,
6255 publisher_priority,
6256 }
6257 ),
6258 #[cfg(feature = "draft13")]
6259 Self::Draft13(c) => open!(
6260 c,
6261 Draft13,
6262 moqtap_codec::draft13::data_stream::SubgroupHeader {
6263 stream_type: moqtap_codec::draft13::data_stream::StreamType::SubgroupExplicit,
6264 track_alias: alias,
6265 group_id: group,
6266 subgroup_id: subgroup,
6267 publisher_priority,
6268 }
6269 ),
6270 #[cfg(feature = "draft14")]
6271 Self::Draft14(c) => open!(
6272 c,
6273 Draft14,
6274 moqtap_codec::draft14::data_stream::SubgroupHeader {
6275 // Subgroup ID field present, not taken from the first
6276 // object, no extensions, not the end of the group.
6277 stream_type: moqtap_codec::draft14::data_stream::SubgroupStreamType::from_flags(
6278 true, false, false, false,
6279 ),
6280 track_alias: alias,
6281 group_id: group,
6282 subgroup_id: Some(subgroup),
6283 publisher_priority,
6284 }
6285 ),
6286 #[cfg(feature = "draft15")]
6287 Self::Draft15(c) => open!(
6288 c,
6289 Draft15,
6290 moqtap_codec::draft15::data_stream::SubgroupHeader {
6291 header_type: 0x14,
6292 track_alias: alias,
6293 group_id: group,
6294 subgroup_id: subgroup,
6295 publisher_priority: Some(publisher_priority),
6296 }
6297 ),
6298 #[cfg(feature = "draft16")]
6299 Self::Draft16(c) => open!(
6300 c,
6301 Draft16,
6302 moqtap_codec::draft16::data_stream::SubgroupHeader {
6303 header_type: 0x14,
6304 track_alias: alias,
6305 group_id: group,
6306 subgroup_id: subgroup,
6307 publisher_priority: Some(publisher_priority),
6308 }
6309 ),
6310 #[cfg(feature = "draft17")]
6311 Self::Draft17(c) => open!(
6312 c,
6313 Draft17,
6314 moqtap_codec::draft17::data_stream::SubgroupHeader {
6315 header_type: 0x14,
6316 track_alias: alias,
6317 group_id: group,
6318 subgroup_id: subgroup,
6319 publisher_priority: Some(publisher_priority),
6320 }
6321 ),
6322 #[cfg(feature = "draft18")]
6323 Self::Draft18(c) => open!(
6324 c,
6325 Draft18,
6326 moqtap_codec::draft18::data_stream::SubgroupHeader {
6327 header_type: 0x14,
6328 track_alias: alias,
6329 group_id: group,
6330 subgroup_id: subgroup,
6331 publisher_priority: Some(publisher_priority),
6332 }
6333 ),
6334 #[cfg(feature = "draft19")]
6335 Self::Draft19(c) => open!(
6336 c,
6337 Draft19,
6338 moqtap_codec::draft19::data_stream::SubgroupHeader {
6339 header_type: 0x14,
6340 track_alias: alias,
6341 group_id: group,
6342 subgroup_id: subgroup,
6343 publisher_priority: Some(publisher_priority),
6344 }
6345 ),
6346 #[cfg(feature = "draft20")]
6347 Self::Draft20(c) => open!(
6348 c,
6349 Draft20,
6350 moqtap_codec::draft20::data_stream::SubgroupHeader {
6351 header_type: 0x14,
6352 track_alias: alias,
6353 group_id: group,
6354 subgroup_id: subgroup,
6355 publisher_priority: Some(publisher_priority),
6356 }
6357 ),
6358 #[cfg(feature = "draft21")]
6359 Self::Draft21(c) => open!(
6360 c,
6361 Draft21,
6362 moqtap_codec::draft21::data_stream::SubgroupHeader {
6363 header_type: 0x14,
6364 track_alias: alias,
6365 group_id: group,
6366 subgroup_id: subgroup,
6367 publisher_priority: Some(publisher_priority),
6368 }
6369 ),
6370 #[cfg(feature = "draft22")]
6371 Self::Draft22(c) => open!(
6372 c,
6373 Draft22,
6374 moqtap_codec::draft22::data_stream::SubgroupHeader {
6375 header_type: 0x14,
6376 track_alias: alias,
6377 group_id: group,
6378 subgroup_id: subgroup,
6379 publisher_priority: Some(publisher_priority),
6380 }
6381 ),
6382 #[cfg(feature = "draft23")]
6383 Self::Draft23(c) => open!(
6384 c,
6385 Draft23,
6386 moqtap_codec::draft23::data_stream::SubgroupHeader {
6387 header_type: 0x14,
6388 track_alias: alias,
6389 group_id: group,
6390 subgroup_id: subgroup,
6391 publisher_priority: Some(publisher_priority),
6392 }
6393 ),
6394 #[allow(unreachable_patterns)]
6395 _ => Err(AnyConnectionError::facade("no draft feature is enabled")),
6396 }
6397 }
6398
6399 /// Accept the next subgroup stream the peer opens, and read its header.
6400 ///
6401 /// The header comes back beside the reader because it is the only place
6402 /// several things are said: which track the objects belong to (by Track
6403 /// Alias), which group, and — through
6404 /// [`carries_extension_block`](moqtap_codec::dispatch::AnySubgroupHeader::carries_extension_block)
6405 /// — whether the objects on this stream write an extension block at all. A
6406 /// reader that did not know the last of those could not frame a single
6407 /// object.
6408 ///
6409 /// Reading the header is not separable from accepting the stream, and the
6410 /// per-draft connections do both in one call for a reason beyond
6411 /// convenience: the header settles the track's forwarding preference and
6412 /// binds the stream to the endpoint's object bookkeeping for that alias, so
6413 /// a stream accepted without its header read would be a stream whose
6414 /// objects nothing is measuring.
6415 ///
6416 /// This waits on the *next* unidirectional stream, whatever it carries. A
6417 /// draft that sends something else on one — and every draft does, for
6418 /// fetches — will fail to parse a subgroup header out of it, which is the
6419 /// same wall [`AnyConnection::accept_fetch`] hits from the other side.
6420 /// Nothing here reads the stream type first and branches: a caller that
6421 /// could receive either has to know which it is expecting, because the
6422 /// header decides how every object after it is framed.
6423 pub async fn accept_subgroup(
6424 &self,
6425 ) -> Result<(moqtap_codec::dispatch::AnySubgroupHeader, AnySubgroupReader), AnyConnectionError>
6426 {
6427 #[allow(unused_macros)]
6428 macro_rules! accept {
6429 ($c:ident, $variant:ident) => {{
6430 $c.accept_subgroup_stream()
6431 .await
6432 .map(|(header, stream)| (header, AnySubgroupReader::$variant(stream)))
6433 .map_err(AnyConnectionError::from)
6434 }};
6435 }
6436
6437 match self {
6438 #[cfg(feature = "draft07")]
6439 Self::Draft07(c) => accept!(c, Draft07),
6440 #[cfg(feature = "draft08")]
6441 Self::Draft08(c) => accept!(c, Draft08),
6442 #[cfg(feature = "draft09")]
6443 Self::Draft09(c) => accept!(c, Draft09),
6444 #[cfg(feature = "draft10")]
6445 Self::Draft10(c) => accept!(c, Draft10),
6446 #[cfg(feature = "draft11")]
6447 Self::Draft11(c) => accept!(c, Draft11),
6448 #[cfg(feature = "draft12")]
6449 Self::Draft12(c) => accept!(c, Draft12),
6450 #[cfg(feature = "draft13")]
6451 Self::Draft13(c) => accept!(c, Draft13),
6452 #[cfg(feature = "draft14")]
6453 Self::Draft14(c) => accept!(c, Draft14),
6454 #[cfg(feature = "draft15")]
6455 Self::Draft15(c) => accept!(c, Draft15),
6456 #[cfg(feature = "draft16")]
6457 Self::Draft16(c) => accept!(c, Draft16),
6458 #[cfg(feature = "draft17")]
6459 Self::Draft17(c) => accept!(c, Draft17),
6460 #[cfg(feature = "draft18")]
6461 Self::Draft18(c) => accept!(c, Draft18),
6462 #[cfg(feature = "draft19")]
6463 Self::Draft19(c) => accept!(c, Draft19),
6464 #[cfg(feature = "draft20")]
6465 Self::Draft20(c) => accept!(c, Draft20),
6466 #[cfg(feature = "draft21")]
6467 Self::Draft21(c) => accept!(c, Draft21),
6468 #[cfg(feature = "draft22")]
6469 Self::Draft22(c) => accept!(c, Draft22),
6470 #[cfg(feature = "draft23")]
6471 Self::Draft23(c) => accept!(c, Draft23),
6472 #[allow(unreachable_patterns)]
6473 _ => Err(AnyConnectionError::facade("no draft feature is enabled")),
6474 }
6475 }
6476
6477 /// Accept the next fetch stream the peer opens, and read its FETCH_HEADER.
6478 ///
6479 /// [`AnyConnection::accept_subgroup`]'s twin, and it waits on the same
6480 /// queue: the *next* unidirectional stream, whatever it carries. A subgroup
6481 /// stream arriving here fails to parse as a FETCH_HEADER, and a fetch stream
6482 /// arriving there fails the other way. A caller reading both kinds on one
6483 /// session has to know which is next, and this facade will not guess for it
6484 /// — see `accept_subgroup` for why guessing is the wrong shape.
6485 ///
6486 /// The header comes back because it names the request: everything on the
6487 /// stream after it answers the FETCH whose Request ID it carries, and a
6488 /// caller with more than one fetch in flight has no other way to tell the
6489 /// streams apart.
6490 ///
6491 /// # Why this takes a Group Order and `accept_subgroup` takes nothing
6492 ///
6493 /// Because on four drafts the reader cannot be started without it, and
6494 /// starting it wrong is silent. Drafts 18 through 23 encode an object's
6495 /// Group ID as a **delta**, and draft-18 Section 11.4.4.1 makes that delta
6496 /// count upward under Ascending and downward under Descending. A reader
6497 /// started in the wrong direction still parses every frame and reports Group
6498 /// IDs that walk the wrong way — the same trap
6499 /// `Connection::accept_fill_stream` documents on draft-20, where the
6500 /// endpoint happens to know the answer and this facade does not.
6501 ///
6502 /// Here nothing on this side holds it: the order is on the FETCH_OK, which
6503 /// is a control message a data stream never sees, and which the caller has
6504 /// already read. [`fetch_group_order`] takes it off that message so the
6505 /// lookup is written once rather than per caller.
6506 ///
6507 /// On the other drafts the argument is inert, and it is an argument
6508 /// rather than an `Option` because a caller that has a FETCH_OK in hand can
6509 /// always answer it, and one that cannot has not read the answer yet.
6510 #[allow(unused_variables)]
6511 pub async fn accept_fetch(
6512 &self,
6513 group_order: moqtap_codec::types::GroupOrder,
6514 ) -> Result<(moqtap_codec::dispatch::AnyFetchHeader, AnyFetchReader), AnyConnectionError> {
6515 #[allow(unused_macros)]
6516 macro_rules! accept {
6517 ($c:ident, $variant:ident) => {{
6518 $c.accept_fetch_stream()
6519 .await
6520 .map(|(header, stream)| (header, AnyFetchReader::$variant(stream)))
6521 .map_err(AnyConnectionError::from)
6522 }};
6523 // The three drafts whose object reader has to be pointed in a
6524 // direction before it will read anything. Each of them declares its
6525 // own two-valued `GroupOrder` beside the reader that consumes it,
6526 // because a delta either adds or subtracts and there is no third
6527 // thing it could do. `Publisher` — the third value of the control
6528 // plane's own enum, and the one a FETCH_OK omitting the property
6529 // leaves — becomes Ascending here, which is what draft-20 Section
6530 // 10.2.8 makes the default.
6531 ($c:ident, $variant:ident, $module:ident) => {{
6532 let order = match group_order {
6533 moqtap_codec::types::GroupOrder::Descending => {
6534 moqtap_codec::$module::data_stream::GroupOrder::Descending
6535 }
6536 _ => moqtap_codec::$module::data_stream::GroupOrder::Ascending,
6537 };
6538 $c.accept_fetch_stream()
6539 .await
6540 .map(|(header, mut stream)| {
6541 stream.begin_fetch_objects(order);
6542 (header, AnyFetchReader::$variant(stream))
6543 })
6544 .map_err(AnyConnectionError::from)
6545 }};
6546 }
6547
6548 match self {
6549 #[cfg(feature = "draft07")]
6550 Self::Draft07(c) => accept!(c, Draft07),
6551 #[cfg(feature = "draft08")]
6552 Self::Draft08(c) => accept!(c, Draft08),
6553 #[cfg(feature = "draft09")]
6554 Self::Draft09(c) => accept!(c, Draft09),
6555 #[cfg(feature = "draft10")]
6556 Self::Draft10(c) => accept!(c, Draft10),
6557 #[cfg(feature = "draft11")]
6558 Self::Draft11(c) => accept!(c, Draft11),
6559 #[cfg(feature = "draft12")]
6560 Self::Draft12(c) => accept!(c, Draft12),
6561 #[cfg(feature = "draft13")]
6562 Self::Draft13(c) => accept!(c, Draft13),
6563 #[cfg(feature = "draft14")]
6564 Self::Draft14(c) => accept!(c, Draft14),
6565 #[cfg(feature = "draft15")]
6566 Self::Draft15(c) => accept!(c, Draft15),
6567 // The resolver is created here rather than seeded from the header,
6568 // because draft-16's first object supplies its own Location and
6569 // every later one inherits from the object before it.
6570 #[cfg(feature = "draft16")]
6571 Self::Draft16(c) => c
6572 .accept_fetch_stream()
6573 .await
6574 .map(|(header, stream)| {
6575 (
6576 header,
6577 AnyFetchReader::Draft16(Draft16FetchStream {
6578 stream,
6579 reader: moqtap_codec::draft16::data_stream::FetchObjectReader::new(),
6580 }),
6581 )
6582 })
6583 .map_err(AnyConnectionError::from),
6584 #[cfg(feature = "draft17")]
6585 Self::Draft17(c) => accept!(c, Draft17),
6586 #[cfg(feature = "draft18")]
6587 Self::Draft18(c) => accept!(c, Draft18, draft18),
6588 #[cfg(feature = "draft19")]
6589 Self::Draft19(c) => accept!(c, Draft19, draft19),
6590 #[cfg(feature = "draft20")]
6591 Self::Draft20(c) => accept!(c, Draft20, draft20),
6592 #[cfg(feature = "draft21")]
6593 Self::Draft21(c) => accept!(c, Draft21, draft21),
6594 #[cfg(feature = "draft22")]
6595 Self::Draft22(c) => accept!(c, Draft22, draft22),
6596 #[cfg(feature = "draft23")]
6597 Self::Draft23(c) => accept!(c, Draft23, draft23),
6598 #[allow(unreachable_patterns)]
6599 _ => Err(AnyConnectionError::facade("no draft feature is enabled")),
6600 }
6601 }
6602
6603 /// Send a SUBSCRIBE_UPDATE for an active subscription. Draft-14 only.
6604 ///
6605 /// # Why no later draft is wired, and why that is not this call's to fix
6606 ///
6607 /// Draft-15 renamed the message REQUEST_UPDATE and rebuilt it, and from
6608 /// draft-17 it travels **on the request's own bidirectional stream** rather
6609 /// than on a control stream — so an update needs the [`AnyRequest`] the
6610 /// original request returned, which this signature does not take and cannot
6611 /// be given without becoming a different call. Draft-20 goes further and
6612 /// has no `start_location` / `end_group` pair at all: Section 10.9 carries
6613 /// the new range as a `LOCATION_FILTER` parameter, on the same inclusive
6614 /// terms [`FetchRange`] describes. Reach a draft's own
6615 /// `Connection::send_on_request_stream` through the variant.
6616 #[allow(unused_variables)]
6617 pub async fn subscribe_update(
6618 &mut self,
6619 subscription_request_id: moqtap_codec::varint::VarInt,
6620 start_location: moqtap_codec::types::Location,
6621 end_group: moqtap_codec::varint::VarInt,
6622 subscriber_priority: u8,
6623 forward: moqtap_codec::types::Forward,
6624 ) -> Result<(), AnyConnectionError> {
6625 match self {
6626 #[cfg(feature = "draft14")]
6627 Self::Draft14(c) => c
6628 .subscribe_update(
6629 subscription_request_id,
6630 start_location,
6631 end_group,
6632 subscriber_priority,
6633 forward,
6634 Vec::new(),
6635 )
6636 .await
6637 .map(|_| ())
6638 .map_err(AnyConnectionError::from),
6639 #[allow(unreachable_patterns)]
6640 other => Err(AnyConnectionError::facade(format!(
6641 "subscribe_update: not yet wired up for draft {:?} via AnyConnection",
6642 other.draft()
6643 ))),
6644 }
6645 }
6646}
6647
6648/// Trait for receiving events from an [`AnyConnection`].
6649///
6650/// Implementations must be `Send + Sync` because the adapter installed on
6651/// the inner draft-specific connection may emit events from async tasks.
6652/// `on_event` takes `&self` — implementations that need mutation should use
6653/// interior mutability (e.g. `Mutex`, `mpsc::Sender`).
6654///
6655/// The per-draft adapter clones the draft-specific event into the matching
6656/// [`AnyClientEvent`] variant before invoking `on_event`.
6657pub trait AnyConnectionObserver: Send + Sync {
6658 /// Called when a connection event occurs on any draft.
6659 fn on_event(&self, event: &AnyClientEvent);
6660}
6661
6662/// A no-op observer that discards all events.
6663pub struct NoOpObserver;
6664
6665impl AnyConnectionObserver for NoOpObserver {
6666 fn on_event(&self, _event: &AnyClientEvent) {}
6667}
6668
6669/// One control message as it crossed the wire, lifted out of whichever draft's
6670/// event carried it.
6671///
6672/// [`AnyClientEvent`] wraps a draft's own `ClientEvent`, so an observer holding
6673/// one can read [`draft`](AnyClientEvent::draft) and nothing else without
6674/// matching every enabled variant — which means a consumer outside this crate
6675/// writing its own cascade over every draft, the exact duplication this module
6676/// exists to hold in one place. [`AnyClientEvent::control_frame`] answers with
6677/// this instead.
6678///
6679/// Everything here borrows from the event, so it is a view rather than a
6680/// record: a consumer that wants to keep a frame copies the bytes out.
6681#[non_exhaustive]
6682#[derive(Debug, Clone, Copy)]
6683pub struct ControlFrame<'a> {
6684 /// The draft whose rules were used to decode it.
6685 pub draft: DraftVersion,
6686 /// Whether this endpoint sent it, as opposed to receiving it.
6687 ///
6688 /// A `bool` rather than a shared direction enum because each draft declares
6689 /// its own `Direction` and there is no cross-draft one to lift them into;
6690 /// inventing a shared one to convert every draft's into would be a type
6691 /// whose only purpose is to be converted.
6692 pub outbound: bool,
6693 /// The decoded message. [`message_type_id`] and [`message_type_name`] name
6694 /// the codepoint it arrived under.
6695 ///
6696 /// [`message_type_id`]: moqtap_codec::dispatch::AnyControlMessage::message_type_id
6697 /// [`message_type_name`]: moqtap_codec::dispatch::AnyControlMessage::message_type_name
6698 pub message: &'a moqtap_codec::dispatch::AnyControlMessage,
6699 /// The framed wire bytes — type, length and payload — as they arrived.
6700 ///
6701 /// `None` when the connection was reading without an observer attached and
6702 /// therefore never cloned them. That cannot happen for an event an observer
6703 /// is being handed, so in practice this is `None` only for a frame built by
6704 /// hand.
6705 ///
6706 /// Kept because the encoding is evidence the decoding discards: two peers
6707 /// sending the same field can still disagree on how wide a varint they
6708 /// wrote it in, and the decoded form answers the same either way.
6709 pub raw: Option<&'a [u8]>,
6710}