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;