Skip to main content

zincio_http/h2/connection/
mod.rs

1//! HTTP/2 connection (RFC 9113 Sections 3.5, 5.1, 6.5, 8.1).
2//!
3//! Drives one HTTP/2 connection over an async I/O stream: reads the
4//! client's 24-octet preface (with a timeout), sends the server
5//! SETTINGS, maintains the peer's SETTINGS state (applying
6//! `SETTINGS_HEADER_TABLE_SIZE` to the HPACK encoder and
7//! `SETTINGS_MAX_FRAME_SIZE` to the outgoing frame writer), answers
8//! SETTINGS frames with ACK, echoes PING, and reports protocol
9//! violations with GOAWAY before closing.
10//!
11//! Frame parsing and validation happen in [`super::codec`]; this module
12//! adds the connection- and stream-level wiring: per-stream state,
13//! request/response header parsing, request dispatch to the handler,
14//! response framing, and the connection-level error handling that
15//! h2spec's `http2/4.3`, `http2/5.1`, `http2/6.1`, `http2/6.2`,
16//! `http2/6.4` and `http2/8.1` groups cover.
17
18use std::{
19    collections::VecDeque,
20    future::Future,
21    pin::Pin,
22    sync::{atomic::AtomicBool, Arc},
23    task::{Context, Poll},
24    time::Duration,
25};
26
27use bytes::Bytes;
28use futures_util::{pin_mut, FutureExt};
29use http::{Request, Response, StatusCode};
30use http_body::Body;
31use rustc_hash::{FxHashMap, FxHashSet};
32use tokio_util::sync::CancellationToken;
33
34use super::codec::{
35    Frame, FrameDecoder, FrameWriter, Setting, CLIENT_PREFACE, DEFAULT_INITIAL_WINDOW_SIZE,
36    DEFAULT_MAX_FRAME_SIZE, MAX_FRAME_SIZE_LIMIT,
37};
38use super::date::DateCache;
39use super::error::Reason;
40use super::hpack::{Decoder as HpackDecoder, Encoder, Header as HpackHeader, HpackError};
41use super::sanitize_response;
42use super::stream::{
43    BodyMsg, H2Body, MalformedRequest, ParsedRequest, StreamDriver, StreamEntry, StreamMsg,
44};
45use crate::early_hints::EarlyHints;
46use crate::Incoming;
47
48/// Per-connection behavior options.
49#[derive(Debug, Clone, Copy)]
50pub struct ConnectionOptions {
51    /// Answer requests with `100 Continue` when they carry
52    /// `expect: 100-continue`.
53    pub send_continue_response: bool,
54    /// Add a `Date` header to responses.
55    pub send_date_header: bool,
56    /// `SETTINGS_MAX_CONCURRENT_STREAMS` announced to the peer.
57    pub max_concurrent_streams: u32,
58    /// `SETTINGS_INITIAL_WINDOW_SIZE`: per-stream DATA credit we start with.
59    pub initial_stream_window_size: u32,
60    /// Connection-level DATA credit we start with (RFC 9113 Section 6.9.1).
61    pub initial_connection_window_size: u32,
62    /// Largest frame (payload) we send or receive (`SETTINGS_MAX_FRAME_SIZE`).
63    pub max_frame_size: u32,
64    /// Largest decoded header list we accept (`SETTINGS_MAX_HEADER_LIST_SIZE`).
65    pub max_header_list_size: u32,
66    /// Whether to enable Extended CONNECT
67    pub enable_connect_protocol: bool,
68    /// Close the connection after this long with no frame from the peer
69    /// (RFC 9113 Section 10.5). `None` disables the idle timeout.
70    pub idle_timeout: Option<Duration>,
71    /// Maximum number of RST_STREAM frames we send in response to
72    /// protocol errors made by the peer over the connection's lifetime.
73    /// `None` disables the limit; when it is exceeded the connection
74    /// closes with GOAWAY `ENHANCE_YOUR_CALM` (RFC 9113 Section 10.5.2).
75    pub max_local_error_reset_streams: Option<usize>,
76    /// Maximum number of streams the peer reset before we accepted them.
77    /// `None` disables the limit; when it is exceeded the connection
78    /// closes with GOAWAY `ENHANCE_YOUR_CALM` (RFC 9113 Section 10.5.2).
79    pub max_pending_accept_reset_streams: Option<usize>,
80    /// Maximum number of frames that may compose a single, not-yet-finalized
81    /// header field block (HEADERS/CONTINUATION). Beyond this a stream is
82    /// reset as a CONTINUATION flood (CVE-2024-27919).
83    pub max_continuation_frames: usize,
84}
85
86impl Default for ConnectionOptions {
87    #[inline]
88    fn default() -> Self {
89        ConnectionOptions {
90            send_continue_response: false,
91            send_date_header: true,
92            max_concurrent_streams: 100,
93            initial_stream_window_size: DEFAULT_INITIAL_WINDOW_SIZE,
94            initial_connection_window_size: DEFAULT_INITIAL_WINDOW_SIZE,
95            max_frame_size: DEFAULT_MAX_FRAME_SIZE as u32,
96            max_header_list_size: u32::MAX,
97            enable_connect_protocol: false,
98            idle_timeout: None,
99            max_local_error_reset_streams: Some(1024),
100            max_pending_accept_reset_streams: Some(20),
101            max_continuation_frames: 16,
102        }
103    }
104}
105
106/// The peer's current SETTINGS values. Defaults are the RFC 9113
107/// Section 6.5.2 initial values; they change as non-ACK SETTINGS frames
108/// arrive.
109#[derive(Debug, Clone, Copy, PartialEq, Eq)]
110pub struct PeerSettings {
111    pub(crate) header_table_size: u32,
112    pub(crate) enable_push: u32,
113    pub(crate) initial_window_size: u32,
114    pub(crate) max_frame_size: usize,
115    /// Kept for the field block decoder's bomb protection (C2).
116    #[allow(dead_code)]
117    pub(crate) max_header_list_size: u32,
118}
119
120impl Default for PeerSettings {
121    #[inline]
122    fn default() -> Self {
123        PeerSettings {
124            header_table_size: 4096,
125            enable_push: 1,
126            initial_window_size: DEFAULT_INITIAL_WINDOW_SIZE,
127            max_frame_size: DEFAULT_MAX_FRAME_SIZE,
128            max_header_list_size: u32::MAX,
129        }
130    }
131}
132
133/// An HTTP/2 server connection.
134///
135/// `Io` must be a raw transport: the preface is read byte-exact, so
136/// any buffering layer between the socket and this type breaks the
137/// initial read.
138pub struct Connection<Io> {
139    io: Io,
140    decoder: FrameDecoder,
141    writer: FrameWriter,
142    out: Vec<u8>,
143    /// Encoder for the field blocks this connection sends; its table
144    /// size follows the peer's `SETTINGS_HEADER_TABLE_SIZE`.
145    encoder: Encoder,
146    /// Decoder for the peer's field blocks; sees every header block on
147    /// this connection (RFC 9113 Section 4.3).
148    request_decoder: HpackDecoder,
149    /// The peer's settings, updated by non-ACK SETTINGS frames.
150    peer: PeerSettings,
151    /// The settings this connection announced (public RFC defaults
152    /// plus `SETTINGS_MAX_CONCURRENT_STREAMS`).
153    #[allow(dead_code)]
154    local: PeerSettings,
155    /// Bounds the wait for the client's 24-octet preface. A peer that
156    /// is too slow is disconnected without a GOAWAY.
157    preface_timeout: Option<Duration>,
158    /// Active streams, keyed by stream id (RFC 9113 Section 5.1).
159    streams: FxHashMap<u32, StreamEntry>,
160    /// Connection-level send window for DATA payloads (RFC 9113
161    /// Section 6.9.1); initial 65,535 octets.
162    conn_window: i64,
163    /// Stream ids whose streams have ended (closed state per RFC 9113
164    /// Section 5.1); used to tell closed-stream frames apart from
165    /// idle-stream frames.
166    closed_streams: FxHashSet<u32>,
167    /// LRU order for `closed_streams` to avoid bulk clear at 4096.
168    closed_order: VecDeque<u32>,
169    /// Behavior options for this connection (used by [`Connection::handle`]).
170    opts: ConnectionOptions,
171    /// RST_STREAM frames this endpoint has sent in response to the
172    /// peer's protocol errors (bounded by `opts.max_local_error_reset_streams`).
173    local_error_resets: usize,
174    /// Streams the peer reset before this endpoint accepted them
175    /// (bounded by `opts.max_pending_accept_reset_streams`).
176    pending_accept_resets: usize,
177    /// Wakes the drive loop when a stream task fills its outbound
178    /// channel; the loop drains channels between reads.
179    wake_tx: Option<kanal::AsyncSender<()>>,
180    /// Stream ids whose FIELD_BLOCK completed (END_HEADERS seen) and
181    /// awaits finalization; drained one per frame by
182    /// [`Connection::process_frames`] in FIFO order for fairness.
183    complete_blocks: VecDeque<u32>,
184    /// Maximum number of frames a single header field block may span before
185    /// it is treated as a CONTINUATION flood and reset (CVE-2024-27919).
186    max_continuation_frames: usize,
187    /// Scratch buffer reused by [`Connection::drain_pending_data`] to
188    /// snapshot stream ids before pumping (avoids a per-call
189    /// allocation).
190    drain_ids: Vec<u32>,
191    /// Highest stream id opened by the peer (RFC 9113 Section 5.1.1).
192    highest_stream_id: u32,
193    /// A connection error is pending; the loop stops after flushing.
194    closing: bool,
195    /// Graceful shutdown is in progress: the first GOAWAY is sent and we
196    /// only drain already-open streams until they finish or the drain
197    /// window elapses (RFC 9113 Section 6.8).
198    graceful: bool,
199    /// Last stream id advertised in the graceful-shutdown GOAWAY; peers
200    /// must not open streams beyond it.
201    graceful_last_stream: u32,
202    /// Optional token that triggers graceful shutdown when cancelled.
203    shutdown: Option<CancellationToken>,
204    /// Shared `Date` header value, refreshed periodically for the
205    /// responses the stream tasks emit.
206    date_cache: Arc<DateCache>,
207    /// Buffer for HTTP/2 frame encoding reuse
208    frame_buffer: Vec<u8>,
209}
210
211impl<Io> Connection<Io>
212where
213    Io: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
214{
215    /// Creates a connection over `io`.
216    #[inline]
217    pub fn new(io: Io, preface_timeout: Option<Duration>) -> Connection<Io> {
218        Connection {
219            io,
220            decoder: FrameDecoder::new(DEFAULT_MAX_FRAME_SIZE),
221            writer: FrameWriter::new(DEFAULT_MAX_FRAME_SIZE),
222            out: Vec::new(),
223            encoder: Encoder::new(4096),
224            request_decoder: HpackDecoder::new(4096),
225            peer: PeerSettings::default(),
226            local: PeerSettings::default(),
227            preface_timeout,
228            streams: FxHashMap::default(),
229            conn_window: DEFAULT_INITIAL_WINDOW_SIZE as i64,
230            closed_streams: FxHashSet::default(),
231            closed_order: VecDeque::with_capacity(4096),
232            opts: ConnectionOptions::default(),
233            local_error_resets: 0,
234            pending_accept_resets: 0,
235            wake_tx: None,
236            complete_blocks: VecDeque::new(),
237            max_continuation_frames: 16,
238            drain_ids: Vec::new(),
239            highest_stream_id: 0,
240            closing: false,
241            graceful: false,
242            graceful_last_stream: 0,
243            shutdown: None,
244            date_cache: Arc::new(DateCache::new()),
245            frame_buffer: Vec::new(),
246        }
247    }
248
249    /// Arms a [`CancellationToken`] that triggers a graceful shutdown
250    /// when cancelled: the connection sends GOAWAY, stops opening new
251    /// streams, drains in-flight responses, then closes (RFC 9113
252    /// Section 6.8).
253    #[inline]
254    pub fn with_shutdown(mut self, token: CancellationToken) -> Self {
255        self.shutdown = Some(token);
256        self
257    }
258
259    /// Drives a connection that never serves requests: the preface
260    /// handshake, SETTINGS/PING maintenance and error handling, but no
261    /// request dispatch (any request stream is refused).
262    ///
263    /// Equivalent to [`Connection::handle`] with a handler that never
264    /// completes; kept for tests and for callers that only want the
265    /// connection-level behavior.
266    #[inline]
267    pub async fn drive(self) -> std::io::Result<()> {
268        self.handle(
269            Arc::new(|_| std::future::pending::<Result<Response<Incoming>, std::io::Error>>()),
270            ConnectionOptions::default(),
271        )
272        .await
273    }
274
275    /// Serves requests to completion: peer EOF, GOAWAY received,
276    /// preface timeout, or an unrecoverable protocol error.
277    ///
278    /// Each decoded request starts a stream task running `request_fn`
279    /// (a clone-free sequential borrow: the loop owns the closure for
280    /// the connection's lifetime). Responses are framed onto the wire
281    /// by this task as the stream tasks emit them.
282    #[inline]
283    pub async fn handle<F, Fut, ResB, ResBE, ResE>(
284        mut self,
285        request_fn: Arc<F>,
286        options: ConnectionOptions,
287    ) -> std::io::Result<()>
288    where
289        F: Fn(Request<Incoming>) -> Fut + 'static,
290        Fut: Future<Output = Result<Response<ResB>, ResE>> + 'static,
291        ResB: Body<Data = Bytes, Error = ResBE> + Unpin + 'static,
292        ResBE: std::error::Error + 'static,
293        ResE: std::error::Error + 'static,
294    {
295        self.opts = options;
296        // Apply the negotiated native settings before the handshake.
297        self.request_decoder
298            .set_max_header_list_size(self.opts.max_header_list_size as usize);
299        self.decoder
300            .set_max_frame_size(self.opts.max_frame_size as usize);
301        self.conn_window = self.opts.initial_connection_window_size as i64;
302        // Resolved CONTINUATION-flood limit (CVE-2024-27919); `Http2Options`
303        // already applies the safe default when none was configured.
304        self.max_continuation_frames = options.max_continuation_frames;
305        match self.read_preface().await? {
306            None => return Ok(()), // preface timeout: close quietly
307            Some(false) => {
308                // Invalid preface: connection error PROTOCOL_ERROR
309                // (RFC 9113 Section 3.5), then close.
310                self.goaway(Reason::ProtocolError, b"invalid connection preface");
311                self.flush().await?;
312                return Ok(());
313            }
314            Some(true) => {}
315        }
316
317        // Our connection preface: SETTINGS announcing our flow-control
318        // windows, frame/header limits and concurrency (RFC 9113
319        // Sections 3.5 and 6.5.2).
320        self.writer.write_settings(
321            &mut self.out,
322            &[
323                Setting {
324                    id: 0x03,
325                    value: self.opts.max_concurrent_streams,
326                },
327                Setting {
328                    id: 0x04,
329                    value: self.opts.initial_stream_window_size,
330                },
331                Setting {
332                    id: 0x05,
333                    value: self.opts.max_frame_size,
334                },
335                Setting {
336                    id: 0x08,
337                    value: if self.opts.enable_connect_protocol {
338                        1
339                    } else {
340                        0
341                    },
342                },
343            ],
344        );
345        self.flush().await?;
346
347        let (wake_tx, wake_rx) = kanal::bounded_async(1);
348        self.wake_tx = Some(wake_tx);
349        // Pre-reserve output buffers to avoid per-response reallocations
350        // and enable single flush after drain_outbound + pending_data.
351        self.out.reserve(64 * 1024);
352        self.frame_buffer.reserve(self.opts.max_frame_size as usize);
353
354        let mut buf = [0u8; 8192];
355        let mut wake_rx_drain = Vec::new();
356        let mut peer_goaway = false;
357        while !peer_goaway && !(self.graceful && self.streams.is_empty()) {
358            let wake_recv = wake_rx.recv().fuse();
359            let read = tokio::io::AsyncReadExt::read(&mut self.io, &mut buf).fuse();
360            // Graceful-shutdown signal: never fires without a token. The
361            // clone lives for the loop iteration so the boxed future can
362            // borrow it.
363            let shutdown_token = self.shutdown.clone();
364            let shutdown_fut: Pin<Box<dyn futures_util::future::FusedFuture<Output = ()> + Send>> =
365                match &shutdown_token {
366                    Some(token) => Box::pin(token.cancelled().fuse()),
367                    None => Box::pin(futures_util::future::pending().fuse()),
368                };
369            pin_mut!(wake_recv);
370            pin_mut!(read);
371            pin_mut!(shutdown_fut);
372            // Idle timeout (RFC 9113 Section 10.5): no frame received from the
373            // peer within `idle_timeout` => graceful shutdown. Recreated each
374            // iteration so it measures the gap since the last received frame.
375            let timeout = self.opts.idle_timeout;
376            let idle_unfuse = std::pin::pin!(async move {
377                if let Some(d) = timeout {
378                    zincio::time::sleep(d).await;
379                } else {
380                    futures_util::future::pending::<()>().await;
381                }
382            });
383            let mut idle = idle_unfuse.fuse();
384            futures_util::select! {
385                n = read => {
386                    let n = match n {
387                        Ok(n) => n,
388                        Err(e) if self.streams.is_empty()
389                            && matches!(
390                                e.kind(),
391                                std::io::ErrorKind::BrokenPipe
392                                    | std::io::ErrorKind::ConnectionReset
393                                    | std::io::ErrorKind::ConnectionAborted
394                                    | std::io::ErrorKind::UnexpectedEof
395                            ) => {
396                            // Connection abruptly closed while idle (no streams)...
397                            return Ok(())
398                        }
399                        Err(e) => Err(e)?
400                    };
401                    if n == 0 {
402                        break; // peer closed; nothing more to say
403                    }
404                    self.decoder.extend(&buf[..n]);
405                    peer_goaway = self.process_frames(&request_fn).await?;
406                    self.drain_outbound();
407                    self.flush().await?;
408                }
409                _ = wake_recv => {
410                    // A stream task parked on a full channel; drain it.
411
412                    // But first, drain the wake notifications to prevent busy looping
413                    let _ = wake_rx.drain_into(&mut wake_rx_drain);
414                    wake_rx_drain.clear();
415                    self.drain_outbound();
416                    self.flush().await?;
417                }
418                _ = shutdown_fut => {
419                    self.begin_graceful_shutdown();
420                    self.flush().await?;
421                }
422                _ = idle => {
423                    // The peer was silent for `idle_timeout`; close the
424                    // connection gracefully (GOAWAY) and stop.
425                    self.begin_graceful_shutdown();
426                    self.flush().await?;
427                    break;
428                }
429            }
430        }
431        if self.graceful {
432            self.finish_graceful_shutdown();
433        }
434        self.flush().await?;
435        Ok(())
436    }
437
438    /// Reads the 24-octet client preface.
439    ///
440    /// Returns `Ok(None)` on timeout (the connection closes quietly —
441    /// answering a peer that never spoke is meaningless), `Ok(Some(
442    /// true))` on a match, and `Ok(Some(false))` when the peer sent
443    /// something else (the caller answers with GOAWAY).
444    #[inline]
445    async fn read_preface(&mut self) -> std::io::Result<Option<bool>> {
446        let mut magic = [0u8; CLIENT_PREFACE.len()];
447        match self.preface_timeout {
448            Some(timeout) => {
449                match zincio::time::timeout(
450                    timeout,
451                    tokio::io::AsyncReadExt::read_exact(&mut self.io, &mut magic),
452                )
453                .await
454                {
455                    Ok(result) => {
456                        result?;
457                    }
458                    Err(_elapsed) => return Ok(None),
459                }
460            }
461            None => {
462                tokio::io::AsyncReadExt::read_exact(&mut self.io, &mut magic).await?;
463            }
464        }
465        Ok(Some(magic == CLIENT_PREFACE))
466    }
467
468    /// Decodes and handles every frame currently buffered. Returns
469    /// `Ok(true)` when the connection should end (peer GOAWAY or an
470    /// error we answered with GOAWAY).
471    #[inline]
472    async fn process_frames<F, Fut, ResB, ResBE, ResE>(
473        &mut self,
474        request_fn: &Arc<F>,
475    ) -> std::io::Result<bool>
476    where
477        F: Fn(Request<Incoming>) -> Fut + 'static,
478        Fut: Future<Output = Result<Response<ResB>, ResE>> + 'static,
479        ResB: Body<Data = Bytes, Error = ResBE> + Unpin + 'static,
480        ResBE: std::error::Error + 'static,
481        ResE: std::error::Error + 'static,
482    {
483        loop {
484            let frame = match self.decoder.next_frame() {
485                Ok(Some(frame)) => frame,
486                Ok(None) => return Ok(false),
487                Err(error) => {
488                    // Frame-level violation: GOAWAY with the code the
489                    // codec determined (RFC 9113 Sections 6.10, 6.5.2),
490                    // then close.
491                    self.goaway(error.reason, b"frame error");
492                    self.flush().await?;
493                    return Ok(true);
494                }
495            };
496            match frame {
497                Frame::Settings {
498                    ack: false,
499                    settings,
500                } => {
501                    self.apply_peer_settings(&settings);
502                    self.writer.write_settings_ack(&mut self.out);
503                }
504                Frame::Settings { ack: true, .. } => {}
505                Frame::Ping {
506                    ack: false,
507                    payload,
508                } => {
509                    self.writer.write_ping_ack(&mut self.out, &payload);
510                }
511                Frame::Ping { ack: true, .. } => {}
512                Frame::GoAway { .. } => return Ok(true),
513                Frame::Headers {
514                    stream_id,
515                    end_stream,
516                    end_headers,
517                    block,
518                    ..
519                } => {
520                    self.handle_headers_frame(stream_id, end_stream, end_headers, &block);
521                }
522                Frame::Continuation {
523                    stream_id,
524                    end_headers,
525                    block,
526                } => {
527                    self.handle_continuation(stream_id, end_headers, &block);
528                }
529                Frame::Data {
530                    stream_id,
531                    end_stream,
532                    data,
533                } => {
534                    self.handle_data_frame(stream_id, end_stream, data).await;
535                }
536                Frame::Reset {
537                    stream_id,
538                    error_code,
539                } => {
540                    self.handle_reset_frame(stream_id, error_code);
541                }
542                Frame::Priority { .. } => {}
543                Frame::WindowUpdate {
544                    stream_id,
545                    increment,
546                } => self.handle_window_update(stream_id, increment),
547                Frame::PushPromise { .. } => {
548                    self.goaway(Reason::ProtocolError, b"push promise to server");
549                }
550                Frame::Unknown { .. } => {}
551            }
552            // One HEADERS/CONTINUATION chain completes at most one
553            // field block per frame arrival; act on it while the
554            // handler closure is in scope.
555            if let Some(id) = self.take_complete_block() {
556                self.finalize_field_block(id, request_fn).await;
557            }
558            if self.closing {
559                self.flush().await?;
560                return Ok(true);
561            }
562        }
563    }
564}
565/// A boxed response body: the handler's body type is erased at the
566/// stream boundary so `Connection` stays monomorphic.
567struct ConnBody {
568    inner: Pin<Box<dyn Body<Data = Bytes, Error = std::io::Error>>>,
569}
570
571impl ConnBody {
572    #[inline]
573    fn new<ResB, ResBE>(body: ResB) -> Self
574    where
575        ResB: Body<Data = Bytes, Error = ResBE> + Unpin + 'static,
576        ResBE: std::error::Error + 'static,
577    {
578        ConnBody {
579            inner: Box::pin(BodyAdapter(Some(Box::pin(body)))),
580        }
581    }
582}
583
584impl Body for ConnBody {
585    type Data = Bytes;
586    type Error = std::io::Error;
587
588    #[inline]
589    fn poll_frame(
590        mut self: Pin<&mut Self>,
591        cx: &mut Context<'_>,
592    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
593        self.inner.as_mut().poll_frame(cx)
594    }
595
596    #[inline]
597    fn size_hint(&self) -> http_body::SizeHint {
598        self.inner.size_hint()
599    }
600}
601
602/// Converts any displayable error into an [`io::Error`] without requiring
603/// `Send + Sync` (the native connection layer only needs the message, not the
604/// source; this keeps the public trait free of `Send`/`Sync` so it works on
605/// runtimes such as `zincio` that do not demand them).
606#[inline]
607fn e2io<E: std::fmt::Display>(e: E) -> std::io::Error {
608    #[derive(Debug)]
609    struct Msg(String);
610    impl std::fmt::Display for Msg {
611        #[inline]
612        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
613            f.write_str(&self.0)
614        }
615    }
616    impl std::error::Error for Msg {}
617    std::io::Error::other(Msg(format!("{e}")))
618}
619
620/// Adapts an arbitrary body whose error is not `io::Error` into one
621/// that is (the stream layer only deals in `io::Error`).
622struct BodyAdapter<ResB>(Option<Pin<Box<ResB>>>);
623
624impl<ResB, ResBE> Body for BodyAdapter<ResB>
625where
626    ResB: Body<Data = Bytes, Error = ResBE>,
627    ResBE: std::error::Error + 'static,
628{
629    type Data = Bytes;
630    type Error = std::io::Error;
631
632    #[inline]
633    fn poll_frame(
634        self: Pin<&mut Self>,
635        cx: &mut Context<'_>,
636    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
637        let this = self.get_mut();
638        let Some(inner) = this.0.as_mut() else {
639            return Poll::Ready(None);
640        };
641        match inner.as_mut().poll_frame(cx) {
642            Poll::Ready(Some(Ok(frame))) => Poll::Ready(Some(Ok(frame))),
643            Poll::Ready(Some(Err(error))) => Poll::Ready(Some(Err(e2io(error)))),
644            Poll::Ready(None) => {
645                this.0 = None;
646                Poll::Ready(None)
647            }
648            Poll::Pending => Poll::Pending,
649        }
650    }
651
652    #[inline]
653    fn size_hint(&self) -> http_body::SizeHint {
654        match &self.0 {
655            Some(body) => body.size_hint(),
656            None => http_body::SizeHint::default(),
657        }
658    }
659}
660
661#[derive(Clone, Copy, Debug, PartialEq, Eq)]
662enum StreamDataState {
663    Idle,
664    Closed,
665    Bad,
666    Gone,
667    Ok,
668}
669
670mod handlers;
671
672#[cfg(test)]
673mod tests;