1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
//! WebTransport send stream wrapper around `quinn::SendStream`.
use std::{
io,
pin::Pin,
task::{Context, Poll},
};
use bytes::{Buf, Bytes};
use crate::{ClosedStream, SessionError, WriteError};
/// A stream that can be used to send bytes. See [`quinn::SendStream`].
///
/// This wrapper exists primarily to adapt error codes. WebTransport uses u32 error
/// codes that map into the reserved HTTP/3 error space.
#[derive(Debug)]
pub struct SendStream {
stream: quinn::SendStream,
}
impl SendStream {
pub(crate) fn new(stream: quinn::SendStream) -> Self {
Self { stream }
}
/// Abruptly reset the stream with the provided error code. See [`quinn::SendStream::reset`].
/// WebTransport uses a u32 because it shares the error space with HTTP/3.
pub fn reset(&mut self, code: u32) -> Result<(), ClosedStream> {
let code = webtrans_proto::error_to_http3(code);
let code = quinn::VarInt::try_from(code).unwrap();
self.stream.reset(code).map_err(Into::into)
}
/// Wait until the stream has been stopped and return the error code. See [`quinn::SendStream::stopped`].
///
/// Unlike Quinn, this returns `None` when the code is not a valid WebTransport error code.
/// It also returns `SessionError` (not `StoppedError`) because 0-RTT is not supported.
pub async fn stopped(&self) -> Result<Option<u32>, SessionError> {
match self.stream.stopped().await {
Ok(Some(code)) => Ok(webtrans_proto::error_from_http3(code.into_inner())),
Ok(None) => Ok(None),
Err(quinn::StoppedError::ConnectionLost(e)) => Err(e.into()),
Err(quinn::StoppedError::ZeroRttRejected) => unreachable!("0-RTT not supported"),
}
}
// Wrap Quinn errors so they map into WebTransport error types.
/// Write some data to the stream, returning the size written. See [`quinn::SendStream::write`].
pub async fn write(&mut self, buf: &[u8]) -> Result<usize, WriteError> {
self.stream.write(buf).await.map_err(Into::into)
}
/// Write all of the data to the stream. See [`quinn::SendStream::write_all`].
pub async fn write_all(&mut self, buf: &[u8]) -> Result<(), WriteError> {
self.stream.write_all(buf).await.map_err(Into::into)
}
/// Write chunks of data to the stream. See [`quinn::SendStream::write_chunks`].
pub async fn write_chunks(&mut self, bufs: &mut [Bytes]) -> Result<quinn::Written, WriteError> {
self.stream.write_chunks(bufs).await.map_err(Into::into)
}
/// Write a chunk of data to the stream. See [`quinn::SendStream::write_chunk`].
pub async fn write_chunk(&mut self, buf: Bytes) -> Result<(), WriteError> {
self.stream.write_chunk(buf).await.map_err(Into::into)
}
/// Write all of the chunks of data to the stream. See [`quinn::SendStream::write_all_chunks`].
pub async fn write_all_chunks(&mut self, bufs: &mut [Bytes]) -> Result<(), WriteError> {
self.stream.write_all_chunks(bufs).await.map_err(Into::into)
}
/// Mark the stream as finished so no more data can be written. See [`quinn::SendStream::finish`].
///
/// WARNING: Quinn implicitly calls this on drop. Dropping futures can lead to
/// incomplete writes, so prefer explicit shutdown when possible.
pub fn finish(&mut self) -> Result<(), ClosedStream> {
self.stream.finish().map_err(Into::into)
}
/// Set stream scheduling priority.
///
/// Lower values are generally scheduled earlier by QUIC.
pub fn set_priority(&self, order: i32) -> Result<(), ClosedStream> {
self.stream.set_priority(order).map_err(Into::into)
}
/// Get the current stream scheduling priority.
pub fn priority(&self) -> Result<i32, ClosedStream> {
self.stream.priority().map_err(Into::into)
}
/// Return the underlying QUIC stream ID.
///
/// > **Warning**
/// >
/// > WebTransport sessions share the QUIC connection with HTTP/3 and other sessions.
/// > The [quinn::StreamId::index] may not increment by 1 as it does in a
/// > standalone [quinn] connection. The JavaScript WebTransport API therefore
/// > does not expose stream IDs.
pub fn quic_id(&self) -> quinn::StreamId {
self.stream.id()
}
}
impl tokio::io::AsyncWrite for SendStream {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
// Use the trait method because Quinn defines its own `poll_write`.
tokio::io::AsyncWrite::poll_write(Pin::new(&mut self.stream), cx, buf)
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
Pin::new(&mut self.stream).poll_flush(cx)
}
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
Pin::new(&mut self.stream).poll_shutdown(cx)
}
}
impl webtrans_trait::SendStream for SendStream {
type Error = WriteError;
fn set_priority(&mut self, order: u8) {
Self::set_priority(self, order.into()).ok();
}
fn reset(&mut self, code: u32) {
Self::reset(self, code).ok();
}
fn finish(&mut self) -> Result<(), Self::Error> {
Self::finish(self).map_err(|_| WriteError::ClosedStream)
}
async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
Self::write(self, buf).await
}
async fn write_buf<B: Buf + Send>(&mut self, buf: &mut B) -> Result<usize, Self::Error> {
// This avoids a copy when the buffer is already `Bytes`, since Quinn allocates anyway.
let size = buf.chunk().len();
let chunk = buf.copy_to_bytes(size);
self.write_chunk(chunk).await?;
Ok(size)
}
async fn write_chunk(&mut self, chunk: Bytes) -> Result<(), Self::Error> {
self.write_chunk(chunk).await
}
async fn closed(&mut self) -> Result<(), Self::Error> {
// NOTE: Older Quinn versions required `&mut` for `stopped`.
match self.stopped().await? {
Some(code) => Err(WriteError::Stopped(code)),
None => Ok(()),
}
}
}