Allow building an HTTP/1 connection with prebuffered bytes

Fixes #3704.

src/proto/h1/conn.rs

@@ -48,8 +48,12 @@ where
     T: Http1Transaction,
 {
     pub(crate) fn new(io: I) -> Conn<I, B, T> {
+        Self::new_buffered(Bytes::new(), io)
+    }
+
+    pub(crate) fn new_buffered(buffered: Bytes, io: I) -> Conn<I, B, T> {
         Conn {
-            io: Buffered::new(io),
+            io: Buffered::new_buffered(buffered, io),
             state: State {
                 allow_half_close: false,
                 cached_headers: None,

src/proto/h1/io.rs

@@ -55,7 +55,12 @@ where
     T: Read + Write + Unpin,
     B: Buf,
 {
+    #[cfg(test)]
     pub(crate) fn new(io: T) -> Buffered<T, B> {
+        Self::new_buffered(Bytes::new(), io)
+    }
+
+    pub(crate) fn new_buffered(buffered: Bytes, io: T) -> Buffered<T, B> {
         let strategy = if io.is_write_vectored() {
             WriteStrategy::Queue
         } else {
@@ -66,7 +71,9 @@ where
             flush_pipeline: false,
             io,
             read_blocked: false,
-            read_buf: BytesMut::with_capacity(0),
+            // FIXME: This should use `From<Bytes>` which has not been released yet.
+            // https://github.com/tokio-rs/bytes/commit/fa1daac3ae1dcb07dffe3a41a041dffd6edf177b
+            read_buf: BytesMut::from(&*buffered),
             read_buf_strategy: ReadStrategy::default(),
             write_buf,
         }

src/server/conn/http1.rs

@@ -425,7 +425,27 @@ impl Builder {
         <S::ResBody as Body>::Error: Into<Box<dyn StdError + Send + Sync>>,
         I: Read + Write + Unpin,
     {
-        let mut conn = proto::Conn::new(io);
+        self.serve_buffered_connection(Bytes::new(), io, service)
+    }
+
+    /// Bind a connection together with a [`Service`](crate::service::Service)
+    /// with a prebuffered chunk of bytes.
+    ///
+    /// See [`Self::serve_connection`]  for more info.
+    pub fn serve_buffered_connection<I, S>(
+        &self,
+        buffered: Bytes,
+        io: I,
+        service: S,
+    ) -> Connection<I, S>
+    where
+        S: HttpService<IncomingBody>,
+        S::Error: Into<Box<dyn StdError + Send + Sync>>,
+        S::ResBody: 'static,
+        <S::ResBody as Body>::Error: Into<Box<dyn StdError + Send + Sync>>,
+        I: Read + Write + Unpin,
+    {
+        let mut conn = proto::Conn::new_buffered(buffered, io);
         conn.set_timer(self.timer.clone());
         if !self.h1_keep_alive {
             conn.disable_keep_alive();