webtrans_quinn/
recv.rs

1//! WebTransport receive stream wrapper around `quinn::RecvStream`.
2
3use std::{
4    io,
5    pin::Pin,
6    task::{Context, Poll},
7};
8
9use bytes::Bytes;
10
11use crate::{ReadError, ReadExactError, ReadToEndError, SessionError};
12
13/// A stream that can be used to receive bytes. See [`quinn::RecvStream`].
14#[derive(Debug)]
15pub struct RecvStream {
16    inner: quinn::RecvStream,
17}
18
19impl RecvStream {
20    pub(crate) fn new(stream: quinn::RecvStream) -> Self {
21        Self { inner: stream }
22    }
23
24    /// Tell the peer to stop sending data with the given error code. See [`quinn::RecvStream::stop`].
25    /// WebTransport uses a u32 because it shares the error space with HTTP/3.
26    pub fn stop(&mut self, code: u32) -> Result<(), quinn::ClosedStream> {
27        let code = webtrans_proto::error_to_http3(code);
28        let code = quinn::VarInt::try_from(code).unwrap();
29        self.inner.stop(code)
30    }
31
32    // Wrap Quinn errors so they map into WebTransport error types.
33
34    /// Read some data into the buffer and return the amount read. See [`quinn::RecvStream::read`].
35    pub async fn read(&mut self, buf: &mut [u8]) -> Result<Option<usize>, ReadError> {
36        self.inner.read(buf).await.map_err(Into::into)
37    }
38
39    /// Fill the entire buffer with data. See [`quinn::RecvStream::read_exact`].
40    pub async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), ReadExactError> {
41        self.inner.read_exact(buf).await.map_err(Into::into)
42    }
43
44    /// Read a chunk of data from the stream. See [`quinn::RecvStream::read_chunk`].
45    pub async fn read_chunk(
46        &mut self,
47        max_length: usize,
48        ordered: bool,
49    ) -> Result<Option<quinn::Chunk>, ReadError> {
50        self.inner
51            .read_chunk(max_length, ordered)
52            .await
53            .map_err(Into::into)
54    }
55
56    /// Read chunks of data from the stream. See [`quinn::RecvStream::read_chunks`].
57    pub async fn read_chunks(&mut self, bufs: &mut [Bytes]) -> Result<Option<usize>, ReadError> {
58        self.inner.read_chunks(bufs).await.map_err(Into::into)
59    }
60
61    /// Read until the end of the stream or the limit is hit. See [`quinn::RecvStream::read_to_end`].
62    pub async fn read_to_end(&mut self, size_limit: usize) -> Result<Vec<u8>, ReadToEndError> {
63        self.inner.read_to_end(size_limit).await.map_err(Into::into)
64    }
65
66    /// Block until the stream has been reset and return the error code. See [`quinn::RecvStream::received_reset`].
67    ///
68    /// Unlike Quinn, this returns `SessionError` (not `ResetError`) because 0-RTT is not supported.
69    pub async fn received_reset(&mut self) -> Result<Option<u32>, SessionError> {
70        match self.inner.received_reset().await {
71            Ok(None) => Ok(None),
72            Ok(Some(code)) => Ok(webtrans_proto::error_from_http3(code.into_inner())),
73            Err(quinn::ResetError::ConnectionLost(e)) => Err(e.into()),
74            Err(quinn::ResetError::ZeroRttRejected) => unreachable!("0-RTT not supported"),
75        }
76    }
77
78    /// Return the underlying QUIC stream ID.
79    ///
80    /// > **Warning**
81    /// >
82    /// > WebTransport sessions share the QUIC connection with HTTP/3 and other sessions.
83    /// > The [quinn::StreamId::index] may not increment by 1 as it does in a
84    /// > standalone [quinn] connection. The JavaScript WebTransport API therefore
85    /// > does not expose stream IDs.
86    pub fn quic_id(&self) -> quinn::StreamId {
87        self.inner.id()
88    }
89
90    // 0-RTT is intentionally not exposed because it is invalid for WebTransport.
91}
92
93impl tokio::io::AsyncRead for RecvStream {
94    fn poll_read(
95        mut self: Pin<&mut Self>,
96        cx: &mut Context<'_>,
97        buf: &mut tokio::io::ReadBuf,
98    ) -> Poll<io::Result<()>> {
99        Pin::new(&mut self.inner).poll_read(cx, buf)
100    }
101}
102
103impl webtrans_trait::RecvStream for RecvStream {
104    type Error = ReadError;
105
106    fn stop(&mut self, code: u32) {
107        Self::stop(self, code).ok();
108    }
109
110    async fn read(&mut self, dst: &mut [u8]) -> Result<Option<usize>, Self::Error> {
111        self.read(dst).await
112    }
113
114    async fn read_chunk(&mut self, max: usize) -> Result<Option<Bytes>, Self::Error> {
115        self.read_chunk(max, true)
116            .await
117            .map(|r| r.map(|chunk| chunk.bytes))
118    }
119
120    async fn closed(&mut self) -> Result<(), Self::Error> {
121        self.received_reset().await?;
122        Ok(())
123    }
124}
webtrans_quinn/recv.rs

webtrans_quinn/
recv.rs
