refactor(client): re-enable http2 client

Author: seanmonstar

Cargo.toml

@@ -29,7 +29,7 @@ futures-util-preview = { version = "0.3.0-alpha.17" }
 http = "0.1.15"
 http-body = "0.1"
 httparse = "1.0"
-h2 = "0.1.10"
+h2 = { git = "https://github.com/hyperium/h2" }
 iovec = "0.1"
 itoa = "0.4.1"
 log = "0.4"

src/client/conn.rs

@@ -32,18 +32,19 @@ type Http1Dispatcher<T, B, R> = proto::dispatch::Dispatcher<
 >;
 type ConnEither<T, B> = Either<
     Http1Dispatcher<T, B, proto::h1::ClientTransaction>,
-    proto::h2::Client<T, B>,
+    proto::h2::ClientTask<B>,
 >;
 
-/// Returns a `Handshake` future over some IO.
+/// Returns a handshake future over some IO.
 ///
 /// This is a shortcut for `Builder::new().handshake(io)`.
-pub fn handshake<T>(io: T) -> Handshake<T, crate::Body>
+pub async fn handshake<T>(io: T) -> crate::Result<(SendRequest<crate::Body>, Connection<T, crate::Body>)>
 where
     T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
 {
     Builder::new()
         .handshake(io)
+        .await
 }
 
 /// The sender side of an established connection.
@@ -68,7 +69,7 @@ where
 
 /// A builder to configure an HTTP connection.
 ///
-/// After setting options, the builder is used to create a `Handshake` future.
+/// After setting options, the builder is used to create a handshake future.
 #[derive(Clone, Debug)]
 pub struct Builder {
     pub(super) exec: Exec,
@@ -80,16 +81,6 @@ pub struct Builder {
     h2_builder: h2::client::Builder,
 }
 
-/// A future setting up HTTP over an IO object.
-///
-/// If successful, yields a `(SendRequest, Connection)` pair.
-#[must_use = "futures do nothing unless polled"]
-pub struct Handshake<T, B> {
-    builder: Builder,
-    io: Option<T>,
-    _marker: PhantomData<fn(B)>,
-}
-
 /// A future returned by `SendRequest::send_request`.
 ///
 /// Yields a `Response` if successful.
@@ -400,6 +391,7 @@ impl<T, B> Future for Connection<T, B>
 where
     T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     B: Payload + Unpin + 'static,
+    B::Data: Unpin,
 {
     type Output = crate::Result<()>;
 
@@ -522,70 +514,46 @@ impl Builder {
     }
 
     /// Constructs a connection with the configured options and IO.
-    #[inline]
-    pub fn handshake<T, B>(&self, io: T) -> Handshake<T, B>
+    pub async fn handshake<T, B>(self, io: T) -> crate::Result<(SendRequest<B>, Connection<T, B>)>
     where
         T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
         B: Payload + 'static,
+        B::Data: Unpin,
     {
         trace!("client handshake HTTP/{}", if self.http2 { 2 } else { 1 });
-        Handshake {
-            builder: self.clone(),
-            io: Some(io),
-            _marker: PhantomData,
-        }
-    }
-}
-
-// ===== impl Handshake
-
-impl<T, B> Future for Handshake<T, B>
-where
-    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
-    B: Payload + 'static,
-{
-    type Output = crate::Result<(SendRequest<B>, Connection<T, B>)>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
-        let io = self.io.take().expect("polled more than once");
         let (tx, rx) = dispatch::channel();
-        let either = if !self.builder.http2 {
+        let either = if !self.http2 {
             let mut conn = proto::Conn::new(io);
-            if !self.builder.h1_writev {
+            if !self.h1_writev {
                 conn.set_write_strategy_flatten();
             }
-            if self.builder.h1_title_case_headers {
+            if self.h1_title_case_headers {
                 conn.set_title_case_headers();
             }
-            if let Some(sz) = self.builder.h1_read_buf_exact_size {
+            if let Some(sz) = self.h1_read_buf_exact_size {
                 conn.set_read_buf_exact_size(sz);
             }
-            if let Some(max) = self.builder.h1_max_buf_size {
+            if let Some(max) = self.h1_max_buf_size {
                 conn.set_max_buf_size(max);
             }
             let cd = proto::h1::dispatch::Client::new(rx);
             let dispatch = proto::h1::Dispatcher::new(cd, conn);
             Either::Left(dispatch)
         } else {
-            let h2 = proto::h2::Client::new(io, rx, &self.builder.h2_builder, self.builder.exec.clone());
+            let h2 = proto::h2::client::handshake(io, rx, &self.h2_builder, self.exec.clone())
+                .await?;
             Either::Right(h2)
         };
 
-        Poll::Ready(Ok((
+        Ok((
             SendRequest {
                 dispatch: tx,
             },
             Connection {
                 inner: Some(either),
             },
-        )))
-    }
-}
-
-impl<T, B> fmt::Debug for Handshake<T, B> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Handshake")
-            .finish()
+        ))
     }
 }
 

src/client/mod.rs

@@ -163,7 +163,7 @@ where C: Connect + Sync + 'static,
       C::Transport: 'static,
       C::Future: 'static,
       B: Payload + Unpin + Send + 'static,
-      B::Data: Send,
+      B::Data: Send + Unpin,
 {
     /// Send a `GET` request to the supplied `Uri`.
     ///
@@ -512,8 +512,10 @@ where C: Connect + Sync + 'static,
                         connecting
                     };
                     let is_h2 = is_ver_h2 || connected.alpn == Alpn::H2;
-                    Either::Left(conn_builder
+                    Either::Left(Box::pin(conn_builder
                         .http2_only(is_h2)
+                        // TODO: convert client::conn::Builder to be by-value?
+                        .clone()
                         .handshake(io)
                         .and_then(move |(tx, conn)| {
                             trace!("handshake complete, spawning background dispatcher task");
@@ -541,7 +543,7 @@ where C: Connect + Sync + 'static,
                                     PoolTx::Http1(tx)
                                 },
                             })
-                        }))
+                        })))
                 }))
         })
     }