hyperium · Michael-J-Ward · Aug 24, 2022 · Aug 24, 2022 · Aug 24, 2022 · Aug 24, 2022
diff --git a/examples/client.rs b/examples/client.rs
@@ -39,7 +39,7 @@ async fn fetch_url(url: hyper::Uri) -> Result<()> {
     let addr = format!("{}:{}", host, port);
     let stream = TcpStream::connect(addr).await?;
 
-    let (mut sender, conn) = hyper::client::conn::handshake(stream).await?;
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(stream).await?;
     tokio::task::spawn(async move {
         if let Err(err) = conn.await {
             println!("Connection failed: {:?}", err);

diff --git a/examples/client_json.rs b/examples/client_json.rs
@@ -29,7 +29,7 @@ async fn fetch_json(url: hyper::Uri) -> Result<Vec<User>> {
 
     let stream = TcpStream::connect(addr).await?;
 
-    let (mut sender, conn) = hyper::client::conn::handshake(stream).await?;
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(stream).await?;
     tokio::task::spawn(async move {
         if let Err(err) = conn.await {
             println!("Connection failed: {:?}", err);

diff --git a/examples/gateway.rs b/examples/gateway.rs
@@ -43,7 +43,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             async move {
                 let client_stream = TcpStream::connect(addr).await.unwrap();
 
-                let (mut sender, conn) = hyper::client::conn::handshake(client_stream).await?;
+                let (mut sender, conn) =
+                    hyper::client::conn::http1::handshake(client_stream).await?;
                 tokio::task::spawn(async move {
                     if let Err(err) = conn.await {
                         println!("Connection failed: {:?}", err);

diff --git a/examples/http_proxy.rs b/examples/http_proxy.rs
@@ -2,7 +2,7 @@
 
 use std::net::SocketAddr;
 
-use hyper::client::conn::Builder;
+use hyper::client::conn::http1::Builder;
 use hyper::server::conn::Http;
 use hyper::service::service_fn;
 use hyper::upgrade::Upgraded;

diff --git a/examples/upgrades.rs b/examples/upgrades.rs
@@ -95,7 +95,7 @@ async fn client_upgrade_request(addr: SocketAddr) -> Result<()> {
         .unwrap();
 
     let stream = TcpStream::connect(addr).await?;
-    let (mut sender, conn) = hyper::client::conn::handshake(stream).await?;
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(stream).await?;
 
     tokio::task::spawn(async move {
         if let Err(err) = conn.await {

diff --git a/examples/web_api.rs b/examples/web_api.rs
@@ -29,7 +29,7 @@ async fn client_request_response() -> Result<Response<Body>> {
     let port = req.uri().port_u16().expect("uri has no port");
     let stream = TcpStream::connect(format!("{}:{}", host, port)).await?;
 
-    let (mut sender, conn) = hyper::client::conn::handshake(stream).await?;
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(stream).await?;
 
     tokio::task::spawn(async move {
         if let Err(err) = conn.await {

diff --git a/src/client/conn/http1.rs b/src/client/conn/http1.rs
@@ -4,6 +4,7 @@ use std::error::Error as StdError;
 use std::fmt;
 use std::sync::Arc;
 
+use bytes::Bytes;
 use http::{Request, Response};
 use httparse::ParserConfig;
 use tokio::io::{AsyncRead, AsyncWrite};
@@ -27,6 +28,27 @@ pub struct SendRequest<B> {
     dispatch: dispatch::Sender<Request<B>, Response<Body>>,
 }
 
+/// Deconstructed parts of a `Connection`.
+///
+/// This allows taking apart a `Connection` at a later time, in order to
+/// reclaim the IO object, and additional related pieces.
+#[derive(Debug)]
+pub struct Parts<T> {
+    /// The original IO object used in the handshake.
+    pub io: T,
+    /// A buffer of bytes that have been read but not processed as HTTP.
+    ///
+    /// For instance, if the `Connection` is used for an HTTP upgrade request,
+    /// it is possible the server sent back the first bytes of the new protocol
+    /// along with the response upgrade.
+    ///
+    /// 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.
 ///
 /// In most cases, this should just be spawned into an executor, so that it
@@ -40,6 +62,41 @@ where
     inner: Option<Dispatcher<T, B>>,
 }
 
+impl<T, B> Connection<T, B>
+where
+    T: AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    B: HttpBody + 'static,
+    B::Error: Into<Box<dyn StdError + Send + Sync>>,
+{
+    /// Return the inner IO object, and additional information.
+    ///
+    /// Only works for HTTP/1 connections. HTTP/2 connections will panic.
+    pub fn into_parts(self) -> Parts<T> {
+
+        let (io, read_buf, _) = self.inner.expect("already upgraded").into_inner();
+        Parts {
+            io,
+            read_buf,
+            _inner: (),
+        }
+    }
+
+    /// Poll the connection for completion, but without calling `shutdown`
+    /// on the underlying IO.
+    ///
+    /// This is useful to allow running a connection while doing an HTTP
+    /// upgrade. Once the upgrade is completed, the connection would be "done",
+    /// but it is not desired to actually shutdown the IO object. Instead you
+    /// would take it back using `into_parts`.
+    ///
+    /// Use [`poll_fn`](https://docs.rs/futures/0.1.25/futures/future/fn.poll_fn.html)
+    /// and [`try_ready!`](https://docs.rs/futures/0.1.25/futures/macro.try_ready.html)
+    /// to work with this function; or use the `without_shutdown` wrapper.
+    pub fn poll_without_shutdown(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
+        self.inner.as_mut().expect("algready upgraded").poll_without_shutdown(cx)
+    }
+}
+
 /// A builder to configure an HTTP connection.
 ///
 /// After setting options, the builder is used to create a handshake future.
@@ -80,6 +137,13 @@ impl<B> SendRequest<B> {
         self.dispatch.poll_ready(cx)
     }
 
+    /// Waits until the dispatcher is ready
+    ///
+    /// If the associated connection is closed, this returns an Error.
+    pub async fn ready(&mut self) -> crate::Result<()> {
+        futures_util::future::poll_fn(|cx| self.poll_ready(cx)).await
+    }
+
     /*
     pub(super) async fn when_ready(self) -> crate::Result<Self> {
         let mut me = Some(self);
@@ -125,7 +189,7 @@ where
     ///
     /// ```
     /// # use http::header::HOST;
-    /// # use hyper::client::conn::SendRequest;
+    /// # use hyper::client::conn::http1::SendRequest;
     /// # use hyper::Body;
     /// use hyper::Request;
     ///

diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs
@@ -74,6 +74,13 @@ impl<B> SendRequest<B> {
         }
     }
 
+    /// Waits until the dispatcher is ready
+    ///
+    /// If the associated connection is closed, this returns an Error.
+    pub async fn ready(&mut self) -> crate::Result<()> {
+        futures_util::future::poll_fn(|cx| self.poll_ready(cx)).await
+    }
+
     /*
     pub(super) async fn when_ready(self) -> crate::Result<Self> {
         let mut me = Some(self);
@@ -119,7 +126,7 @@ where
     ///
     /// ```
     /// # use http::header::HOST;
-    /// # use hyper::client::conn::SendRequest;
+    /// # use hyper::client::conn::http2::SendRequest;
     /// # use hyper::Body;
     /// use hyper::Request;
     ///