Repository: hyperium/hyper
Pull request #3607 (2024-03-21 07:41:18)

forward implement Service for reference types
simplifies the usage of `dyn Service`s and making `Service` types `Clone` by wrapping them in a `Rc`or `Arc`


Patch:
diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml
index bd03dbaf..fa765ae9 100644
--- a/.github/workflows/CI.yml
+++ b/.github/workflows/CI.yml
@@ -23,6 +23,7 @@ jobs:
       - features
       - ffi
       - ffi-header
+      - unexpected-cfgs
       - doc
       - check-external-types
       - udeps
@@ -220,6 +221,22 @@ jobs:
       - name: Ensure that hyper.h is up to date
         run: ./capi/gen_header.sh --verify
 
+  unexpected-cfgs:
+    runs-on: ubuntu-latest
+    needs: [style]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@nightly
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo check --all-features
+        env:
+          RUSTFLAGS: >-
+            -D unexpected_cfgs
+            --cfg hyper_internal_check_unexpected_cfgs
+            --cfg hyper_unstable_tracing
+            --cfg hyper_unstable_ffi
+            --check-cfg=cfg(hyper_internal_check_unexpected_cfgs,hyper_unstable_tracing,hyper_unstable_ffi)
+
   doc:
     name: Build docs
     needs: [style, test]
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5f24b182..a2a2683a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,34 @@
+### v1.3.1 (2024-04-16)
+
+#### Bug Fixes
+
+* **client:** revert auto content-length header for some requests (#3633)
+
+
+## v1.3.0 (2024-04-15)
+
+
+#### Bug Fixes
+
+* **client:** send content-length even with no body ([172fdfaf](https://github.com/hyperium/hyper/commit/172fdfaf0e0d9222917f271a83339238082e2657))
+* **http2:**
+  * `max_header_list_size(num)` defaults to 16kb ([203d1b09](https://github.com/hyperium/hyper/commit/203d1b090d0d0349c7e373e881ac4ddba72129be))
+  * `initial_max_send_streams` defaults to 100 ([2d1bd708](https://github.com/hyperium/hyper/commit/2d1bd7085e37a55ed6393f0e3f1b9a0b06db4d5d))
+* **server:**
+  * avoid unwrapping for the `Future` impl of HTTP/1 `UpgradeableConnection` (#3627) ([b79be911](https://github.com/hyperium/hyper/commit/b79be911696f6a93e8d408080ebbf558b612ce3c), closes [#3621](https://github.com/hyperium/hyper/issues/3621))
+  * avoid  `graceful_shutdown` panic on upgraded H1 connection (#3616) ([6ecf8521](https://github.com/hyperium/hyper/commit/6ecf85218fb24531184c53d5ed0eb7caf13cdcef))
+
+
+#### Features
+
+* **client:**
+  * add `max_header_list_size(num)` to `http2::Builder`. ([1c5b1b87](https://github.com/hyperium/hyper/commit/1c5b1b87ae1497a702e30ea82a486fb61a3f8133))
+  * add `max_pending_accept_reset_streams` HTTP2 option (#3617) ([330ddf1d](https://github.com/hyperium/hyper/commit/330ddf1de1ca2841469d30d24143902e5ff06365))
+* **ext:** implement From ReasonPhrase for Bytes ([dc27043a](https://github.com/hyperium/hyper/commit/dc27043aa319c0e630b7385a36aca0f3bee70670))
+* **service:** expose Service and HttpService trait unconditionally ([6aee2e6e](https://github.com/hyperium/hyper/commit/6aee2e6e260e7d407256d6b7da6a0d90c1bb9c67))
+* **server:** relax `'static` from connection IO trait bounds  (#3595) ([0013bdda](https://github.com/hyperium/hyper/commit/0013bdda5cd34ed6fca089eceb0133395b7be041))
+
+
 ## v1.2.0 (2024-02-21)
 
 
diff --git a/Cargo.toml b/Cargo.toml
index 860b10c3..a0c02d8c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "hyper"
-version = "1.2.0"
+version = "1.3.1"
 description = "A fast and correct HTTP library."
 readme = "README.md"
 homepage = "https://hyper.rs"
@@ -20,7 +20,7 @@ include = [
 ]
 
 [dependencies]
-bytes = "1"
+bytes = "1.2"
 http = "1"
 http-body = "1"
 tokio = { version = "1", features = ["sync"] }
diff --git a/benches/support/tokiort.rs b/benches/support/tokiort.rs
index 6b0948e9..0beea038 100644
--- a/benches/support/tokiort.rs
+++ b/benches/support/tokiort.rs
@@ -49,6 +49,13 @@ impl Timer for TokioTimer {
     }
 }
 
+impl TokioTimer {
+    /// Create a new TokioTimer
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
 // Use TokioSleep to get tokio::time::Sleep to implement Unpin.
 // see https://docs.rs/tokio/latest/tokio/time/struct.Sleep.html
 pin_project! {
diff --git a/examples/hello.rs b/examples/hello.rs
index 3a094829..17284a75 100644
--- a/examples/hello.rs
+++ b/examples/hello.rs
@@ -53,7 +53,7 @@ pub async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
             // Handle the connection from the client using HTTP1 and pass any
             // HTTP requests received on that connection to the `hello` function
             if let Err(err) = http1::Builder::new()
-                .timer(TokioTimer)
+                .timer(TokioTimer::new())
                 .serve_connection(io, service_fn(hello))
                 .await
             {
diff --git a/examples/send_file.rs b/examples/send_file.rs
index d369e1af..b2ef328a 100644
--- a/examples/send_file.rs
+++ b/examples/send_file.rs
@@ -49,7 +49,7 @@ async fn response_examples(
     match (req.method(), req.uri().path()) {
         (&Method::GET, "/") | (&Method::GET, "/index.html") => simple_file_send(INDEX).await,
         (&Method::GET, "/no_file.html") => {
-            // Test what happens when file cannot be be found
+            // Test what happens when file cannot be found
             simple_file_send("this_file_should_not_exist.html").await
         }
         _ => Ok(not_found()),
diff --git a/examples/service_struct_impl.rs b/examples/service_struct_impl.rs
index 25f33fe1..d0be1eb7 100644
--- a/examples/service_struct_impl.rs
+++ b/examples/service_struct_impl.rs
@@ -65,8 +65,7 @@ impl Service<Request<IncomingBody>> for Svc {
                 "authors extraordinare! counter = {:?}",
                 self.counter
             )),
-            // Return the 404 Not Found for other routes, and don't increment counter.
-            _ => return Box::pin(async { mk_response("oh no! not found".into()) }),
+            _ => mk_response("oh no! not found".into()),
         };
 
         Box::pin(async { res })
diff --git a/src/body/incoming.rs b/src/body/incoming.rs
index e0beba61..dcfb71d5 100644
--- a/src/body/incoming.rs
+++ b/src/body/incoming.rs
@@ -317,6 +317,13 @@ impl Body for Incoming {
 
 impl fmt::Debug for Incoming {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        #[cfg(any(
+            all(
+                any(feature = "http1", feature = "http2"),
+                any(feature = "client", feature = "server")
+            ),
+            feature = "ffi"
+        ))]
         #[derive(Debug)]
         struct Streaming;
         #[derive(Debug)]
@@ -403,6 +410,19 @@ impl Sender {
             .map_err(|err| err.into_inner().expect("just sent Ok"))
     }
 
+    #[cfg(feature = "http1")]
+    pub(crate) fn try_send_trailers(
+        &mut self,
+        trailers: HeaderMap,
+    ) -> Result<(), Option<HeaderMap>> {
+        let tx = match self.trailers_tx.take() {
+            Some(tx) => tx,
+            None => return Err(None),
+        };
+
+        tx.send(trailers).map_err(Some)
+    }
+
     #[cfg(test)]
     pub(crate) fn abort(mut self) {
         self.send_error(crate::Error::new_body_write_aborted());
diff --git a/src/body/length.rs b/src/body/length.rs
index 80ba8acd..e5eab744 100644
--- a/src/body/length.rs
+++ b/src/body/length.rs
@@ -15,11 +15,11 @@ impl From<Option<u64>> for DecodedLength {
 }
 
 #[cfg(any(feature = "http1", feature = "http2", test))]
-const MAX_LEN: u64 = std::u64::MAX - 2;
+const MAX_LEN: u64 = u64::MAX - 2;
 
 impl DecodedLength {
-    pub(crate) const CLOSE_DELIMITED: DecodedLength = DecodedLength(::std::u64::MAX);
-    pub(crate) const CHUNKED: DecodedLength = DecodedLength(::std::u64::MAX - 1);
+    pub(crate) const CLOSE_DELIMITED: DecodedLength = DecodedLength(u64::MAX);
+    pub(crate) const CHUNKED: DecodedLength = DecodedLength(u64::MAX - 1);
     pub(crate) const ZERO: DecodedLength = DecodedLength(0);
 
     #[cfg(test)]
diff --git a/src/body/mod.rs b/src/body/mod.rs
index e3402083..7b71d98b 100644
--- a/src/body/mod.rs
+++ b/src/body/mod.rs
@@ -35,6 +35,10 @@ pub(crate) use self::incoming::Sender;
 pub(crate) use self::length::DecodedLength;
 
 mod incoming;
+#[cfg(all(
+    any(feature = "http1", feature = "http2"),
+    any(feature = "client", feature = "server")
+))]
 mod length;
 
 fn _assert_send_sync() {
diff --git a/src/client/conn/http1.rs b/src/client/conn/http1.rs
index 9bf0938d..2a9ca4f6 100644
--- a/src/client/conn/http1.rs
+++ b/src/client/conn/http1.rs
@@ -29,6 +29,7 @@ pub struct SendRequest<B> {
 /// This allows taking apart a `Connection` at a later time, in order to
 /// reclaim the IO object, and additional related pieces.
 #[derive(Debug)]
+#[non_exhaustive]
 pub struct Parts<T> {
     /// The original IO object used in the handshake.
     pub io: T,
@@ -41,7 +42,6 @@ pub struct Parts<T> {
     /// You will want to check for any existing bytes if you plan to continue
     /// communicating on the IO object.
     pub read_buf: Bytes,
-    _inner: (),
 }
 
 /// A future that processes all HTTP state for the IO object.
@@ -68,11 +68,7 @@ where
     /// Only works for HTTP/1 connections. HTTP/2 connections will panic.
     pub fn into_parts(self) -> Parts<T> {
         let (io, read_buf, _) = self.inner.into_inner();
-        Parts {
-            io,
-            read_buf,
-            _inner: (),
-        }
+        Parts { io, read_buf }
     }
 
     /// Poll the connection for completion, but without calling `shutdown`
@@ -589,11 +585,7 @@ mod upgrades {
             match ready!(Pin::new(&mut self.inner.as_mut().unwrap().inner).poll(cx)) {
                 Ok(proto::Dispatched::Shutdown) => Poll::Ready(Ok(())),
                 Ok(proto::Dispatched::Upgrade(pending)) => {
-                    let Parts {
-                        io,
-                        read_buf,
-                        _inner,
-                    } = self.inner.take().unwrap().into_parts();
+                    let Parts { io, read_buf } = self.inner.take().unwrap().into_parts();
                     pending.fulfill(Upgraded::new(io, read_buf));
                     Poll::Ready(Ok(()))
                 }
diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs
index 823dd87d..6d967713 100644
--- a/src/client/conn/http2.rs
+++ b/src/client/conn/http2.rs
@@ -342,6 +342,14 @@ where
         self
     }
 
+    /// Sets the max size of received header frames.
+    ///
+    /// Default is currently 16KB, but can change.
+    pub fn max_header_list_size(&mut self, max: u32) -> &mut Self {
+        self.h2_builder.max_header_list_size = max;
+        self
+    }
+
     /// Sets an interval for HTTP2 Ping frames should be sent to keep a
     /// connection alive.
     ///
@@ -398,11 +406,22 @@ where
     ///
     /// The value must be no larger than `u32::MAX`.
     pub fn max_send_buf_size(&mut self, max: usize) -> &mut Self {
-        assert!(max <= std::u32::MAX as usize);
+        assert!(max <= u32::MAX as usize);
         self.h2_builder.max_send_buffer_size = max;
         self
     }
 
+    /// Configures the maximum number of pending reset streams allowed before a GOAWAY will be sent.
+    ///
+    /// This will default to the default value set by the [`h2` crate](https://crates.io/crates/h2).
+    /// As of v0.4.0, it is 20.
+    ///
+    /// See <https://github.com/hyperium/hyper/issues/2877> for more information.
+    pub fn max_pending_accept_reset_streams(&mut self, max: impl Into<Option<usize>>) -> &mut Self {
+        self.h2_builder.max_pending_accept_reset_streams = max.into();
+        self
+    }
+
     /// Constructs a connection with the configured options and IO.
     /// See [`client::conn`](crate::client::conn) for more.
     ///
diff --git a/src/client/dispatch.rs b/src/client/dispatch.rs
index 60504e68..d0aefc0c 100644
--- a/src/client/dispatch.rs
+++ b/src/client/dispatch.rs
@@ -40,7 +40,7 @@ pub(crate) struct Sender<T, U> {
     /// without notice.
     #[cfg(feature = "http1")]
     buffered_once: bool,
-    /// The Giver helps watch that the the Receiver side has been polled
+    /// The Giver helps watch that the Receiver side has been polled
     /// when the queue is empty. This helps us know when a request and
     /// response have been fully processed, and a connection is ready
     /// for more.
diff --git a/src/error.rs b/src/error.rs
index 483291c4..9ad4c0e5 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -98,6 +98,7 @@ pub(super) enum Parse {
     #[cfg(feature = "http1")]
     Header(Header),
     #[cfg(any(feature = "http1", feature = "http2"))]
+    #[cfg_attr(feature = "http2", allow(unused))]
     TooLarge,
     Status,
     #[cfg(all(any(feature = "client", feature = "server"), feature = "http1"))]
diff --git a/src/ext/h1_reason_phrase.rs b/src/ext/h1_reason_phrase.rs
index 8837ae45..c6f52333 100644
--- a/src/ext/h1_reason_phrase.rs
+++ b/src/ext/h1_reason_phrase.rs
@@ -107,9 +107,9 @@ impl TryFrom<Bytes> for ReasonPhrase {
     }
 }
 
-impl Into<Bytes> for ReasonPhrase {
-    fn into(self) -> Bytes {
-        self.0
+impl From<ReasonPhrase> for Bytes {
+    fn from(reason: ReasonPhrase) -> Self {
+        reason.0
     }
 }
 
@@ -147,7 +147,7 @@ const fn is_valid_byte(b: u8) -> bool {
     //
     // The 0xFF comparison is technically redundant, but it matches the text of the spec more
     // clearly and will be optimized away.
-    #[allow(unused_comparisons)]
+    #[allow(unused_comparisons, clippy::absurd_extreme_comparisons)]
     const fn is_obs_text(b: u8) -> bool {
         0x80 <= b && b <= 0xFF
     }
diff --git a/src/lib.rs b/src/lib.rs
index 23087a0e..0e734aa5 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,5 +1,6 @@
 #![deny(missing_docs)]
 #![deny(missing_debug_implementations)]
+#![cfg_attr(not(hyper_internal_check_unexpected_cfgs), allow(unexpected_cfgs))]
 #![cfg_attr(test, deny(rust_2018_idioms))]
 #![cfg_attr(all(test, feature = "full"), deny(unreachable_pub))]
 #![cfg_attr(all(test, feature = "full"), deny(warnings))]
diff --git a/src/proto/h1/conn.rs b/src/proto/h1/conn.rs
index f880f97d..88b17720 100644
--- a/src/proto/h1/conn.rs
+++ b/src/proto/h1/conn.rs
@@ -11,6 +11,7 @@ use bytes::{Buf, Bytes};
 use futures_util::ready;
 use http::header::{HeaderValue, CONNECTION, TE};
 use http::{HeaderMap, Method, Version};
+use http_body::Frame;
 use httparse::ParserConfig;
 
 use super::io::Buffered;
@@ -62,6 +63,8 @@ where
                 #[cfg(feature = "server")]
                 h1_header_read_timeout_running: false,
                 #[cfg(feature = "server")]
+                date_header: true,
+                #[cfg(feature = "server")]
                 timer: Time::Empty,
                 preserve_header_case: false,
                 #[cfg(feature = "ffi")]
@@ -266,18 +269,27 @@ where
                 self.try_keep_alive(cx);
             }
         } else if msg.expect_continue && msg.head.version.gt(&Version::HTTP_10) {
-            self.state.reading = Reading::Continue(Decoder::new(msg.decode));
+            let h1_max_header_size = None; // TODO: remove this when we land h1_max_header_size support
+            self.state.reading = Reading::Continue(Decoder::new(
+                msg.decode,
+                self.state.h1_max_headers,
+                h1_max_header_size,
+            ));
             wants = wants.add(Wants::EXPECT);
         } else {
-            self.state.reading = Reading::Body(Decoder::new(msg.decode));
+            let h1_max_header_size = None; // TODO: remove this when we land h1_max_header_size support
+            self.state.reading = Reading::Body(Decoder::new(
+                msg.decode,
+                self.state.h1_max_headers,
+                h1_max_header_size,
+            ));
         }
 
         self.state.allow_trailer_fields = msg
             .head
             .headers
             .get(TE)
-            .map(|te_header| te_header == "trailers")
-            .unwrap_or(false);
+            .map_or(false, |te_header| te_header == "trailers");
 
         Poll::Ready(Some(Ok((msg.head, msg.decode, wants))))
     }
@@ -311,33 +323,41 @@ where
     pub(crate) fn poll_read_body(
         &mut self,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<io::Result<Bytes>>> {
+    ) -> Poll<Option<io::Result<Frame<Bytes>>>> {
         debug_assert!(self.can_read_body());
 
         let (reading, ret) = match self.state.reading {
             Reading::Body(ref mut decoder) => {
                 match ready!(decoder.decode(cx, &mut self.io)) {
-                    Ok(slice) => {
-                        let (reading, chunk) = if decoder.is_eof() {
-                            debug!("incoming body completed");
-                            (
-                                Reading::KeepAlive,
-                                if !slice.is_empty() {
-                                    Some(Ok(slice))
-                                } else {
-                                    None
-                                },
-                            )
-                        } else if slice.is_empty() {
-                            error!("incoming body unexpectedly ended");
-                            // This should be unreachable, since all 3 decoders
-                            // either set eof=true or return an Err when reading
-                            // an empty slice...
-                            (Reading::Closed, None)
+                    Ok(frame) => {
+                        if frame.is_data() {
+                            let slice = frame.data_ref().unwrap_or_else(|| unreachable!());
+                            let (reading, maybe_frame) = if decoder.is_eof() {
+                                debug!("incoming body completed");
+                                (
+                                    Reading::KeepAlive,
+                                    if !slice.is_empty() {
+                                        Some(Ok(frame))
+                                    } else {
+                                        None
+                                    },
+                                )
+                            } else if slice.is_empty() {
+                                error!("incoming body unexpectedly ended");
+                                // This should be unreachable, since all 3 decoders
+                                // either set eof=true or return an Err when reading
+                                // an empty slice...
+                                (Reading::Closed, None)
+                            } else {
+                                return Poll::Ready(Some(Ok(frame)));
+                            };
+                            (reading, Poll::Ready(maybe_frame))
+                        } else if frame.is_trailers() {
+                            (Reading::Closed, Poll::Ready(Some(Ok(frame))))
                         } else {
-                            return Poll::Ready(Some(Ok(slice)));
-                        };
-                        (reading, Poll::Ready(chunk))
+                            trace!("discarding unknown frame");
+                            (Reading::Closed, Poll::Ready(None))
+                        }
                     }
                     Err(e) => {
                         debug!("incoming body decode error: {}", e);
@@ -565,6 +585,8 @@ where
                 keep_alive: self.state.wants_keep_alive(),
                 req_method: &mut self.state.method,
                 title_case_headers: self.state.title_case_headers,
+                #[cfg(feature = "server")]
+                date_header: self.state.date_header,
             },
             buf,
         ) {
@@ -594,8 +616,7 @@ where
         let outgoing_is_keep_alive = head
             .headers
             .get(CONNECTION)
-            .map(connection_keep_alive)
-            .unwrap_or(false);
+            .map_or(false, connection_keep_alive);
 
         if !outgoing_is_keep_alive {
             match head.version {
@@ -861,6 +882,8 @@ struct State {
     #[cfg(feature = "server")]
     h1_header_read_timeout_running: bool,
     #[cfg(feature = "server")]
+    date_header: bool,
+    #[cfg(feature = "server")]
     timer: Time,
     preserve_header_case: bool,
     #[cfg(feature = "ffi")]
diff --git a/src/proto/h1/decode.rs b/src/proto/h1/decode.rs
index 05f3cc6e..275d6a9b 100644
--- a/src/proto/h1/decode.rs
+++ b/src/proto/h1/decode.rs
@@ -2,12 +2,14 @@ use std::error::Error as StdError;
 use std::fmt;
 use std::io;
 use std::task::{Context, Poll};
-use std::usize;
 
-use bytes::Bytes;
+use bytes::{BufMut, Bytes, BytesMut};
 use futures_util::ready;
+use http::{HeaderMap, HeaderName, HeaderValue};
+use http_body::Frame;
 
 use super::io::MemRead;
+use super::role::DEFAULT_MAX_HEADERS;
 use super::DecodedLength;
 
 use self::Kind::{Chunked, Eof, Length};
@@ -17,6 +19,11 @@ use self::Kind::{Chunked, Eof, Length};
 /// This limit is currentlty applied for the entire body, not per chunk.
 const CHUNKED_EXTENSIONS_LIMIT: u64 = 1024 * 16;
 
+/// Maximum number of bytes allowed for all trailer fields.
+///
+/// TODO: remove this when we land h1_max_header_size support
+const TRAILER_LIMIT: usize = 1024 * 16;
+
 /// Decoders to handle different Transfer-Encodings.
 ///
 /// If a message body does not include a Transfer-Encoding, it *should*
@@ -26,7 +33,7 @@ pub(crate) struct Decoder {
     kind: Kind,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq)]
+#[derive(Debug, Clone, PartialEq)]
 enum Kind {
     /// A Reader used when a Content-Length header is passed with a positive integer.
     Length(u64),
@@ -35,6 +42,10 @@ enum Kind {
         state: ChunkedState,
         chunk_len: u64,
         extensions_cnt: u64,
+        trailers_buf: Option<BytesMut>,
+        trailers_cnt: usize,
+        h1_max_headers: Option<usize>,
+        h1_max_header_size: Option<usize>,
     },
     /// A Reader used for responses that don't indicate a length or chunked.
     ///
@@ -81,12 +92,19 @@ impl Decoder {
         }
     }
 
-    pub(crate) fn chunked() -> Decoder {
+    pub(crate) fn chunked(
+        h1_max_headers: Option<usize>,
+        h1_max_header_size: Option<usize>,
+    ) -> Decoder {
         Decoder {
             kind: Kind::Chunked {
                 state: ChunkedState::new(),
                 chunk_len: 0,
                 extensions_cnt: 0,
+                trailers_buf: None,
+                trailers_cnt: 0,
+                h1_max_headers,
+                h1_max_header_size,
             },
         }
     }
@@ -97,9 +115,13 @@ impl Decoder {
         }
     }
 
-    pub(super) fn new(len: DecodedLength) -> Self {
+    pub(super) fn new(
+        len: DecodedLength,
+        h1_max_headers: Option<usize>,
+        h1_max_header_size: Option<usize>,
+    ) -> Self {
         match len {
-            DecodedLength::CHUNKED => Decoder::chunked(),
+            DecodedLength::CHUNKED => Decoder::chunked(h1_max_headers, h1_max_header_size),
             DecodedLength::CLOSE_DELIMITED => Decoder::eof(),
             length => Decoder::length(length.danger_len()),
         }
@@ -123,12 +145,12 @@ impl Decoder {
         &mut self,
         cx: &mut Context<'_>,
         body: &mut R,
-    ) -> Poll<Result<Bytes, io::Error>> {
+    ) -> Poll<Result<Frame<Bytes>, io::Error>> {
         trace!("decode; state={:?}", self.kind);
         match self.kind {
             Length(ref mut remaining) => {
                 if *remaining == 0 {
-                    Poll::Ready(Ok(Bytes::new()))
+                    Poll::Ready(Ok(Frame::data(Bytes::new())))
                 } else {
                     let to_read = *remaining as usize;
                     let buf = ready!(body.read_mem(cx, to_read))?;
@@ -143,37 +165,77 @@ impl Decoder {
                     } else {
                         *remaining -= num;
                     }
-                    Poll::Ready(Ok(buf))
+                    Poll::Ready(Ok(Frame::data(buf)))
                 }
             }
             Chunked {
                 ref mut state,
                 ref mut chunk_len,
                 ref mut extensions_cnt,
+                ref mut trailers_buf,
+                ref mut trailers_cnt,
+                ref h1_max_headers,
+                ref h1_max_header_size,
             } => {
+                let h1_max_headers = h1_max_headers.unwrap_or(DEFAULT_MAX_HEADERS);
+                let h1_max_header_size = h1_max_header_size.unwrap_or(TRAILER_LIMIT);
                 loop {
                     let mut buf = None;
                     // advances the chunked state
-                    *state = ready!(state.step(cx, body, chunk_len, extensions_cnt, &mut buf))?;
+                    *state = ready!(state.step(
+                        cx,
+                        body,
+                        chunk_len,
+                        extensions_cnt,
+                        &mut buf,
+                        trailers_buf,
+                        trailers_cnt,
+                        h1_max_headers,
+                        h1_max_header_size
+                    ))?;
                     if *state == ChunkedState::End {
                         trace!("end of chunked");
-                        return Poll::Ready(Ok(Bytes::new()));
+
+                        if trailers_buf.is_some() {
+                            trace!("found possible trailers");
+
+                            // decoder enforces that trailers count will not exceed h1_max_headers
+                            if *trailers_cnt >= h1_max_headers {
+                                return Poll::Ready(Err(io::Error::new(
+                                    io::ErrorKind::InvalidData,
+                                    "chunk trailers count overflow",
+                                )));
+                            }
+                            match decode_trailers(
+                                &mut trailers_buf.take().expect("Trailer is None"),
+                                *trailers_cnt,
+                            ) {
+                                Ok(headers) => {
+                                    return Poll::Ready(Ok(Frame::trailers(headers)));
+                                }
+                                Err(e) => {
+                                    return Poll::Ready(Err(e));
+                                }
+                            }
+                        }
+
+                        return Poll::Ready(Ok(Frame::data(Bytes::new())));
                     }
                     if let Some(buf) = buf {
-                        return Poll::Ready(Ok(buf));
+                        return Poll::Ready(Ok(Frame::data(buf)));
                     }
                 }
             }
             Eof(ref mut is_eof) => {
                 if *is_eof {
-                    Poll::Ready(Ok(Bytes::new()))
+                    Poll::Ready(Ok(Frame::data(Bytes::new())))
                 } else {
                     // 8192 chosen because its about 2 packets, there probably
                     // won't be that much available, so don't have MemReaders
                     // allocate buffers to big
                     body.read_mem(cx, 8192).map_ok(|slice| {
                         *is_eof = slice.is_empty();
-                        slice
+                        Frame::data(slice)
                     })
                 }
             }
@@ -181,7 +243,7 @@ impl Decoder {
     }
 
     #[cfg(test)]
-    async fn decode_fut<R: MemRead>(&mut self, body: &mut R) -> Result<Bytes, io::Error> {
+    async fn decode_fut<R: MemRead>(&mut self, body: &mut R) -> Result<Frame<Bytes>, io::Error> {
         futures_util::future::poll_fn(move |cx| self.decode(cx, body)).await
     }
 }
@@ -216,6 +278,19 @@ macro_rules! or_overflow {
     )
 }
 
+macro_rules! put_u8 {
+    ($trailers_buf:expr, $byte:expr, $limit:expr) => {
+        $trailers_buf.put_u8($byte);
+
+        if $trailers_buf.len() >= $limit {
+            return Poll::Ready(Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "chunk trailers bytes over limit",
+            )));
+        }
+    };
+}
+
 impl ChunkedState {
     fn new() -> ChunkedState {
         ChunkedState::Start
@@ -227,6 +302,10 @@ impl ChunkedState {
         size: &mut u64,
         extensions_cnt: &mut u64,
         buf: &mut Option<Bytes>,
+        trailers_buf: &mut Option<BytesMut>,
+        trailers_cnt: &mut usize,
+        h1_max_headers: usize,
+        h1_max_header_size: usize,
     ) -> Poll<Result<ChunkedState, io::Error>> {
         use self::ChunkedState::*;
         match *self {
@@ -238,10 +317,17 @@ impl ChunkedState {
             Body => ChunkedState::read_body(cx, body, size, buf),
             BodyCr => ChunkedState::read_body_cr(cx, body),
             BodyLf => ChunkedState::read_body_lf(cx, body),
-            Trailer => ChunkedState::read_trailer(cx, body),
-            TrailerLf => ChunkedState::read_trailer_lf(cx, body),
-            EndCr => ChunkedState::read_end_cr(cx, body),
-            EndLf => ChunkedState::read_end_lf(cx, body),
+            Trailer => ChunkedState::read_trailer(cx, body, trailers_buf, h1_max_header_size),
+            TrailerLf => ChunkedState::read_trailer_lf(
+                cx,
+                body,
+                trailers_buf,
+                trailers_cnt,
+                h1_max_headers,
+                h1_max_header_size,
+            ),
+            EndCr => ChunkedState::read_end_cr(cx, body, trailers_buf, h1_max_header_size),
+            EndLf => ChunkedState::read_end_lf(cx, body, trailers_buf, h1_max_header_size),
             End => Poll::Ready(Ok(ChunkedState::End)),
         }
     }
@@ -442,19 +528,51 @@ impl ChunkedState {
     fn read_trailer<R: MemRead>(
         cx: &mut Context<'_>,
         rdr: &mut R,
+        trailers_buf: &mut Option<BytesMut>,
+        h1_max_header_size: usize,
     ) -> Poll<Result<ChunkedState, io::Error>> {
         trace!("read_trailer");
-        match byte!(rdr, cx) {
+        let byte = byte!(rdr, cx);
+
+        put_u8!(
+            trailers_buf.as_mut().expect("trailers_buf is None"),
+            byte,
+            h1_max_header_size
+        );
+
+        match byte {
             b'\r' => Poll::Ready(Ok(ChunkedState::TrailerLf)),
             _ => Poll::Ready(Ok(ChunkedState::Trailer)),
         }
     }
+
     fn read_trailer_lf<R: MemRead>(
         cx: &mut Context<'_>,
         rdr: &mut R,
+        trailers_buf: &mut Option<BytesMut>,
+        trailers_cnt: &mut usize,
+        h1_max_headers: usize,
+        h1_max_header_size: usize,
     ) -> Poll<Result<ChunkedState, io::Error>> {
-        match byte!(rdr, cx) {
-            b'\n' => Poll::Ready(Ok(ChunkedState::EndCr)),
+        let byte = byte!(rdr, cx);
+        match byte {
+            b'\n' => {
+                if *trailers_cnt >= h1_max_headers {
+                    return Poll::Ready(Err(io::Error::new(
+                        io::ErrorKind::InvalidData,
+                        "chunk trailers count overflow",
+                    )));
+                }
+                *trailers_cnt += 1;
+
+                put_u8!(
+                    trailers_buf.as_mut().expect("trailers_buf is None"),
+                    byte,
+                    h1_max_header_size
+                );
+
+                Poll::Ready(Ok(ChunkedState::EndCr))
+            }
             _ => Poll::Ready(Err(io::Error::new(
                 io::ErrorKind::InvalidInput,
                 "Invalid trailer end LF",
@@ -465,18 +583,48 @@ impl ChunkedState {
     fn read_end_cr<R: MemRead>(
         cx: &mut Context<'_>,
         rdr: &mut R,
+        trailers_buf: &mut Option<BytesMut>,
+        h1_max_header_size: usize,
     ) -> Poll<Result<ChunkedState, io::Error>> {
-        match byte!(rdr, cx) {
-            b'\r' => Poll::Ready(Ok(ChunkedState::EndLf)),
-            _ => Poll::Ready(Ok(ChunkedState::Trailer)),
+        let byte = byte!(rdr, cx);
+        match byte {
+            b'\r' => {
+                if let Some(trailers_buf) = trailers_buf {
+                    put_u8!(trailers_buf, byte, h1_max_header_size);
+                }
+                Poll::Ready(Ok(ChunkedState::EndLf))
+            }
+            byte => {
+                match trailers_buf {
+                    None => {
+                        // 64 will fit a single Expires header without reallocating
+                        let mut buf = BytesMut::with_capacity(64);
+                        buf.put_u8(byte);
+                        *trailers_buf = Some(buf);
+                    }
+                    Some(ref mut trailers_buf) => {
+                        put_u8!(trailers_buf, byte, h1_max_header_size);
+                    }
+                }
+
+                Poll::Ready(Ok(ChunkedState::Trailer))
+            }
         }
     }
     fn read_end_lf<R: MemRead>(
         cx: &mut Context<'_>,
         rdr: &mut R,
+        trailers_buf: &mut Option<BytesMut>,
+        h1_max_header_size: usize,
     ) -> Poll<Result<ChunkedState, io::Error>> {
-        match byte!(rdr, cx) {
-            b'\n' => Poll::Ready(Ok(ChunkedState::End)),
+        let byte = byte!(rdr, cx);
+        match byte {
+            b'\n' => {
+                if let Some(trailers_buf) = trailers_buf {
+                    put_u8!(trailers_buf, byte, h1_max_header_size);
+                }
+                Poll::Ready(Ok(ChunkedState::End))
+            }
             _ => Poll::Ready(Err(io::Error::new(
                 io::ErrorKind::InvalidInput,
                 "Invalid chunk end LF",
@@ -485,6 +633,48 @@ impl ChunkedState {
     }
 }
 
+// TODO: disallow Transfer-Encoding, Content-Length, Trailer, etc in trailers ??
+fn decode_trailers(buf: &mut BytesMut, count: usize) -> Result<HeaderMap, io::Error> {
+    let mut trailers = HeaderMap::new();
+    let mut headers = vec![httparse::EMPTY_HEADER; count];
+    let res = httparse::parse_headers(buf, &mut headers);
+    match res {
+        Ok(httparse::Status::Complete((_, headers))) => {
+            for header in headers.iter() {
+                use std::convert::TryFrom;
+                let name = match HeaderName::try_from(header.name) {
+                    Ok(name) => name,
+                    Err(_) => {
+                        return Err(io::Error::new(
+                            io::ErrorKind::InvalidInput,
+                            format!("Invalid header name: {:?}", &header),
+                        ));
+                    }
+                };
+
+                let value = match HeaderValue::from_bytes(header.value) {
+                    Ok(value) => value,
+                    Err(_) => {
+                        return Err(io::Error::new(
+                            io::ErrorKind::InvalidInput,
+                            format!("Invalid header value: {:?}", &header),
+                        ));
+                    }
+                };
+
+                trailers.insert(name, value);
+            }
+
+            Ok(trailers)
+        }
+        Ok(httparse::Status::Partial) => Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "Partial header",
+        )),
+        Err(e) => Err(io::Error::new(io::ErrorKind::InvalidInput, e)),
+    }
+}
+
 #[derive(Debug)]
 struct IncompleteBody;
 
@@ -554,9 +744,20 @@ mod tests {
             let rdr = &mut s.as_bytes();
             let mut size = 0;
             let mut ext_cnt = 0;
+            let mut trailers_cnt = 0;
             loop {
                 let result = futures_util::future::poll_fn(|cx| {
-                    state.step(cx, rdr, &mut size, &mut ext_cnt, &mut None)
+                    state.step(
+                        cx,
+                        rdr,
+                        &mut size,
+                        &mut ext_cnt,
+                        &mut None,
+                        &mut None,
+                        &mut trailers_cnt,
+                        DEFAULT_MAX_HEADERS,
+                        TRAILER_LIMIT,
+                    )
                 })
                 .await;
                 let desc = format!("read_size failed for {:?}", s);
@@ -573,9 +774,20 @@ mod tests {
             let rdr = &mut s.as_bytes();
             let mut size = 0;
             let mut ext_cnt = 0;
+            let mut trailers_cnt = 0;
             loop {
                 let result = futures_util::future::poll_fn(|cx| {
-                    state.step(cx, rdr, &mut size, &mut ext_cnt, &mut None)
+                    state.step(
+                        cx,
+                        rdr,
+                        &mut size,
+                        &mut ext_cnt,
+                        &mut None,
+                        &mut None,
+                        &mut trailers_cnt,
+                        DEFAULT_MAX_HEADERS,
+                        TRAILER_LIMIT,
+                    )
                 })
                 .await;
                 state = match result {
@@ -639,7 +851,16 @@ mod tests {
     async fn test_read_sized_early_eof() {
         let mut bytes = &b"foo bar"[..];
         let mut decoder = Decoder::length(10);
-        assert_eq!(decoder.decode_fut(&mut bytes).await.unwrap().len(), 7);
+        assert_eq!(
+            decoder
+                .decode_fut(&mut bytes)
+                .await
+                .unwrap()
+                .data_ref()
+                .unwrap()
+                .len(),
+            7
+        );
         let e = decoder.decode_fut(&mut bytes).await.unwrap_err();
         assert_eq!(e.kind(), io::ErrorKind::UnexpectedEof);
     }
@@ -651,8 +872,17 @@ mod tests {
             9\r\n\
             foo bar\
         "[..];
-        let mut decoder = Decoder::chunked();
-        assert_eq!(decoder.decode_fut(&mut bytes).await.unwrap().len(), 7);
+        let mut decoder = Decoder::chunked(None, None);
+        assert_eq!(
+            decoder
+                .decode_fut(&mut bytes)
+                .await
+                .unwrap()
+                .data_ref()
+                .unwrap()
+                .len(),
+            7
+        );
         let e = decoder.decode_fut(&mut bytes).await.unwrap_err();
         assert_eq!(e.kind(), io::ErrorKind::UnexpectedEof);
     }
@@ -661,10 +891,12 @@ mod tests {
     #[tokio::test]
     async fn test_read_chunked_single_read() {
         let mut mock_buf = &b"10\r\n1234567890abcdef\r\n0\r\n"[..];
-        let buf = Decoder::chunked()
+        let buf = Decoder::chunked(None, None)
             .decode_fut(&mut mock_buf)
             .await
-            .expect("decode");
+            .expect("decode")
+            .into_data()
+            .expect("unknown frame type");
         assert_eq!(16, buf.len());
         let result = String::from_utf8(buf.as_ref().to_vec()).expect("decode String");
         assert_eq!("1234567890abcdef", &result);
@@ -684,8 +916,13 @@ mod tests {
         scratch.extend(b"0\r\n\r\n");
         let mut mock_buf = Bytes::from(scratch);
 
-        let mut decoder = Decoder::chunked();
-        let buf1 = decoder.decode_fut(&mut mock_buf).await.expect("decode1");
+        let mut decoder = Decoder::chunked(None, None);
+        let buf1 = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .expect("decode1")
+            .into_data()
+            .expect("unknown frame type");
         assert_eq!(&buf1[..], b"A");
 
         let err = decoder
@@ -700,7 +937,7 @@ mod tests {
     #[tokio::test]
     async fn test_read_chunked_trailer_with_missing_lf() {
         let mut mock_buf = &b"10\r\n1234567890abcdef\r\n0\r\nbad\r\r\n"[..];
-        let mut decoder = Decoder::chunked();
+        let mut decoder = Decoder::chunked(None, None);
         decoder.decode_fut(&mut mock_buf).await.expect("decode");
         let e = decoder.decode_fut(&mut mock_buf).await.unwrap_err();
         assert_eq!(e.kind(), io::ErrorKind::InvalidInput);
@@ -710,20 +947,35 @@ mod tests {
     #[tokio::test]
     async fn test_read_chunked_after_eof() {
         let mut mock_buf = &b"10\r\n1234567890abcdef\r\n0\r\n\r\n"[..];
-        let mut decoder = Decoder::chunked();
+        let mut decoder = Decoder::chunked(None, None);
 
         // normal read
-        let buf = decoder.decode_fut(&mut mock_buf).await.unwrap();
+        let buf = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .unwrap()
+            .into_data()
+            .expect("unknown frame type");
         assert_eq!(16, buf.len());
         let result = String::from_utf8(buf.as_ref().to_vec()).expect("decode String");
         assert_eq!("1234567890abcdef", &result);
 
         // eof read
-        let buf = decoder.decode_fut(&mut mock_buf).await.expect("decode");
+        let buf = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .expect("decode")
+            .into_data()
+            .expect("unknown frame type");
         assert_eq!(0, buf.len());
 
         // ensure read after eof also returns eof
-        let buf = decoder.decode_fut(&mut mock_buf).await.expect("decode");
+        let buf = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .expect("decode")
+            .into_data()
+            .expect("unknown frame type");
         assert_eq!(0, buf.len());
     }
 
@@ -751,7 +1003,9 @@ mod tests {
             let buf = decoder
                 .decode_fut(&mut ins)
                 .await
-                .expect("unexpected decode error");
+                .expect("unexpected decode error")
+                .into_data()
+                .expect("unexpected frame type");
             if buf.is_empty() {
                 break; // eof
             }
@@ -783,7 +1037,7 @@ mod tests {
     async fn test_read_chunked_async() {
         let content = "3\r\nfoo\r\n3\r\nbar\r\n0\r\n\r\n";
         let expected = "foobar";
-        all_async_cases(content, expected, Decoder::chunked()).await;
+        all_async_cases(content, expected, Decoder::chunked(None, None)).await;
     }
 
     #[cfg(not(miri))]
@@ -808,10 +1062,15 @@ mod tests {
         b.bytes = LEN as u64;
 
         b.iter(|| {
-            let mut decoder = Decoder::chunked();
+            let mut decoder = Decoder::chunked(None, None);
             rt.block_on(async {
                 let mut raw = content.clone();
-                let chunk = decoder.decode_fut(&mut raw).await.unwrap();
+                let chunk = decoder
+                    .decode_fut(&mut raw)
+                    .await
+                    .unwrap()
+                    .into_data()
+                    .unwrap();
                 assert_eq!(chunk.len(), LEN);
             });
         });
@@ -830,7 +1089,12 @@ mod tests {
             let mut decoder = Decoder::length(LEN as u64);
             rt.block_on(async {
                 let mut raw = content.clone();
-                let chunk = decoder.decode_fut(&mut raw).await.unwrap();
+                let chunk = decoder
+                    .decode_fut(&mut raw)
+                    .await
+                    .unwrap()
+                    .into_data()
+                    .unwrap();
                 assert_eq!(chunk.len(), LEN);
             });
         });
@@ -843,4 +1107,110 @@ mod tests {
             .build()
             .expect("rt build")
     }
+
+    #[test]
+    fn test_decode_trailers() {
+        let mut buf = BytesMut::new();
+        buf.extend_from_slice(
+            b"Expires: Wed, 21 Oct 2015 07:28:00 GMT\r\nX-Stream-Error: failed to decode\r\n\r\n",
+        );
+        let headers = decode_trailers(&mut buf, 2).expect("decode_trailers");
+        assert_eq!(headers.len(), 2);
+        assert_eq!(
+            headers.get("Expires").unwrap(),
+            "Wed, 21 Oct 2015 07:28:00 GMT"
+        );
+        assert_eq!(headers.get("X-Stream-Error").unwrap(), "failed to decode");
+    }
+
+    #[tokio::test]
+    async fn test_trailer_max_headers_enforced() {
+        let h1_max_headers = 10;
+        let mut scratch = vec![];
+        scratch.extend(b"10\r\n1234567890abcdef\r\n0\r\n");
+        for i in 0..h1_max_headers {
+            scratch.extend(format!("trailer{}: {}\r\n", i, i).as_bytes());
+        }
+        scratch.extend(b"\r\n");
+        let mut mock_buf = Bytes::from(scratch);
+
+        let mut decoder = Decoder::chunked(Some(h1_max_headers), None);
+
+        // ready chunked body
+        let buf = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .unwrap()
+            .into_data()
+            .expect("unknown frame type");
+        assert_eq!(16, buf.len());
+
+        // eof read
+        let err = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .expect_err("trailer fields over limit");
+        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+    }
+
+    #[tokio::test]
+    async fn test_trailer_max_header_size_huge_trailer() {
+        let max_header_size = 1024;
+        let mut scratch = vec![];
+        scratch.extend(b"10\r\n1234567890abcdef\r\n0\r\n");
+        scratch.extend(format!("huge_trailer: {}\r\n", "x".repeat(max_header_size)).as_bytes());
+        scratch.extend(b"\r\n");
+        let mut mock_buf = Bytes::from(scratch);
+
+        let mut decoder = Decoder::chunked(None, Some(max_header_size));
+
+        // ready chunked body
+        let buf = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .unwrap()
+            .into_data()
+            .expect("unknown frame type");
+        assert_eq!(16, buf.len());
+
+        // eof read
+        let err = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .expect_err("trailers over limit");
+        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+    }
+
+    #[tokio::test]
+    async fn test_trailer_max_header_size_many_small_trailers() {
+        let max_headers = 10;
+        let header_size = 64;
+        let mut scratch = vec![];
+        scratch.extend(b"10\r\n1234567890abcdef\r\n0\r\n");
+
+        for i in 0..max_headers {
+            scratch.extend(format!("trailer{}: {}\r\n", i, "x".repeat(header_size)).as_bytes());
+        }
+
+        scratch.extend(b"\r\n");
+        let mut mock_buf = Bytes::from(scratch);
+
+        let mut decoder = Decoder::chunked(None, Some(max_headers * header_size));
+
+        // ready chunked body
+        let buf = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .unwrap()
+            .into_data()
+            .expect("unknown frame type");
+        assert_eq!(16, buf.len());
+
+        // eof read
+        let err = decoder
+            .decode_fut(&mut mock_buf)
+            .await
+            .expect_err("trailers over limit");
+        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+    }
 }
diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs
index 3a4faf04..be0a88c1 100644
--- a/src/proto/h1/dispatch.rs
+++ b/src/proto/h1/dispatch.rs
@@ -213,17 +213,39 @@ where
                         }
                     }
                     match self.conn.poll_read_body(cx) {
-                        Poll::Ready(Some(Ok(chunk))) => match body.try_send_data(chunk) {
-                            Ok(()) => {
-                                self.body_tx = Some(body);
-                            }
-                            Err(_canceled) => {
-                                if self.conn.can_read_body() {
-                                    trace!("body receiver dropped before eof, closing");
-                                    self.conn.close_read();
+                        Poll::Ready(Some(Ok(frame))) => {
+                            if frame.is_data() {
+                                let chunk = frame.into_data().unwrap_or_else(|_| unreachable!());
+                                match body.try_send_data(chunk) {
+                                    Ok(()) => {
+                                        self.body_tx = Some(body);
+                                    }
+                                    Err(_canceled) => {
+                                        if self.conn.can_read_body() {
+                                            trace!("body receiver dropped before eof, closing");
+                                            self.conn.close_read();
+                                        }
+                                    }
+                                }
+                            } else if frame.is_trailers() {
+                                let trailers =
+                                    frame.into_trailers().unwrap_or_else(|_| unreachable!());
+                                match body.try_send_trailers(trailers) {
+                                    Ok(()) => {
+                                        self.body_tx = Some(body);
+                                    }
+                                    Err(_canceled) => {
+                                        if self.conn.can_read_body() {
+                                            trace!("body receiver dropped before eof, closing");
+                                            self.conn.close_read();
+                                        }
+                                    }
                                 }
+                            } else {
+                                // we should have dropped all unknown frames in poll_read_body
+                                error!("unexpected frame");
                             }
-                        },
+                        }
                         Poll::Ready(None) => {
                             // just drop, the body will close automatically
                         }
diff --git a/src/proto/h1/encode.rs b/src/proto/h1/encode.rs
index fe582901..2f24a8a3 100644
--- a/src/proto/h1/encode.rs
+++ b/src/proto/h1/encode.rs
@@ -165,9 +165,10 @@ impl Encoder {
         trailers: HeaderMap,
         title_case_headers: bool,
     ) -> Option<EncodedBuf<B>> {
+        trace!("encoding trailers");
         match &self.kind {
-            Kind::Chunked(Some(ref allowed_trailer_fields)) => {
-                let allowed_trailer_field_map = allowed_trailer_field_map(&allowed_trailer_fields);
+            Kind::Chunked(Some(allowed_trailer_fields)) => {
+                let allowed_trailer_field_map = allowed_trailer_field_map(allowed_trailer_fields);
 
                 let mut cur_name = None;
                 let mut allowed_trailers = HeaderMap::new();
@@ -178,10 +179,14 @@ impl Encoder {
                     }
                     let name = cur_name.as_ref().expect("current header name");
 
-                    if allowed_trailer_field_map.contains_key(name.as_str())
-                        && is_valid_trailer_field(name)
-                    {
-                        allowed_trailers.insert(name, value);
+                    if allowed_trailer_field_map.contains_key(name.as_str()) {
+                        if is_valid_trailer_field(name) {
+                            allowed_trailers.insert(name, value);
+                        } else {
+                            debug!("trailer field is not valid: {}", &name);
+                        }
+                    } else {
+                        debug!("trailer header name not found in trailer header: {}", &name);
                     }
                 }
 
@@ -200,6 +205,10 @@ impl Encoder {
                     kind: BufKind::Trailers(b"0\r\n".chain(Bytes::from(buf)).chain(b"\r\n")),
                 })
             }
+            Kind::Chunked(None) => {
+                debug!("attempted to encode trailers, but the trailer header is not set");
+                None
+            }
             _ => {
                 debug!("attempted to encode trailers for non-chunked response");
                 None
diff --git a/src/proto/h1/io.rs b/src/proto/h1/io.rs
index ae9be995..34eb477f 100644
--- a/src/proto/h1/io.rs
+++ b/src/proto/h1/io.rs
@@ -3,7 +3,6 @@ use std::fmt;
 #[cfg(feature = "server")]
 use std::future::Future;
 use std::io::{self, IoSlice};
-use std::mem::MaybeUninit;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -246,8 +245,9 @@ where
             self.read_buf.reserve(next);
         }
 
-        let dst = self.read_buf.chunk_mut();
-        let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
+        // SAFETY: ReadBuf and poll_read promise not to set any uninitialized
+        // bytes onto `dst`.
+        let dst = unsafe { self.read_buf.chunk_mut().as_uninit_slice_mut() };
         let mut buf = ReadBuf::uninit(dst);
         match Pin::new(&mut self.io).poll_read(cx, buf.unfilled()) {
             Poll::Ready(Ok(_)) => {
@@ -450,7 +450,7 @@ fn prev_power_of_two(n: usize) -> usize {
     // Only way this shift can underflow is if n is less than 4.
     // (Which would means `usize::MAX >> 64` and underflowed!)
     debug_assert!(n >= 4);
-    (::std::usize::MAX >> (n.leading_zeros() + 2)) + 1
+    (usize::MAX >> (n.leading_zeros() + 2)) + 1
 }
 
 impl Default for ReadStrategy {
@@ -763,7 +763,7 @@ mod tests {
         assert_eq!(strategy.next(), 32768);
 
         // Enormous records still increment at same rate
-        strategy.record(::std::usize::MAX);
+        strategy.record(usize::MAX);
         assert_eq!(strategy.next(), 65536);
 
         let max = strategy.max();
@@ -833,7 +833,7 @@ mod tests {
         fn fuzz(max: usize) {
             let mut strategy = ReadStrategy::with_max(max);
             while strategy.next() < max {
-                strategy.record(::std::usize::MAX);
+                strategy.record(usize::MAX);
             }
             let mut next = strategy.next();
             while next > 8192 {
@@ -854,7 +854,7 @@ mod tests {
             fuzz(max);
             max = (max / 2).saturating_mul(3);
         }
-        fuzz(::std::usize::MAX);
+        fuzz(usize::MAX);
     }
 
     #[test]
diff --git a/src/proto/h1/mod.rs b/src/proto/h1/mod.rs
index 5b9872cd..074d0b88 100644
--- a/src/proto/h1/mod.rs
+++ b/src/proto/h1/mod.rs
@@ -103,6 +103,8 @@ pub(crate) struct Encode<'a, T> {
     keep_alive: bool,
     req_method: &'a mut Option<Method>,
     title_case_headers: bool,
+    #[cfg(feature = "server")]
+    date_header: bool,
 }
 
 /// Extra flags that a request "wants", like expect-continue or upgrades.
diff --git a/src/proto/h1/role.rs b/src/proto/h1/role.rs
index 95834077..847453a0 100644
--- a/src/proto/h1/role.rs
+++ b/src/proto/h1/role.rs
@@ -1,7 +1,7 @@
-use std::{fmt, mem::MaybeUninit};
+use std::mem::MaybeUninit;
 
 #[cfg(feature = "client")]
-use std::fmt::Write;
+use std::fmt::{self, Write as _};
 #[cfg(feature = "server")]
 use std::time::Instant;
 
@@ -30,7 +30,7 @@ use crate::proto::h1::{
 use crate::proto::RequestHead;
 use crate::proto::{BodyLength, MessageHead, RequestLine};
 
-const DEFAULT_MAX_HEADERS: usize = 100;
+pub(crate) const DEFAULT_MAX_HEADERS: usize = 100;
 const AVERAGE_HEADER_SIZE: usize = 30; // totally scientific
 #[cfg(feature = "server")]
 const MAX_URI_LEN: usize = (u16::MAX - 1) as usize;
@@ -934,7 +934,8 @@ impl Server {
         }
 
         // cached date is much faster than formatting every request
-        if !wrote_date {
+        // don't force the write if disabled
+        if !wrote_date && msg.date_header {
             dst.reserve(date::DATE_VALUE_LENGTH + 8);
             header_name_writer.write_header_name_with_colon(dst, "date: ", header::DATE);
             date::extend(dst);
@@ -1604,8 +1605,10 @@ fn write_headers_original_case(
     }
 }
 
+#[cfg(feature = "client")]
 struct FastWrite<'a>(&'a mut Vec<u8>);
 
+#[cfg(feature = "client")]
 impl<'a> fmt::Write for FastWrite<'a> {
     #[inline]
     fn write_str(&mut self, s: &str) -> fmt::Result {
@@ -2503,6 +2506,7 @@ mod tests {
                 keep_alive: true,
                 req_method: &mut None,
                 title_case_headers: true,
+                date_header: true,
             },
             &mut vec,
         )
@@ -2534,6 +2538,7 @@ mod tests {
                 keep_alive: true,
                 req_method: &mut None,
                 title_case_headers: false,
+                date_header: true,
             },
             &mut vec,
         )
@@ -2568,6 +2573,7 @@ mod tests {
                 keep_alive: true,
                 req_method: &mut None,
                 title_case_headers: true,
+                date_header: true,
             },
             &mut vec,
         )
@@ -2592,6 +2598,7 @@ mod tests {
                 keep_alive: true,
                 req_method: &mut Some(Method::CONNECT),
                 title_case_headers: false,
+                date_header: true,
             },
             &mut vec,
         )
@@ -2621,6 +2628,7 @@ mod tests {
                 keep_alive: true,
                 req_method: &mut None,
                 title_case_headers: true,
+                date_header: true,
             },
             &mut vec,
         )
@@ -2655,6 +2663,7 @@ mod tests {
                 keep_alive: true,
                 req_method: &mut None,
                 title_case_headers: false,
+                date_header: true,
             },
             &mut vec,
         )
@@ -2689,17 +2698,54 @@ mod tests {
                 keep_alive: true,
                 req_method: &mut None,
                 title_case_headers: true,
+                date_header: true,
             },
             &mut vec,
         )
         .unwrap();
 
+        // this will also test that the date does exist
         let expected_response =
             b"HTTP/1.1 200 OK\r\nCONTENT-LENGTH: 10\r\nContent-Type: application/json\r\nDate: ";
 
         assert_eq!(&vec[..expected_response.len()], &expected_response[..]);
     }
 
+    #[test]
+    fn test_disabled_date_header() {
+        use crate::proto::BodyLength;
+        use http::header::{HeaderValue, CONTENT_LENGTH};
+
+        let mut head = MessageHead::default();
+        head.headers
+            .insert("content-length", HeaderValue::from_static("10"));
+        head.headers
+            .insert("content-type", HeaderValue::from_static("application/json"));
+
+        let mut orig_headers = HeaderCaseMap::default();
+        orig_headers.insert(CONTENT_LENGTH, "CONTENT-LENGTH".into());
+        head.extensions.insert(orig_headers);
+
+        let mut vec = Vec::new();
+        Server::encode(
+            Encode {
+                head: &mut head,
+                body: Some(BodyLength::Known(10)),
+                keep_alive: true,
+                req_method: &mut None,
+                title_case_headers: true,
+                date_header: false,
+            },
+            &mut vec,
+        )
+        .unwrap();
+
+        let expected_response =
+            b"HTTP/1.1 200 OK\r\nCONTENT-LENGTH: 10\r\nContent-Type: application/json\r\n\r\n";
+
+        assert_eq!(&vec, &expected_response);
+    }
+
     #[test]
     fn parse_header_htabs() {
         let mut bytes = BytesMut::from("HTTP/1.1 200 OK\r\nserver: hello\tworld\r\n\r\n");
@@ -3037,6 +3083,7 @@ mod tests {
                     keep_alive: true,
                     req_method: &mut Some(Method::GET),
                     title_case_headers: false,
+                    date_header: true,
                 },
                 &mut vec,
             )
@@ -3065,6 +3112,7 @@ mod tests {
                     keep_alive: true,
                     req_method: &mut Some(Method::GET),
                     title_case_headers: false,
+                    date_header: true,
                 },
                 &mut vec,
             )
diff --git a/src/proto/h2/client.rs b/src/proto/h2/client.rs
index 7bc62de5..7cb6c6ed 100644
--- a/src/proto/h2/client.rs
+++ b/src/proto/h2/client.rs
@@ -51,6 +51,7 @@ const DEFAULT_CONN_WINDOW: u32 = 1024 * 1024 * 5; // 5mb
 const DEFAULT_STREAM_WINDOW: u32 = 1024 * 1024 * 2; // 2mb
 const DEFAULT_MAX_FRAME_SIZE: u32 = 1024 * 16; // 16kb
 const DEFAULT_MAX_SEND_BUF_SIZE: usize = 1024 * 1024; // 1mb
+const DEFAULT_MAX_HEADER_LIST_SIZE: u32 = 1024 * 16; // 16kb
 
 // The maximum number of concurrent streams that the client is allowed to open
 // before it receives the initial SETTINGS frame from the server.
@@ -68,11 +69,13 @@ pub(crate) struct Config {
     pub(crate) initial_stream_window_size: u32,
     pub(crate) initial_max_send_streams: usize,
     pub(crate) max_frame_size: u32,
+    pub(crate) max_header_list_size: u32,
     pub(crate) keep_alive_interval: Option<Duration>,
     pub(crate) keep_alive_timeout: Duration,
     pub(crate) keep_alive_while_idle: bool,
     pub(crate) max_concurrent_reset_streams: Option<usize>,
     pub(crate) max_send_buffer_size: usize,
+    pub(crate) max_pending_accept_reset_streams: Option<usize>,
 }
 
 impl Default for Config {
@@ -83,11 +86,13 @@ impl Default for Config {
             initial_stream_window_size: DEFAULT_STREAM_WINDOW,
             initial_max_send_streams: DEFAULT_INITIAL_MAX_SEND_STREAMS,
             max_frame_size: DEFAULT_MAX_FRAME_SIZE,
+            max_header_list_size: DEFAULT_MAX_HEADER_LIST_SIZE,
             keep_alive_interval: None,
             keep_alive_timeout: Duration::from_secs(20),
             keep_alive_while_idle: false,
             max_concurrent_reset_streams: None,
             max_send_buffer_size: DEFAULT_MAX_SEND_BUF_SIZE,
+            max_pending_accept_reset_streams: None,
         }
     }
 }
@@ -99,11 +104,15 @@ fn new_builder(config: &Config) -> Builder {
         .initial_window_size(config.initial_stream_window_size)
         .initial_connection_window_size(config.initial_conn_window_size)
         .max_frame_size(config.max_frame_size)
+        .max_header_list_size(config.max_header_list_size)
         .max_send_buffer_size(config.max_send_buffer_size)
         .enable_push(false);
     if let Some(max) = config.max_concurrent_reset_streams {
         builder.max_concurrent_reset_streams(max);
     }
+    if let Some(max) = config.max_pending_accept_reset_streams {
+        builder.max_pending_accept_reset_streams(max);
+    }
     builder
 }
 
diff --git a/src/proto/h2/mod.rs b/src/proto/h2/mod.rs
index d880502b..adb6de87 100644
--- a/src/proto/h2/mod.rs
+++ b/src/proto/h2/mod.rs
@@ -8,7 +8,7 @@ use std::task::{Context, Poll};
 use bytes::{Buf, Bytes};
 use futures_util::ready;
 use h2::{Reason, RecvStream, SendStream};
-use http::header::{HeaderName, CONNECTION, TE, TRAILER, TRANSFER_ENCODING, UPGRADE};
+use http::header::{HeaderName, CONNECTION, TE, TRANSFER_ENCODING, UPGRADE};
 use http::HeaderMap;
 use pin_project_lite::pin_project;
 
@@ -31,15 +31,13 @@ cfg_server! {
 /// Default initial stream window size defined in HTTP2 spec.
 pub(crate) const SPEC_WINDOW_SIZE: u32 = 65_535;
 
-// List of connection headers from:
-// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Connection
+// List of connection headers from RFC 9110 Section 7.6.1
 //
 // TE headers are allowed in HTTP/2 requests as long as the value is "trailers", so they're
 // tested separately.
-static CONNECTION_HEADERS: [HeaderName; 5] = [
+static CONNECTION_HEADERS: [HeaderName; 4] = [
     HeaderName::from_static("keep-alive"),
     HeaderName::from_static("proxy-connection"),
-    TRAILER,
     TRANSFER_ENCODING,
     UPGRADE,
 ];
@@ -54,8 +52,7 @@ fn strip_connection_headers(headers: &mut HeaderMap, is_request: bool) {
     if is_request {
         if headers
             .get(TE)
-            .map(|te_header| te_header != "trailers")
-            .unwrap_or(false)
+            .map_or(false, |te_header| te_header != "trailers")
         {
             warn!("TE headers not set to \"trailers\" are illegal in HTTP/2 requests");
             headers.remove(TE);
diff --git a/src/proto/h2/server.rs b/src/proto/h2/server.rs
index b56e3cfa..ee2d08ea 100644
--- a/src/proto/h2/server.rs
+++ b/src/proto/h2/server.rs
@@ -38,8 +38,7 @@ const DEFAULT_CONN_WINDOW: u32 = 1024 * 1024; // 1mb
 const DEFAULT_STREAM_WINDOW: u32 = 1024 * 1024; // 1mb
 const DEFAULT_MAX_FRAME_SIZE: u32 = 1024 * 16; // 16kb
 const DEFAULT_MAX_SEND_BUF_SIZE: usize = 1024 * 400; // 400kb
-                                                     // 16 MB "sane default" taken from golang http2
-const DEFAULT_SETTINGS_MAX_HEADER_LIST_SIZE: u32 = 16 << 20;
+const DEFAULT_SETTINGS_MAX_HEADER_LIST_SIZE: u32 = 1024 * 16; // 16kb
 const DEFAULT_MAX_LOCAL_ERROR_RESET_STREAMS: usize = 1024;
 
 #[derive(Clone, Debug)]
@@ -56,6 +55,7 @@ pub(crate) struct Config {
     pub(crate) keep_alive_timeout: Duration,
     pub(crate) max_send_buffer_size: usize,
     pub(crate) max_header_list_size: u32,
+    pub(crate) date_header: bool,
 }
 
 impl Default for Config {
@@ -73,6 +73,7 @@ impl Default for Config {
             keep_alive_timeout: Duration::from_secs(20),
             max_send_buffer_size: DEFAULT_MAX_SEND_BUF_SIZE,
             max_header_list_size: DEFAULT_SETTINGS_MAX_HEADER_LIST_SIZE,
+            date_header: true,
         }
     }
 }
@@ -87,6 +88,7 @@ pin_project! {
         timer: Time,
         service: S,
         state: State<T, B>,
+        date_header: bool,
     }
 }
 
@@ -109,6 +111,7 @@ where
     ping: Option<(ping::Recorder, ping::Ponger)>,
     conn: Connection<Compat<T>, SendBuf<B::Data>>,
     closing: Option<crate::Error>,
+    date_header: bool,
 }
 
 impl<T, S, B, E> Server<T, S, B, E>
@@ -168,6 +171,7 @@ where
                 hs: handshake,
             },
             service,
+            date_header: config.date_header,
         }
     }
 
@@ -220,6 +224,7 @@ where
                         ping,
                         conn,
                         closing: None,
+                        date_header: me.date_header,
                     })
                 }
                 State::Serving(ref mut srv) => {
@@ -303,7 +308,13 @@ where
                             req.extensions_mut().insert(Protocol::from_inner(protocol));
                         }
 
-                        let fut = H2Stream::new(service.call(req), connect_parts, respond);
+                        let fut = H2Stream::new(
+                            service.call(req),
+                            connect_parts,
+                            respond,
+                            self.date_header,
+                        );
+
                         exec.execute_h2stream(fut);
                     }
                     Some(Err(e)) => {
@@ -358,6 +369,7 @@ pin_project! {
         reply: SendResponse<SendBuf<B::Data>>,
         #[pin]
         state: H2StreamState<F, B>,
+        date_header: bool,
     }
 }
 
@@ -393,10 +405,12 @@ where
         fut: F,
         connect_parts: Option<ConnectParts>,
         respond: SendResponse<SendBuf<B::Data>>,
+        date_header: bool,
     ) -> H2Stream<F, B> {
         H2Stream {
             reply: respond,
             state: H2StreamState::Service { fut, connect_parts },
+            date_header,
         }
     }
 }
@@ -455,10 +469,12 @@ where
                     let mut res = ::http::Response::from_parts(head, ());
                     super::strip_connection_headers(res.headers_mut(), false);
 
-                    // set Date header if it isn't already set...
-                    res.headers_mut()
-                        .entry(::http::header::DATE)
-                        .or_insert_with(date::update_and_header_value);
+                    // set Date header if it isn't already set if instructed
+                    if *me.date_header {
+                        res.headers_mut()
+                            .entry(::http::header::DATE)
+                            .or_insert_with(date::update_and_header_value);
+                    }
 
                     if let Some(connect_parts) = connect_parts.take() {
                         if res.status().is_success() {
diff --git a/src/server/conn/http1.rs b/src/server/conn/http1.rs
index f9318c8d..cf223162 100644
--- a/src/server/conn/http1.rs
+++ b/src/server/conn/http1.rs
@@ -79,6 +79,7 @@ pub struct Builder {
     h1_writev: Option<bool>,
     max_buf_size: Option<usize>,
     pipeline_flush: bool,
+    date_header: bool,
 }
 
 /// Deconstructed parts of a `Connection`.
@@ -86,6 +87,7 @@ pub struct Builder {
 /// This allows taking apart a `Connection` at a later time, in order to
 /// reclaim the IO object, and additional related pieces.
 #[derive(Debug)]
+#[non_exhaustive]
 pub struct Parts<T, S> {
     /// The original IO object used in the handshake.
     pub io: T,
@@ -100,7 +102,6 @@ pub struct Parts<T, S> {
     pub read_buf: Bytes,
     /// The `Service` used to serve this connection.
     pub service: S,
-    _inner: (),
 }
 
 // ===== impl Connection =====
@@ -151,7 +152,6 @@ where
             io,
             read_buf,
             service: dispatch.into_service(),
-            _inner: (),
         }
     }
 
@@ -247,6 +247,7 @@ impl Builder {
             h1_writev: None,
             max_buf_size: None,
             pipeline_flush: false,
+            date_header: true,
         }
     }
     /// Set whether HTTP/1 connections should support half-closures.
@@ -360,6 +361,16 @@ impl Builder {
         self
     }
 
+    /// Set whether the `date` header should be included in HTTP responses.
+    ///
+    /// Note that including the `date` header is recommended by RFC 7231.
+    ///
+    /// Default is true.
+    pub fn auto_date_header(&mut self, enabled: bool) -> &mut Self {
+        self.date_header = enabled;
+        self
+    }
+
     /// Aggregates flushes to better support pipelined responses.
     ///
     /// Experimental, may have bugs.
@@ -483,7 +494,11 @@ where
     /// This `Connection` should continue to be polled until shutdown
     /// can finish.
     pub fn graceful_shutdown(mut self: Pin<&mut Self>) {
-        Pin::new(self.inner.as_mut().unwrap()).graceful_shutdown()
+        // Connection (`inner`) is `None` if it was upgraded (and `poll` is `Ready`).
+        // In that case, we don't need to call `graceful_shutdown`.
+        if let Some(conn) = self.inner.as_mut() {
+            Pin::new(conn).graceful_shutdown()
+        }
     }
 }
 
@@ -498,14 +513,19 @@ where
     type Output = crate::Result<()>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match ready!(Pin::new(&mut self.inner.as_mut().unwrap().conn).poll(cx)) {
-            Ok(proto::Dispatched::Shutdown) => Poll::Ready(Ok(())),
-            Ok(proto::Dispatched::Upgrade(pending)) => {
-                let (io, buf, _) = self.inner.take().unwrap().conn.into_inner();
-                pending.fulfill(Upgraded::new(io, buf));
-                Poll::Ready(Ok(()))
+        if let Some(conn) = self.inner.as_mut() {
+            match ready!(Pin::new(&mut conn.conn).poll(cx)) {
+                Ok(proto::Dispatched::Shutdown) => Poll::Ready(Ok(())),
+                Ok(proto::Dispatched::Upgrade(pending)) => {
+                    let (io, buf, _) = self.inner.take().unwrap().conn.into_inner();
+                    pending.fulfill(Upgraded::new(io, buf));
+                    Poll::Ready(Ok(()))
+                }
+                Err(e) => Poll::Ready(Err(e)),
             }
-            Err(e) => Poll::Ready(Err(e)),
+        } else {
+            // inner is `None`, meaning the connection was upgraded, thus it's `Poll::Ready(Ok(()))`
+            Poll::Ready(Ok(()))
         }
     }
 }
diff --git a/src/server/conn/http2.rs b/src/server/conn/http2.rs
index 6c8dfb5e..43e435bf 100644
--- a/src/server/conn/http2.rs
+++ b/src/server/conn/http2.rs
@@ -247,7 +247,7 @@ impl<E> Builder<E> {
     ///
     /// The value must be no larger than `u32::MAX`.
     pub fn max_send_buf_size(&mut self, max: usize) -> &mut Self {
-        assert!(max <= std::u32::MAX as usize);
+        assert!(max <= u32::MAX as usize);
         self.h2_builder.max_send_buffer_size = max;
         self
     }
@@ -262,7 +262,7 @@ impl<E> Builder<E> {
 
     /// Sets the max size of received header frames.
     ///
-    /// Default is currently ~16MB, but may change.
+    /// Default is currently 16KB, but can change.
     pub fn max_header_list_size(&mut self, max: u32) -> &mut Self {
         self.h2_builder.max_header_list_size = max;
         self
@@ -277,6 +277,16 @@ impl<E> Builder<E> {
         self
     }
 
+    /// Set whether the `date` header should be included in HTTP responses.
+    ///
+    /// Note that including the `date` header is recommended by RFC 7231.
+    ///
+    /// Default is true.
+    pub fn auto_date_header(&mut self, enabled: bool) -> &mut Self {
+        self.h2_builder.date_header = enabled;
+        self
+    }
+
     /// Bind a connection together with a [`Service`](crate::service::Service).
     ///
     /// This returns a Future that must be polled in order for HTTP to be
diff --git a/src/service/mod.rs b/src/service/mod.rs
index 36facc58..28ffaddb 100644
--- a/src/service/mod.rs
+++ b/src/service/mod.rs
@@ -25,12 +25,6 @@ mod http;
 mod service;
 mod util;
 
-#[cfg(all(any(feature = "http1", feature = "http2"), feature = "server"))]
 pub use self::http::HttpService;
-#[cfg(all(
-    any(feature = "http1", feature = "http2"),
-    any(feature = "server", feature = "client")
-))]
 pub use self::service::Service;
-
 pub use self::util::service_fn;
diff --git a/src/service/service.rs b/src/service/service.rs
index c6db129b..1b9aea51 100644
--- a/src/service/service.rs
+++ b/src/service/service.rs
@@ -38,3 +38,58 @@ pub trait Service<Request> {
     /// The discussion on this is here: <https://github.com/hyperium/hyper/issues/3040>
     fn call(&self, req: Request) -> Self::Future;
 }
+
+impl<Request, S: Service<Request> + ?Sized> Service<Request> for &'_ S {
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    #[inline]
+    fn call(&self, req: Request) -> Self::Future {
+        (**self).call(req)
+    }
+}
+
+impl<Request, S: Service<Request> + ?Sized> Service<Request> for &'_ mut S {
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    #[inline]
+    fn call(&self, req: Request) -> Self::Future {
+        (**self).call(req)
+    }
+}
+
+impl<Request, S: Service<Request> + ?Sized> Service<Request> for Box<S> {
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    #[inline]
+    fn call(&self, req: Request) -> Self::Future {
+        (**self).call(req)
+    }
+}
+
+impl<Request, S: Service<Request> + ?Sized> Service<Request> for std::rc::Rc<S> {
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    #[inline]
+    fn call(&self, req: Request) -> Self::Future {
+        (**self).call(req)
+    }
+}
+
+impl<Request, S: Service<Request> + ?Sized> Service<Request> for std::sync::Arc<S> {
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    #[inline]
+    fn call(&self, req: Request) -> Self::Future {
+        (**self).call(req)
+    }
+}
diff --git a/src/upgrade.rs b/src/upgrade.rs
index f1cb73d1..03d7e98d 100644
--- a/src/upgrade.rs
+++ b/src/upgrade.rs
@@ -79,6 +79,7 @@ pub struct OnUpgrade {
 /// Includes the original IO type, and a read buffer of bytes that the
 /// HTTP state machine may have already read before completing an upgrade.
 #[derive(Debug)]
+#[non_exhaustive]
 pub struct Parts<T> {
     /// The original IO object used before the upgrade.
     pub io: T,
@@ -91,7 +92,6 @@ pub struct Parts<T> {
     /// You will want to check for any existing bytes if you plan to continue
     /// communicating on the IO object.
     pub read_buf: Bytes,
-    _inner: (),
 }
 
 /// Gets a pending HTTP upgrade from this message.
@@ -154,7 +154,6 @@ impl Upgraded {
             Ok(t) => Ok(Parts {
                 io: *t,
                 read_buf: buf,
-                _inner: (),
             }),
             Err(io) => Err(Upgraded {
                 io: Rewind::new_buffered(io, buf),


Test patch:
diff --git a/tests/client.rs b/tests/client.rs
index 43e1f08a..89b90796 100644
--- a/tests/client.rs
+++ b/tests/client.rs
@@ -34,6 +34,17 @@ where
     b.collect().await.map(|c| c.to_bytes())
 }
 
+async fn concat_with_trailers<B>(b: B) -> Result<(Bytes, Option<HeaderMap>), B::Error>
+where
+    B: hyper::body::Body,
+{
+    let collect = b.collect().await?;
+    let trailers = collect.trailers().cloned();
+    let bytes = collect.to_bytes();
+
+    Ok((bytes, trailers))
+}
+
 async fn tcp_connect(addr: &SocketAddr) -> std::io::Result<TokioIo<TcpStream>> {
     TcpStream::connect(*addr).await.map(TokioIo::new)
 }
@@ -122,6 +133,9 @@ macro_rules! test {
                 status: $client_status:ident,
                 headers: { $($response_header_name:expr => $response_header_val:expr,)* },
                 body: $response_body:expr,
+                $(trailers: {$(
+                    $response_trailer_name:expr => $response_trailer_val:expr,
+                )*},)?
     ) => (
         #[test]
         fn $name() {
@@ -158,12 +172,23 @@ macro_rules! test {
                 );
             )*
 
-            let body = rt.block_on(concat(res))
+            let (body, _trailers) = rt.block_on(concat_with_trailers(res))
                 .expect("body concat wait");
 
             let expected_res_body = Option::<&[u8]>::from($response_body)
                 .unwrap_or_default();
             assert_eq!(body.as_ref(), expected_res_body);
+
+            $($(
+                assert_eq!(
+                    _trailers.as_ref().expect("trailers is None")
+                        .get($response_trailer_name)
+                        .expect(concat!("trailer header '", stringify!($response_trailer_name), "'")),
+                    $response_trailer_val,
+                    "trailer '{}'",
+                    stringify!($response_trailer_name),
+                );
+            )*)?
         }
     );
     (
@@ -679,6 +704,94 @@ test! {
             body: None,
 }
 
+test! {
+    name: client_res_body_chunked_with_trailer,
+
+    server:
+        expected: "GET / HTTP/1.1\r\nte: trailers\r\nhost: {addr}\r\n\r\n",
+        reply: "\
+            HTTP/1.1 200 OK\r\n\
+            transfer-encoding: chunked\r\n\
+            trailer: chunky-trailer\r\n\
+            \r\n\
+            5\r\n\
+            hello\r\n\
+            0\r\n\
+            chunky-trailer: header data\r\n\
+            \r\n\
+            ",
+
+    client:
+        request: {
+            method: GET,
+            url: "http://{addr}/",
+            headers: {
+                "te" => "trailers",
+            },
+        },
+        response:
+            status: OK,
+            headers: {
+                "Transfer-Encoding" => "chunked",
+            },
+            body: &b"hello"[..],
+            trailers: {
+                "chunky-trailer" => "header data",
+            },
+}
+
+test! {
+    name: client_res_body_chunked_with_pathological_trailers,
+
+    server:
+        expected: "GET / HTTP/1.1\r\nte: trailers\r\nhost: {addr}\r\n\r\n",
+        reply: "\
+            HTTP/1.1 200 OK\r\n\
+            transfer-encoding: chunked\r\n\
+            trailer: chunky-trailer1, chunky-trailer2, chunky-trailer3, chunky-trailer4, chunky-trailer5\r\n\
+            \r\n\
+            5\r\n\
+            hello\r\n\
+            0\r\n\
+            chunky-trailer1: header data1\r\n\
+            chunky-trailer2: header data2\r\n\
+            chunky-trailer3: header data3\r\n\
+            chunky-trailer4: header data4\r\n\
+            chunky-trailer5: header data5\r\n\
+            sneaky-trailer: not in trailer header\r\n\
+            transfer-encoding: chunked\r\n\
+            content-length: 5\r\n\
+            trailer: foo\r\n\
+            \r\n\
+            ",
+
+    client:
+        request: {
+            method: GET,
+            url: "http://{addr}/",
+            headers: {
+                "te" => "trailers",
+            },
+        },
+        response:
+            status: OK,
+            headers: {
+                "Transfer-Encoding" => "chunked",
+            },
+            body: &b"hello"[..],
+            trailers: {
+                "chunky-trailer1" => "header data1",
+                "chunky-trailer2" => "header data2",
+                "chunky-trailer3" => "header data3",
+                "chunky-trailer4" => "header data4",
+                "chunky-trailer5" => "header data5",
+                "sneaky-trailer" => "not in trailer header",
+                "transfer-encoding" => "chunked",
+                "content-length" => "5",
+                "trailer" => "foo",
+            },
+}
+
 test! {
     name: client_get_req_body_sized,
 
@@ -2009,6 +2122,82 @@ mod conn {
             .expect_err("client should be closed");
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn http2_connect_detect_close() {
+        // Regression test for failure to fully close connections when using HTTP2 CONNECT
+        // We send 2 requests and then drop them. We should see the connection gracefully close.
+        use futures_util::future;
+        let (listener, addr) = setup_tk_test_server().await;
+        let (tx, rxx) = oneshot::channel::<()>();
+
+        tokio::task::spawn(async move {
+            use hyper::server::conn::http2;
+            use hyper::service::service_fn;
+
+            let res = listener.accept().await;
+            let (stream, _) = res.unwrap();
+            let stream = TokioIo::new(stream);
+
+            let service = service_fn(move |req: Request<hyper::body::Incoming>| {
+                tokio::task::spawn(async move {
+                    let io = &mut TokioIo::new(hyper::upgrade::on(req).await.unwrap());
+                    io.write_all(b"hello\n").await.unwrap();
+                });
+
+                future::ok::<_, hyper::Error>(Response::new(Empty::<Bytes>::new()))
+            });
+
+            tokio::task::spawn(async move {
+                let conn = http2::Builder::new(TokioExecutor).serve_connection(stream, service);
+                let _ = conn.await;
+                tx.send(()).unwrap();
+            });
+        });
+
+        let io = tcp_connect(&addr).await.expect("tcp connect");
+        let (mut client, conn) = conn::http2::Builder::new(TokioExecutor)
+            .handshake(io)
+            .await
+            .expect("http handshake");
+
+        tokio::task::spawn(async move {
+            conn.await.expect("client conn");
+        });
+
+        // Sanity check that client is ready
+        future::poll_fn(|ctx| client.poll_ready(ctx))
+            .await
+            .expect("client poll ready sanity");
+        let requests = 2;
+        let mut clients = vec![client.clone(), client];
+        let (tx, rx) = oneshot::channel::<()>();
+        let (tx2, rx2) = oneshot::channel::<()>();
+        let mut rxs = vec![rx, rx2];
+        for _i in 0..requests {
+            let mut client = clients.pop().unwrap();
+            let rx = rxs.pop().unwrap();
+            let req = Request::builder()
+                .method(Method::CONNECT)
+                .uri(format!("{}", addr))
+                .body(Empty::<Bytes>::new())
+                .expect("request builder");
+
+            let resp = client.send_request(req).await.expect("req1 send");
+            assert_eq!(resp.status(), 200);
+            let upgrade = hyper::upgrade::on(resp).await.unwrap();
+            tokio::task::spawn(async move {
+                let _ = rx.await;
+                drop(upgrade);
+            });
+        }
+        drop(tx);
+        drop(tx2);
+        tokio::time::timeout(Duration::from_secs(1), rxx)
+            .await
+            .expect("drop with 1s")
+            .expect("tx dropped without sending");
+    }
+
     #[tokio::test]
     async fn http2_keep_alive_detects_unresponsive_server() {
         let (listener, addr) = setup_tk_test_server().await;
diff --git a/tests/server.rs b/tests/server.rs
index dcbe6695..f3efbf3c 100644
--- a/tests/server.rs
+++ b/tests/server.rs
@@ -1256,6 +1256,67 @@ async fn disable_keep_alive_post_request() {
     child.join().unwrap();
 }
 
+#[tokio::test]
+async fn http1_graceful_shutdown_after_upgrade() {
+    let (listener, addr) = setup_tcp_listener();
+    let (read_101_tx, read_101_rx) = oneshot::channel();
+
+    thread::spawn(move || {
+        let mut tcp = connect(&addr);
+        tcp.write_all(
+            b"\
+            GET / HTTP/1.1\r\n\
+            Upgrade: foobar\r\n\
+            Connection: upgrade\r\n\
+            \r\n\
+            eagerly optimistic\
+        ",
+        )
+        .expect("write 1");
+        let mut buf = [0; 256];
+        tcp.read(&mut buf).expect("read 1");
+
+        let response = s(&buf);
+        assert!(response.starts_with("HTTP/1.1 101 Switching Protocols\r\n"));
+        assert!(!has_header(&response, "content-length"));
+        let _ = read_101_tx.send(());
+    });
+
+    let (upgrades_tx, upgrades_rx) = mpsc::channel();
+    let svc = service_fn(move |req: Request<IncomingBody>| {
+        let on_upgrade = hyper::upgrade::on(req);
+        let _ = upgrades_tx.send(on_upgrade);
+        future::ok::<_, hyper::Error>(
+            Response::builder()
+                .status(101)
+                .header("upgrade", "foobar")
+                .body(Empty::<Bytes>::new())
+                .unwrap(),
+        )
+    });
+
+    let (socket, _) = listener.accept().await.unwrap();
+    let socket = TokioIo::new(socket);
+
+    let mut conn = http1::Builder::new()
+        .serve_connection(socket, svc)
+        .with_upgrades();
+    (&mut conn).await.unwrap();
+
+    let on_upgrade = upgrades_rx.recv().unwrap();
+
+    // wait so that we don't write until other side saw 101 response
+    read_101_rx.await.unwrap();
+
+    let upgraded = on_upgrade.await.expect("on_upgrade");
+    let parts = upgraded.downcast::<TokioIo<TkTcpStream>>().unwrap();
+    assert_eq!(parts.read_buf, "eagerly optimistic");
+
+    pin!(conn);
+    // graceful shutdown doesn't cause issues or panic. It should be ignored after upgrade
+    conn.as_mut().graceful_shutdown();
+}
+
 #[tokio::test]
 async fn empty_parse_eof_does_not_return_error() {
     let (listener, addr) = setup_tcp_listener();
@@ -2468,6 +2529,7 @@ async fn http2_keep_alive_detects_unresponsive_client() {
         .timer(TokioTimer)
         .keep_alive_interval(Duration::from_secs(1))
         .keep_alive_timeout(Duration::from_secs(1))
+        .auto_date_header(true)
         .serve_connection(socket, unreachable_service())
         .await
         .expect_err("serve_connection should error");
@@ -2508,6 +2570,42 @@ async fn http2_keep_alive_with_responsive_client() {
     client.send_request(req).await.expect("client.send_request");
 }
 
+#[tokio::test]
+async fn http2_check_date_header_disabled() {
+    let (listener, addr) = setup_tcp_listener();
+
+    tokio::spawn(async move {
+        let (socket, _) = listener.accept().await.expect("accept");
+        let socket = TokioIo::new(socket);
+
+        http2::Builder::new(TokioExecutor)
+            .timer(TokioTimer)
+            .keep_alive_interval(Duration::from_secs(1))
+            .auto_date_header(false)
+            .keep_alive_timeout(Duration::from_secs(1))
+            .serve_connection(socket, HelloWorld)
+            .await
+            .expect("serve_connection");
+    });
+
+    let tcp = TokioIo::new(connect_async(addr).await);
+    let (mut client, conn) = hyper::client::conn::http2::Builder::new(TokioExecutor)
+        .handshake(tcp)
+        .await
+        .expect("http handshake");
+
+    tokio::spawn(async move {
+        conn.await.expect("client conn");
+    });
+
+    TokioTimer.sleep(Duration::from_secs(4)).await;
+
+    let req = http::Request::new(Empty::<Bytes>::new());
+    let resp = client.send_request(req).await.expect("client.send_request");
+
+    assert!(resp.headers().get("Date").is_none());
+}
+
 fn is_ping_frame(buf: &[u8]) -> bool {
     buf[3] == 6
 }
@@ -2595,7 +2693,7 @@ async fn http2_keep_alive_count_server_pings() {
 }
 
 #[test]
-fn http1_trailer_fields() {
+fn http1_trailer_send_fields() {
     let body = futures_util::stream::once(async move { Ok("hello".into()) });
     let mut headers = HeaderMap::new();
     headers.insert("chunky-trailer", "header data".parse().unwrap());
@@ -2682,6 +2780,35 @@ fn http1_trailer_fields_not_allowed() {
     assert_eq!(body, expected_body);
 }
 
+#[test]
+fn http1_trailer_recv_fields() {
+    let server = serve();
+    let mut req = connect(server.addr());
+    req.write_all(
+        b"\
+        POST / HTTP/1.1\r\n\
+        trailer: chunky-trailer\r\n\
+        host: example.domain\r\n\
+        transfer-encoding: chunked\r\n\
+        \r\n\
+        5\r\n\
+        hello\r\n\
+        0\r\n\
+        chunky-trailer: header data\r\n\
+        \r\n\
+    ",
+    )
+    .expect("writing");
+
+    assert_eq!(server.body(), b"hello");
+
+    let trailers = server.trailers();
+    assert_eq!(
+        trailers.get("chunky-trailer"),
+        Some(&"header data".parse().unwrap())
+    );
+}
+
 // -------------------------------------------------
 // the Server that is used to run all the tests with
 // -------------------------------------------------
@@ -2689,6 +2816,7 @@ fn http1_trailer_fields_not_allowed() {
 struct Serve {
     addr: SocketAddr,
     msg_rx: mpsc::Receiver<Msg>,
+    trailers_rx: mpsc::Receiver<HeaderMap>,
     reply_tx: Mutex<spmc::Sender<Reply>>,
     shutdown_signal: Option<oneshot::Sender<()>>,
     thread: Option<thread::JoinHandle<()>>,
@@ -2722,6 +2850,10 @@ impl Serve {
         Ok(buf)
     }
 
+    fn trailers(&self) -> HeaderMap {
+        self.trailers_rx.recv().expect("trailers")
+    }
+
     fn reply(&self) -> ReplyBuilder<'_> {
         ReplyBuilder { tx: &self.reply_tx }
     }
@@ -2835,6 +2967,7 @@ impl Drop for Serve {
 #[derive(Clone)]
 struct TestService {
     tx: mpsc::Sender<Msg>,
+    trailers_tx: mpsc::Sender<HeaderMap>,
     reply: spmc::Receiver<Reply>,
 }
 
@@ -2865,6 +2998,7 @@ impl Service<Request<IncomingBody>> for TestService {
 
     fn call(&self, mut req: Request<IncomingBody>) -> Self::Future {
         let tx = self.tx.clone();
+        let trailers_tx = self.trailers_tx.clone();
         let replies = self.reply.clone();
 
         Box::pin(async move {
@@ -2874,6 +3008,9 @@ impl Service<Request<IncomingBody>> for TestService {
                         if frame.is_data() {
                             tx.send(Msg::Chunk(frame.into_data().unwrap().to_vec()))
                                 .unwrap();
+                        } else if frame.is_trailers() {
+                            let trailers = frame.into_trailers().unwrap();
+                            trailers_tx.send(trailers).unwrap();
                         }
                     }
                     Err(err) => {
@@ -3002,6 +3139,7 @@ impl ServeOptions {
 
         let (addr_tx, addr_rx) = mpsc::channel();
         let (msg_tx, msg_rx) = mpsc::channel();
+        let (trailers_tx, trailers_rx) = mpsc::channel();
         let (reply_tx, reply_rx) = spmc::channel();
         let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
 
@@ -3025,6 +3163,7 @@ impl ServeOptions {
 
                     loop {
                         let msg_tx = msg_tx.clone();
+                        let trailers_tx = trailers_tx.clone();
                         let reply_rx = reply_rx.clone();
 
                         tokio::select! {
@@ -3037,6 +3176,7 @@ impl ServeOptions {
                                     let reply_rx = reply_rx.clone();
                                     let service = TestService {
                                         tx: msg_tx,
+                                        trailers_tx,
                                         reply: reply_rx,
                                     };
 
@@ -3064,6 +3204,7 @@ impl ServeOptions {
 
         Serve {
             msg_rx,
+            trailers_rx,
             reply_tx: Mutex::new(reply_tx),
             addr,
             shutdown_signal: Some(shutdown_tx),