vibeio_http/h2/
mod.rs

1use http::{Request, Response};
2use http_body::Body;
3use tokio_util::sync::CancellationToken;
4
5use crate::h2::connection::{Connection, ConnectionOptions};
6use crate::h2::date::DateCache;
7use crate::h2::options::Http2Options;
8use crate::{HttpProtocol, Incoming};
9
10pub mod codec;
11pub mod connection;
12mod date;
13pub mod error;
14pub mod hpack;
15pub mod options;
16mod stream;
17
18pub(crate) use stream::H2Body;
19
20/// Header fields a server must strip from a response before sending it
21/// on an HTTP/2 connection (RFC 9113 Section 8.1.2.2): these are
22/// connection-level concerns, not per-message headers.
23pub(crate) const HTTP2_INVALID_HEADERS: [http::header::HeaderName; 5] = [
24    http::header::HeaderName::from_static("keep-alive"),
25    http::header::HeaderName::from_static("proxy-connection"),
26    http::header::CONNECTION,
27    http::header::TRANSFER_ENCODING,
28    http::header::UPGRADE,
29];
30
31/// Mangles a response into HTTP/2-legal shape: injects a `Date` header
32/// when configured and removes connection-specific response headers.
33#[inline]
34pub(super) fn sanitize_response<ResB>(
35    response: &mut Response<ResB>,
36    send_date_header: bool,
37    date_cache: &DateCache,
38) where
39    ResB: Body<Data = bytes::Bytes>,
40{
41    let response_headers = response.headers_mut();
42    if send_date_header {
43        if let Some(http_date) = date_cache.get_date_header_value() {
44            response_headers
45                .entry(http::header::DATE)
46                .or_insert(http_date);
47        }
48    }
49    for header in &HTTP2_INVALID_HEADERS {
50        if let http::header::Entry::Occupied(entry) = response_headers.entry(header) {
51            entry.remove();
52        }
53    }
54    if response_headers
55        .get(http::header::TE)
56        .is_some_and(|v| v != "trailers")
57    {
58        response_headers.remove(http::header::TE);
59    }
60}
61
62/// An HTTP/2 connection handler.
63///
64/// `Http2` wraps an async I/O stream (`Io`) and drives the HTTP/2 server
65/// connection using the native implementation in [`connection`]. It
66/// supports:
67///
68/// - Concurrent request stream handling
69/// - Streaming request/response bodies and trailers
70/// - Automatic `100 Continue` and `103 Early Hints` interim responses
71/// - Per-connection `Date` header caching
72/// - Graceful shutdown via a [`CancellationToken`]
73///
74/// # Construction
75///
76/// ```rust,ignore
77/// let http2 = Http2::new(tcp_stream, Http2Options::default());
78/// ```
79///
80/// # Serving requests
81///
82/// Use the [`HttpProtocol`] trait methods ([`handle`](HttpProtocol::handle) /
83/// [`handle_with_error_fn`](HttpProtocol::handle_with_error_fn)) to drive the
84/// connection to completion.
85pub struct Http2<Io> {
86    io_to_handshake: Option<Io>,
87    options: Http2Options,
88    cancel_token: Option<CancellationToken>,
89}
90
91impl<Io> Http2<Io>
92where
93    Io: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + 'static,
94{
95    /// Creates a new `Http2` connection handler wrapping the given I/O stream.
96    ///
97    /// The `options` value controls HTTP/2 protocol configuration, handshake
98    /// and accept timeouts, and optional behaviour such as automatic
99    /// `100 Continue` responses; see [`Http2Options`] for details.
100    ///
101    /// # Example
102    ///
103    /// ```rust,ignore
104    /// let http2 = Http2::new(tcp_stream, Http2Options::default());
105    /// ```
106    #[inline]
107    pub fn new(io: Io, options: Http2Options) -> Self {
108        Self {
109            io_to_handshake: Some(io),
110            options,
111            cancel_token: None,
112        }
113    }
114
115    /// Attaches a [`CancellationToken`] for graceful shutdown.
116    ///
117    /// When the token is cancelled, the handler sends HTTP/2 graceful shutdown
118    /// signals (GOAWAY), stops accepting new streams, and exits cleanly.
119    #[inline]
120    pub fn graceful_shutdown_token(mut self, token: CancellationToken) -> Self {
121        self.cancel_token = Some(token);
122        self
123    }
124}
125
126impl<Io> HttpProtocol for Http2<Io>
127where
128    Io: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + 'static,
129{
130    #[inline]
131    async fn handle<F, Fut, ResB, ResBE, ResE>(self, request_fn: F) -> Result<(), std::io::Error>
132    where
133        F: Fn(Request<Incoming>) -> Fut + 'static,
134        Fut: std::future::Future<Output = Result<Response<ResB>, ResE>> + 'static,
135        ResB: http_body::Body<Data = bytes::Bytes, Error = ResBE> + Unpin + 'static,
136        ResE: std::error::Error + 'static,
137        ResBE: std::error::Error + 'static,
138    {
139        let preface_timeout = self.options.handshake_timeout;
140        let options = ConnectionOptions {
141            send_continue_response: self.options.send_continue_response,
142            send_date_header: self.options.send_date_header,
143            max_concurrent_streams: self.options.max_concurrent_streams,
144            initial_stream_window_size: self.options.initial_stream_window_size,
145            initial_connection_window_size: self.options.initial_connection_window_size,
146            max_frame_size: self.options.max_frame_size,
147            max_header_list_size: self.options.max_header_list_size,
148            enable_connect_protocol: self.options.enable_connect_protocol,
149            idle_timeout: self.options.idle_timeout,
150        };
151        // The trait hands us a plain `Fn`; the native connection needs
152        // a `Clone` closure (it is reused across streams). Wrap it in an
153        // `Arc` so the spawned task can own a cheap clone.
154        let shared = std::sync::Arc::new(request_fn);
155
156        let connection = Connection::new(
157            self.io_to_handshake
158                .ok_or_else(|| std::io::Error::other("no io to handshake"))?,
159            preface_timeout,
160        );
161        let connection = if let Some(token) = self.cancel_token {
162            connection.with_shutdown(token)
163        } else {
164            connection
165        };
166        connection.handle(shared, options).await
167    }
168}
vibeio_http/h2/mod.rs

vibeio_http/h2/
mod.rs
