h3/server/
stream.rs

1//! Server-side HTTP/3 stream management
2
3use bytes::Buf;
4
5use crate::{
6    error::{
7        connection_error_creators::CloseStream, internal_error::InternalConnectionError, Code,
8        StreamError,
9    },
10    quic::{self},
11    shared_state::{ConnectionState, SharedState},
12};
13
14use super::connection::RequestEnd;
15use std::sync::Arc;
16
17use std::{
18    option::Option,
19    result::Result,
20    task::{Context, Poll},
21};
22
23use bytes::BytesMut;
24use futures_util::future;
25use http::{response, HeaderMap, Response};
26
27use quic::StreamId;
28
29use crate::{
30    proto::{frame::Frame, headers::Header},
31    qpack,
32    quic::SendStream as _,
33    stream::{self},
34};
35
36#[cfg(feature = "tracing")]
37use tracing::{error, instrument};
38
39/// Manage request and response transfer for an incoming request
40///
41/// The [`RequestStream`] struct is used to send and/or receive
42/// information from the client.
43/// After sending the final response, call [`RequestStream::finish`] to close the stream.
44pub struct RequestStream<S, B> {
45    pub(super) inner: crate::connection::RequestStream<S, B>,
46    pub(super) request_end: Arc<RequestEnd>,
47}
48
49impl<S, B> AsMut<crate::connection::RequestStream<S, B>> for RequestStream<S, B> {
50    fn as_mut(&mut self) -> &mut crate::connection::RequestStream<S, B> {
51        &mut self.inner
52    }
53}
54
55impl<S, B> ConnectionState for RequestStream<S, B> {
56    fn shared_state(&self) -> &SharedState {
57        &self.inner.conn_state
58    }
59}
60
61impl<S, B> CloseStream for RequestStream<S, B> {}
62
63impl<S, B> RequestStream<S, B>
64where
65    S: quic::RecvStream,
66    B: Buf,
67{
68    /// Receive data sent from the client
69    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
70    pub async fn recv_data(&mut self) -> Result<Option<impl Buf>, StreamError> {
71        future::poll_fn(|cx| self.poll_recv_data(cx)).await
72    }
73
74    /// Poll for data sent from the client
75    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
76    pub fn poll_recv_data(
77        &mut self,
78        cx: &mut Context<'_>,
79    ) -> Poll<Result<Option<impl Buf>, StreamError>> {
80        self.inner.poll_recv_data(cx)
81    }
82
83    /// Receive an optional set of trailers for the request
84    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
85    pub async fn recv_trailers(&mut self) -> Result<Option<HeaderMap>, StreamError> {
86        future::poll_fn(|cx| self.poll_recv_trailers(cx)).await
87    }
88
89    /// Poll for an optional set of trailers for the request
90    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
91    pub fn poll_recv_trailers(
92        &mut self,
93        cx: &mut Context<'_>,
94    ) -> Poll<Result<Option<HeaderMap>, StreamError>> {
95        self.inner.poll_recv_trailers(cx)
96    }
97
98    /// Tell the peer to stop sending into the underlying QUIC stream
99    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
100    pub fn stop_sending(&mut self, error_code: Code) {
101        self.inner.stream.stop_sending(error_code)
102    }
103
104    /// Returns the underlying stream id
105    pub fn id(&self) -> StreamId {
106        self.inner.stream.id()
107    }
108}
109
110impl<S, B> RequestStream<S, B>
111where
112    S: quic::SendStream<B>,
113    B: Buf,
114{
115    /// Send the HTTP/3 response
116    ///
117    /// This should be called before trying to send any data with
118    /// [`RequestStream::send_data`].
119    pub async fn send_response(&mut self, resp: Response<()>) -> Result<(), StreamError> {
120        let (parts, _) = resp.into_parts();
121        let response::Parts {
122            status, headers, ..
123        } = parts;
124        let headers = Header::response(status, headers);
125
126        let mut block = BytesMut::new();
127        let mem_size = qpack::encode_stateless(&mut block, headers).map_err(|_e| {
128            self.handle_connection_error_on_stream(InternalConnectionError {
129                code: Code::H3_INTERNAL_ERROR,
130                message: "Failed to encode headers".to_string(),
131            })
132        })?;
133
134        let max_mem_size = self.inner.settings().max_field_section_size;
135
136        //= https://www.rfc-editor.org/rfc/rfc9114#section-4.2.2
137        //# An implementation that
138        //# has received this parameter SHOULD NOT send an HTTP message header
139        //# that exceeds the indicated size, as the peer will likely refuse to
140        //# process it.
141        //= https://www.rfc-editor.org/rfc/rfc9114#section-7.2.4.2
142        //# An HTTP implementation MUST NOT send frames or requests that would be
143        //# invalid based on its current understanding of the peer's settings.
144
145        if mem_size > max_mem_size {
146            return Err(StreamError::HeaderTooBig {
147                actual_size: mem_size,
148                max_size: max_mem_size,
149            });
150        }
151
152        stream::write(&mut self.inner.stream, Frame::Headers(block.freeze()))
153            .await
154            .map_err(|e| self.handle_quic_stream_error(e))?;
155
156        Ok(())
157    }
158
159    /// Send some data on the response body.
160    pub async fn send_data(&mut self, buf: B) -> Result<(), StreamError> {
161        self.inner.send_data(buf).await
162    }
163
164    /// Stop a stream with an error code
165    ///
166    /// The code can be [`Code::H3_NO_ERROR`].
167    pub fn stop_stream(&mut self, error_code: Code) {
168        self.inner.stop_stream(error_code);
169    }
170
171    /// Send a set of trailers to end the response.
172    ///
173    /// [`RequestStream::finish`] must be called to finalize a request.
174    pub async fn send_trailers(&mut self, trailers: HeaderMap) -> Result<(), StreamError> {
175        self.inner.send_trailers(trailers).await
176    }
177
178    /// End the response without trailers.
179    ///
180    /// [`RequestStream::finish`] must be called to finalize a request.
181    pub async fn finish(&mut self) -> Result<(), StreamError> {
182        self.inner.finish().await
183    }
184
185    //= https://www.rfc-editor.org/rfc/rfc9114#section-4.1.1
186    //= type=TODO
187    //# Implementations SHOULD cancel requests by abruptly terminating any
188    //# directions of a stream that are still open.  To do so, an
189    //# implementation resets the sending parts of streams and aborts reading
190    //# on the receiving parts of streams; see Section 2.4 of
191    //# [QUIC-TRANSPORT].
192
193    /// Returns the underlying stream id
194    pub fn send_id(&self) -> StreamId {
195        self.inner.stream.send_id()
196    }
197}
198
199impl<S, B> RequestStream<S, B>
200where
201    S: quic::BidiStream<B>,
202    B: Buf,
203{
204    /// Splits the Request-Stream into send and receive.
205    /// This can be used the send and receive data on different tasks.
206    pub fn split(
207        self,
208    ) -> (
209        RequestStream<S::SendStream, B>,
210        RequestStream<S::RecvStream, B>,
211    ) {
212        let (send, recv) = self.inner.split();
213        (
214            RequestStream {
215                inner: send,
216                request_end: self.request_end.clone(),
217            },
218            RequestStream {
219                inner: recv,
220                request_end: self.request_end,
221            },
222        )
223    }
224}
225
226impl Drop for RequestEnd {
227    fn drop(&mut self) {
228        if let Err(_error) = self.request_end.send(self.stream_id) {
229            #[cfg(feature = "tracing")]
230            error!(
231                "failed to notify connection of request end: {} {}",
232                self.stream_id, _error
233            );
234        }
235    }
236}
h3/server/stream.rs

h3/server/
stream.rs
