moqtap_client/transport/mod.rs
1//! Transport abstraction for QUIC and WebTransport.
2//!
3//! Uses enum dispatch (not trait objects) since the transport set is closed.
4//! WebTransport support is behind the `webtransport` feature flag. Nothing
5//! here is per-draft; every draft's connection is carried over the same two.
6
7pub mod quic;
8#[cfg(feature = "webtransport")]
9pub mod webtransport;
10
11pub use quic::{
12 dial_quic, dial_quic_to, show_cipher_suite, CertificateHook, CertificateLog, DialError,
13 DialPhase, QuicDialOptions, QuicTarget, TLS13_CIPHER_SUITES,
14};
15#[cfg(feature = "webtransport")]
16pub use webtransport::{dial_webtransport, dial_webtransport_to};
17
18use std::future::Future;
19
20use bytes::Bytes;
21
22/// A handshake that failed, with the codes the failure carried still intact.
23///
24/// The reason it is a struct and not another `String` variant: a peer's refusal
25/// arrives as a *number*, and flattening it into prose is lossy in a way that
26/// only shows up downstream. A TLS alert reaches QUIC as `0x0100 | alert`, so
27/// `no_application_protocol` (120) is `0x178`, and a caller that wants to
28/// distinguish "this relay does not speak our protocol" from "this relay's
29/// certificate expired" (alert 45, `0x12D`) was left parsing error messages for
30/// digits.
31///
32/// Every field is optional except [`reason`](Self::reason), because not every
33/// failure has a code — a timeout and a DNS miss are real outcomes with nothing
34/// numeric in them.
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct HandshakeFailure {
37 /// The QUIC error code, when the failure carried one.
38 ///
39 /// Meaningless without [`code_space`](Self::code_space) — the same integer
40 /// says different things in the two spaces.
41 pub code: Option<u64>,
42 /// Which space [`code`](Self::code) is a number in.
43 pub code_space: Option<CodeSpace>,
44 /// The TLS alert, when the code was a *transport* code in QUIC's crypto
45 /// range.
46 ///
47 /// Derived and not independently sourced: QUIC has no separate field for
48 /// it, and `0x0100..=0x01ff` *is* how TLS alerts are carried.
49 pub tls_alert: Option<u8>,
50 /// What the stack said, for the detail no code carries — which certificate
51 /// field mismatched, when it expired, which name was expected.
52 ///
53 /// Alert 42 (`bad_certificate`) is rustls's catch-all, so telling an expired
54 /// certificate from a name mismatch still means reading this.
55 pub reason: String,
56}
57
58/// Which number space a QUIC error code belongs to.
59///
60/// Kept because the same integer means different things in each, and a report
61/// that loses the space is one that cannot be read. Transport code 271 is
62/// `0x10F` — inside the range TLS alerts are carried in. Application code 271
63/// is a peer's own close code and has nothing to do with TLS. Reading an alert
64/// out of the second produces a finding that never happened, which is worse
65/// than producing none.
66#[derive(Debug, Clone, Copy, PartialEq, Eq)]
67pub enum CodeSpace {
68 /// QUIC's own transport errors, where TLS alerts arrive as `0x0100 | alert`.
69 Transport,
70 /// The application's close codes — over MoQT, a draft's own error codes.
71 Application,
72}
73
74impl HandshakeFailure {
75 /// A failure carrying a transport-space code, with any TLS alert recovered.
76 ///
77 /// Prefer this over a struct literal: it is what keeps `tls_alert`
78 /// consistent with `code`, and a literal can set the two independently.
79 pub fn transport(code: u64, reason: String) -> Self {
80 Self {
81 code: Some(code),
82 code_space: Some(CodeSpace::Transport),
83 tls_alert: Self::alert_of(code),
84 reason,
85 }
86 }
87
88 /// A failure carrying an application-space close code.
89 ///
90 /// Never carries a TLS alert: application codes are a separate space, and
91 /// the ones that happen to land in `0x0100..=0x01ff` are the trap this
92 /// constructor exists to close.
93 pub fn application(code: u64, reason: String) -> Self {
94 Self { code: Some(code), code_space: Some(CodeSpace::Application), tls_alert: None, reason }
95 }
96
97 /// A failure with nothing numeric in it — a timeout, a DNS miss, a reset.
98 pub fn bare(reason: String) -> Self {
99 Self { code: None, code_space: None, tls_alert: None, reason }
100 }
101
102 /// Recover the alert from a **transport-space** code, if it carries one.
103 ///
104 /// `0x0100 | alert` is the mapping TLS-over-QUIC defines, so the range is
105 /// exactly one byte wide and the low byte is the alert. Applying this to an
106 /// application-space code is a bug — see [`CodeSpace`].
107 pub fn alert_of(code: u64) -> Option<u8> {
108 (0x0100..=0x01ff).contains(&code).then_some((code & 0xff) as u8)
109 }
110}
111
112impl std::fmt::Display for HandshakeFailure {
113 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114 f.write_str(&self.reason)
115 }
116}
117
118/// Errors from the transport layer.
119///
120/// `#[non_exhaustive]` so that a new failure mode is an additive change. The
121/// crate matches on these internally, where the attribute does not apply;
122/// downstream code that matches needs a `_` arm, which is the trade being made
123/// deliberately.
124#[derive(Debug, thiserror::Error)]
125#[non_exhaustive]
126pub enum TransportError {
127 /// Connection-level error (e.g., peer closed, timeout).
128 #[error("connection error: {0}")]
129 Connection(String),
130 /// Error writing to a stream.
131 #[error("write error: {0}")]
132 Write(String),
133 /// Error reading from a stream.
134 #[error("read error: {0}")]
135 Read(String),
136 /// Stream was closed.
137 #[error("stream closed")]
138 StreamClosed,
139 /// Error sending a datagram.
140 #[error("send datagram error: {0}")]
141 SendDatagram(String),
142 /// Connection was lost.
143 #[error("connection lost")]
144 ConnectionLost,
145 /// Error during connection establishment.
146 #[error("connect error: {0}")]
147 Connect(String),
148 /// A handshake reached the peer and did not complete, with its codes kept.
149 ///
150 /// Distinct from [`Connect`](Self::Connect), which is the same class of
151 /// event before there was anything numeric to keep — building the endpoint,
152 /// binding the socket, resolving a name. Both render the same way, so this
153 /// is additive to anything reading the message.
154 #[error("connect error: {0}")]
155 Handshake(HandshakeFailure),
156 /// The peer abandoned transmission by resetting the stream. Carries
157 /// the peer's application error code so a forwarder can mirror it
158 /// verbatim with [`SendStream::reset`].
159 #[error("stream reset by peer: code {0}")]
160 StreamReset(u64),
161 /// The peer is no longer accepting data on this stream
162 /// (`STOP_SENDING`). Carries the peer's application error code so a
163 /// forwarder can mirror it verbatim with [`RecvStream::stop`].
164 #[error("stream stopped by peer: code {0}")]
165 Stopped(u64),
166 /// The peer closed the whole session, with the code it named.
167 ///
168 /// Distinct from [`StreamReset`](Self::StreamReset), which ends one stream
169 /// and leaves the session alive. Over MoQT this code is a draft's own
170 /// session error — `VERSION_NEGOTIATION_FAILED`, `PROTOCOL_VIOLATION` — and
171 /// it is the sharpest thing a refusal says.
172 ///
173 /// It exists because quinn renders `ReadError::ConnectionLost` as the bare
174 /// words "connection lost" and drops the cause entirely, so without this
175 /// variant a relay that ends a session with a code *and* a reason phrase
176 /// reaches callers as neither. The code is on the value and also in the
177 /// message, spelled the way every other code this crate reports is spelled.
178 #[error("{reason} (code {code})")]
179 SessionClosed {
180 /// The peer's application close code.
181 code: u64,
182 /// quinn's own rendering of the close, which is the peer's reason
183 /// phrase where it sent one and the code where it did not.
184 reason: String,
185 },
186}
187
188/// A transport-agnostic connection (QUIC or WebTransport).
189pub enum Transport {
190 /// Raw QUIC via quinn.
191 Quic(quic::QuicTransport),
192 /// WebTransport via h3 + h3-quinn.
193 #[cfg(feature = "webtransport")]
194 WebTransport(webtransport::WebTransportTransport),
195}
196
197impl Transport {
198 /// Open a bidirectional stream.
199 pub async fn open_bi(&self) -> Result<(SendStream, RecvStream), TransportError> {
200 match self {
201 Transport::Quic(t) => t.open_bi().await,
202 #[cfg(feature = "webtransport")]
203 Transport::WebTransport(t) => t.open_bi().await,
204 }
205 }
206
207 /// Accept an incoming bidirectional stream.
208 pub async fn accept_bi(&self) -> Result<(SendStream, RecvStream), TransportError> {
209 match self {
210 Transport::Quic(t) => t.accept_bi().await,
211 #[cfg(feature = "webtransport")]
212 Transport::WebTransport(t) => t.accept_bi().await,
213 }
214 }
215
216 /// Open a unidirectional send stream.
217 pub async fn open_uni(&self) -> Result<SendStream, TransportError> {
218 match self {
219 Transport::Quic(t) => t.open_uni().await,
220 #[cfg(feature = "webtransport")]
221 Transport::WebTransport(t) => t.open_uni().await,
222 }
223 }
224
225 /// Accept an incoming unidirectional stream.
226 pub async fn accept_uni(&self) -> Result<RecvStream, TransportError> {
227 match self {
228 Transport::Quic(t) => t.accept_uni().await,
229 #[cfg(feature = "webtransport")]
230 Transport::WebTransport(t) => t.accept_uni().await,
231 }
232 }
233
234 /// Send a datagram.
235 pub fn send_datagram(&self, data: Bytes) -> Result<(), TransportError> {
236 match self {
237 Transport::Quic(t) => t.send_datagram(data),
238 #[cfg(feature = "webtransport")]
239 Transport::WebTransport(t) => t.send_datagram(data),
240 }
241 }
242
243 /// Receive a datagram.
244 pub async fn recv_datagram(&self) -> Result<Bytes, TransportError> {
245 match self {
246 Transport::Quic(t) => t.recv_datagram().await,
247 #[cfg(feature = "webtransport")]
248 Transport::WebTransport(t) => t.recv_datagram().await,
249 }
250 }
251
252 /// Close the connection.
253 pub fn close(&self, code: u32, reason: &[u8]) {
254 match self {
255 Transport::Quic(t) => t.close(code, reason),
256 #[cfg(feature = "webtransport")]
257 Transport::WebTransport(t) => t.close(code, reason),
258 }
259 }
260
261 /// A future resolving when the session ends, carrying the peer's close.
262 ///
263 /// Take it before handing the transport to a handshake: a peer refusing a
264 /// setup often closes the session after the handshake has already seen
265 /// end-of-stream.
266 ///
267 /// `Some` on every transport this crate dials. Over WebTransport only an
268 /// application or session close carries a code; a QUIC transport close
269 /// arrives as text, because `wtransport` keeps that code private.
270 pub fn closed(&self) -> Option<impl Future<Output = TransportError> + Send + 'static> {
271 let closed: std::pin::Pin<Box<dyn Future<Output = TransportError> + Send>> = match self {
272 Transport::Quic(t) => Box::pin(t.closed()),
273 #[cfg(feature = "webtransport")]
274 Transport::WebTransport(t) => Box::pin(t.closed()),
275 };
276 Some(closed)
277 }
278
279 /// The certificate chain the peer presented, DER-encoded, leaf first.
280 ///
281 /// Empty when there is none to report — the handshake has not completed,
282 /// the peer sent no chain — rather than an error, because none of those
283 /// are failures of this connection.
284 ///
285 /// **This is the whole of the certificate API, and that is deliberate.**
286 /// The bytes are handed over exactly as the peer sent them and nothing
287 /// here parses them, dates them, or decides whether they are trustworthy.
288 /// A certificate verdict is a judgement about what the caller is trying to
289 /// establish: a probe grading a public relay wants to know whether the
290 /// chain reaches a public root, an operator running against their own CA
291 /// wants precisely the opposite answer to count as healthy, and a library
292 /// that picked one would be silently wrong for the other. Whoever knows
293 /// the question owns the answer; this owns the evidence.
294 ///
295 /// One consequence worth naming, because it is the finding this exists to
296 /// make visible: the *length* of what comes back is itself a measurement.
297 /// A relay that serves a leaf and no intermediate hands back a chain of
298 /// one, which a client validating against a fixed root set rejects while a
299 /// browser holding a cached intermediate connects to it happily. That is a
300 /// misconfiguration and not a mystery — but only to a caller that can see
301 /// how many certificates arrived, which is why the chain is returned whole
302 /// rather than as a leaf.
303 ///
304 /// Read off the live connection each time — nothing is cached — so a
305 /// caller that intends to keep the chain must take it while it still holds
306 /// the `Transport`. Closing does not by itself erase the answer (quinn
307 /// keeps the crypto session for the handle's lifetime), but dropping the
308 /// handle does, and a caller that closes by value has done both.
309 pub fn peer_certificates(&self) -> Vec<Vec<u8>> {
310 match self {
311 Transport::Quic(t) => t.peer_certificates(),
312 #[cfg(feature = "webtransport")]
313 Transport::WebTransport(t) => t.peer_certificates(),
314 }
315 }
316
317 /// The application protocol the server selected in the CONNECT response.
318 ///
319 /// `None` over QUIC, where the equivalent answer is the ALPN and
320 /// [`dial_quic_to`] already returns it. Over WebTransport this is
321 /// `WT-Protocol`, which from draft-15 on is where MOQT settles its version
322 /// on that transport. See
323 /// [`WebTransportTransport::wt_protocol`](webtransport::WebTransportTransport::wt_protocol)
324 /// for how the value is read, and why an unquoted one is still read.
325 #[cfg(feature = "wt-protocol")]
326 pub fn wt_protocol(&self) -> Option<String> {
327 match self {
328 Transport::Quic(_) => None,
329 Transport::WebTransport(t) => t.wt_protocol(),
330 }
331 }
332}
333
334/// A transport-agnostic send stream.
335pub enum SendStream {
336 /// Raw QUIC send stream.
337 Quic(quinn::SendStream),
338 /// WebTransport send stream.
339 #[cfg(feature = "webtransport")]
340 WebTransport(webtransport::WtSendStream),
341}
342
343impl SendStream {
344 /// Get the QUIC stream ID (transport-level identifier).
345 pub fn stream_id(&self) -> u64 {
346 match self {
347 SendStream::Quic(s) => u64::from(s.id()),
348 #[cfg(feature = "webtransport")]
349 SendStream::WebTransport(s) => s.stream_id(),
350 }
351 }
352
353 /// Write all bytes to the stream.
354 ///
355 /// Fails with [`TransportError::Stopped`] carrying the peer's
356 /// application error code if the peer sent `STOP_SENDING`.
357 pub async fn write_all(&mut self, buf: &[u8]) -> Result<(), TransportError> {
358 match self {
359 SendStream::Quic(s) => s.write_all(buf).await.map_err(TransportError::from),
360 #[cfg(feature = "webtransport")]
361 SendStream::WebTransport(s) => s.write_all(buf).await,
362 }
363 }
364
365 /// Finish the stream (send FIN).
366 pub fn finish(&mut self) -> Result<(), TransportError> {
367 match self {
368 SendStream::Quic(s) => {
369 s.finish().map_err(|_| TransportError::StreamClosed)?;
370 Ok(())
371 }
372 #[cfg(feature = "webtransport")]
373 SendStream::WebTransport(s) => s.finish(),
374 }
375 }
376
377 /// Reset the stream, telling the peer transmission was abandoned and
378 /// handing it `code` as the `RESET_STREAM` application error code.
379 ///
380 /// This is the only way to abandon a send stream truthfully: simply
381 /// dropping a `SendStream` sends a FIN instead, which tells the peer
382 /// the stream ended *cleanly*. A forwarder that saw the far side
383 /// reset must call this with the code it received, so a truncated
384 /// stream is never laundered into a complete one.
385 ///
386 /// # Errors
387 /// - [`TransportError::StreamClosed`] if the stream was already finished or
388 /// reset.
389 /// - [`TransportError::Write`] if `code` is outside the QUIC varint range
390 /// (`0..2^62`). Nothing is sent in that case and the stream stays usable.
391 pub fn reset(&mut self, code: u64) -> Result<(), TransportError> {
392 match self {
393 SendStream::Quic(s) => {
394 let code = varint_code(code)?;
395 s.reset(code).map_err(|_| TransportError::StreamClosed)
396 }
397 #[cfg(feature = "webtransport")]
398 SendStream::WebTransport(s) => s.reset(code),
399 }
400 }
401
402 /// Set the stream's send priority.
403 ///
404 /// Streams with a higher priority have their locally buffered data
405 /// transmitted first. Every stream starts at priority 0.
406 ///
407 /// # Errors
408 /// [`TransportError::StreamClosed`] once the stream's send state has been
409 /// discarded — but only on the QUIC arm, and quinn keeps that state around
410 /// for a while after a `finish` or `reset`, so this is not a reliable *is
411 /// the stream still live?* probe. The WebTransport arm never reports it at
412 /// all: `wtransport` discards the underlying error and always succeeds. Do
413 /// not treat `Ok(())` as proof the priority took effect.
414 pub fn set_priority(&self, priority: i32) -> Result<(), TransportError> {
415 match self {
416 SendStream::Quic(s) => {
417 s.set_priority(priority).map_err(|_| TransportError::StreamClosed)
418 }
419 #[cfg(feature = "webtransport")]
420 SendStream::WebTransport(s) => s.set_priority(priority),
421 }
422 }
423
424 /// Resolve when this send half stops being useful.
425 ///
426 /// [`write_all`](Self::write_all) only reports `STOP_SENDING` when
427 /// there is something to write, so a forwarder that has gone idle —
428 /// the normal state of a stream waiting on its source — never learns
429 /// that the peer walked away. This is the watcher for that case: it
430 /// borrows nothing, so it can sit in a `select!` beside the read
431 /// branch for the stream's whole life.
432 /// The four outcomes, on quinn 0.11.9:
433 ///
434 /// - The peer sent `STOP_SENDING` → [`TransportError::Stopped`] carrying
435 /// the peer's application error code, the same typed value a failed
436 /// [`write_all`](Self::write_all) produces, so a forwarder can mirror it
437 /// verbatim with no new match arm.
438 /// - The stream was finished and the peer acked every byte → `Ok(())`.
439 /// quinn cannot tell that apart from *the send state was discarded*, so
440 /// `Ok(())` means *this stream is over*, never *the peer is happy*. It
441 /// cannot fire on a live stream.
442 /// - The connection was lost → [`TransportError::Connection`].
443 /// - **The local side reset the stream → this future never resolves.**
444 /// quinn keeps no stopped-notification for a stream it has locally reset,
445 /// so a watcher held across a [`reset`](Self::reset) stays pending until
446 /// the connection ends and holds a `tokio::sync::Notify` alive for that
447 /// long. Retire the future *before* resetting.
448 ///
449 /// The returned future is `'static`: it holds a handle on the
450 /// connection, not on `self`, so it may outlive this `SendStream`
451 /// and be spawned or stored on its own.
452 ///
453 /// # WebTransport arm
454 ///
455 /// Reaches the same quinn future through
456 /// `webtransport::WtSendStream::quic_stream` — spelled as code and not
457 /// as an intra-doc link because the `webtransport` module is behind its
458 /// own feature, so a link to it is broken in the default build and
459 /// `just doc-check` runs `cargo doc --workspace --no-deps` without it.
460 /// This bypasses `wtransport`'s own `stopped`, which collapses stopped /
461 /// closed / disconnected into one error. One asymmetry the QUIC arm
462 /// does not have: [`finish`](Self::finish) moves the inner
463 /// `wtransport` stream out, so calling this afterwards yields a
464 /// future that resolves immediately to
465 /// [`TransportError::StreamClosed`] — the "finished and acked" case
466 /// is unobservable there. Not yet exercised against a live
467 /// WebTransport session.
468 pub fn stopped(&self) -> impl Future<Output = Result<(), TransportError>> + Send + 'static {
469 // Resolve the arm eagerly so the returned future borrows nothing
470 // and both arms hand back the *same* quinn future type.
471 let watched: Result<_, TransportError> = match self {
472 SendStream::Quic(s) => Ok(s.stopped()),
473 #[cfg(feature = "webtransport")]
474 SendStream::WebTransport(s) => s.quic_stream().map(|q| q.stopped()),
475 };
476 async move { stopped_outcome(watched?.await) }
477 }
478}
479
480/// Map quinn's `stopped()` result onto [`TransportError`].
481///
482/// `Ok(None)` is quinn's *the send state is gone*, which it reports both for a
483/// finished-and-acked stream and for one whose state it discarded; neither is
484/// an error, so both become `Ok(())`.
485fn stopped_outcome(
486 outcome: Result<Option<quinn::VarInt>, quinn::StoppedError>,
487) -> Result<(), TransportError> {
488 match outcome {
489 Ok(Some(code)) => Err(TransportError::Stopped(code.into_inner())),
490 Ok(None) => Ok(()),
491 Err(e) => Err(TransportError::Connection(e.to_string())),
492 }
493}
494
495/// Convert an application error code to a quinn `VarInt`.
496///
497/// QUIC application error codes are varints, so values above
498/// 2^62 - 1 cannot be represented on the wire.
499fn varint_code(code: u64) -> Result<quinn::VarInt, TransportError> {
500 quinn::VarInt::from_u64(code)
501 .map_err(|_| TransportError::Write(format!("error code {code} exceeds the varint range")))
502}
503
504/// A transport-agnostic receive stream.
505pub enum RecvStream {
506 /// Raw QUIC receive stream.
507 Quic(quinn::RecvStream),
508 /// WebTransport receive stream.
509 #[cfg(feature = "webtransport")]
510 WebTransport(webtransport::WtRecvStream),
511}
512
513impl RecvStream {
514 /// Get the QUIC stream ID (transport-level identifier).
515 pub fn stream_id(&self) -> u64 {
516 match self {
517 RecvStream::Quic(s) => u64::from(s.id()),
518 #[cfg(feature = "webtransport")]
519 RecvStream::WebTransport(s) => s.stream_id(),
520 }
521 }
522
523 /// Read data into the buffer. Returns `Ok(Some(n))` with bytes read,
524 /// `Ok(None)` on stream end, or `Err` on failure.
525 ///
526 /// Fails with [`TransportError::StreamReset`] carrying the peer's
527 /// application error code if the peer reset the stream, which is how
528 /// callers distinguish an abandoned stream from a clean FIN
529 /// (`Ok(None)`).
530 pub async fn read(&mut self, buf: &mut [u8]) -> Result<Option<usize>, TransportError> {
531 match self {
532 RecvStream::Quic(s) => s.read(buf).await.map_err(TransportError::from),
533 #[cfg(feature = "webtransport")]
534 RecvStream::WebTransport(s) => s.read(buf).await,
535 }
536 }
537
538 /// Wait for the peer to reset this stream — **without reading a byte**.
539 ///
540 /// [`read`](Self::read) is the only other way to learn that a peer sent
541 /// `RESET_STREAM`, and it is unusable by a reader that has stopped
542 /// consuming on purpose: a forwarder applying backpressure holds its
543 /// source unread, so the reset surfaces on a call it is deliberately
544 /// not making, and the abandonment goes unobserved for as long as the
545 /// backpressure lasts. This observes the same event on its own.
546 ///
547 /// **It consumes nothing.** No bytes leave the receive buffer, so no
548 /// `MAX_STREAM_DATA` credit is granted and the peer stays flow-control
549 /// blocked exactly as it was. That is the whole point: it is safe to
550 /// poll *while* backpressure is being applied, which
551 /// [`read`](Self::read) is not.
552 ///
553 /// Cancel-safe: it registers interest and consumes no state, so
554 /// dropping the future loses nothing.
555 ///
556 /// # Returns
557 /// - `Ok(Some(code))` — the peer reset the stream with this application
558 /// error code. The same code [`TransportError::StreamReset`] would have
559 /// carried out of [`read`](Self::read).
560 /// - `Ok(None)` — **no reset is observable on this stream, now or ever**,
561 /// and the caller must stop asking: this resolves immediately every time,
562 /// so a caller that re-polls it in a loop spins. Either the transport
563 /// freed the stream's state (it was finished and fully read, or stopped)
564 /// or, on the WebTransport arm, `wtransport` exposes no reset-only
565 /// observable at all and this answers `Ok(None)` unconditionally.
566 /// - `Err` — a connection-level failure.
567 pub async fn received_reset(&mut self) -> Result<Option<u64>, TransportError> {
568 match self {
569 RecvStream::Quic(s) => match s.received_reset().await {
570 Ok(code) => Ok(code.map(|c| c.into_inner())),
571 Err(e) => Err(TransportError::Connection(e.to_string())),
572 },
573 // `wtransport::RecvStream` has no reset-only observable, so a
574 // WebTransport forwarder keeps the pre-existing behaviour: a
575 // peer reset is seen on the next `read` and not before.
576 #[cfg(feature = "webtransport")]
577 RecvStream::WebTransport(_) => Ok(None),
578 }
579 }
580
581 /// Stop accepting data on the stream, discarding anything unread and
582 /// telling the peer to stop transmitting with `code` as the
583 /// `STOP_SENDING` application error code.
584 ///
585 /// Dropping a `RecvStream` also stops it, but with a hard-coded code
586 /// of 0 — so a forwarder mirroring a peer's `STOP_SENDING` must call
587 /// this explicitly to keep the original code intact.
588 ///
589 /// After a successful call the stream is no longer readable, and the
590 /// two arms say so differently: the QUIC arm's [`read`](Self::read)
591 /// returns [`TransportError::Read`], the WebTransport arm's returns
592 /// [`TransportError::StreamClosed`] (the inner stream is consumed,
593 /// because `wtransport::RecvStream::stop` takes `self` by value).
594 /// Stop reading once you have stopped a stream rather than matching
595 /// on which error comes back.
596 ///
597 /// # Errors
598 /// - [`TransportError::StreamClosed`] if the stream was already stopped,
599 /// finished or reset.
600 /// - [`TransportError::Write`] if `code` is outside the QUIC varint range
601 /// (`0..2^62`) — `Write` because what failed is the `STOP_SENDING` frame
602 /// this endpoint would have sent. Nothing is sent in that case and the
603 /// stream stays readable.
604 pub fn stop(&mut self, code: u64) -> Result<(), TransportError> {
605 match self {
606 RecvStream::Quic(s) => {
607 let code = varint_code(code)?;
608 s.stop(code).map_err(|_| TransportError::StreamClosed)
609 }
610 #[cfg(feature = "webtransport")]
611 RecvStream::WebTransport(s) => s.stop(code),
612 }
613 }
614}