feat(client): remove higher-level hyper::Client (#2941)

This removes the following types and methods from hyper:

- `Client`
- `Error::is_connect()`

BREAKING CHANGE: A pooling client is in the hyper-util crate.

Author: seanmonstar

src/body/body.rs

@@ -10,8 +10,6 @@ use http_body::{Body as HttpBody, SizeHint};
 
 use super::DecodedLength;
 use crate::common::Future;
-#[cfg(all(feature = "client", any(feature = "http1", feature = "http2")))]
-use crate::common::Never;
 use crate::common::{task, watch, Pin, Poll};
 #[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
 use crate::proto::h2::ping;
@@ -29,9 +27,6 @@ type TrailersSender = oneshot::Sender<HeaderMap>;
 #[must_use = "streams do nothing unless polled"]
 pub struct Body {
     kind: Kind,
-    /// Keep the extra bits in an `Option<Box<Extra>>`, so that
-    /// Body stays small in the common case (no extras needed).
-    extra: Option<Box<Extra>>,
 }
 
 enum Kind {
@@ -52,34 +47,6 @@ enum Kind {
     Ffi(crate::ffi::UserBody),
 }
 
-struct Extra {
-    /// Allow the client to pass a future to delay the `Body` from returning
-    /// EOF. This allows the `Client` to try to put the idle connection
-    /// back into the pool before the body is "finished".
-    ///
-    /// The reason for this is so that creating a new request after finishing
-    /// streaming the body of a response could sometimes result in creating
-    /// a brand new connection, since the pool didn't know about the idle
-    /// connection yet.
-    delayed_eof: Option<DelayEof>,
-}
-
-#[cfg(all(feature = "client", any(feature = "http1", feature = "http2")))]
-type DelayEofUntil = oneshot::Receiver<Never>;
-
-enum DelayEof {
-    /// Initial state, stream hasn't seen EOF yet.
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    NotEof(DelayEofUntil),
-    /// Transitions to this state once we've seen `poll` try to
-    /// return EOF (`None`). This future is then polled, and
-    /// when it completes, the Body finally returns EOF (`None`).
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    Eof(DelayEofUntil),
-}
-
 /// A sender half created through [`Body::channel()`].
 ///
 /// Useful when wanting to stream chunks from another thread.
@@ -153,7 +120,7 @@ impl Body {
     }
 
     fn new(kind: Kind) -> Body {
-        Body { kind, extra: None }
+        Body { kind }
     }
 
     #[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
@@ -176,62 +143,6 @@ impl Body {
         body
     }
 
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    pub(crate) fn delayed_eof(&mut self, fut: DelayEofUntil) {
-        self.extra_mut().delayed_eof = Some(DelayEof::NotEof(fut));
-    }
-
-    fn take_delayed_eof(&mut self) -> Option<DelayEof> {
-        self.extra
-            .as_mut()
-            .and_then(|extra| extra.delayed_eof.take())
-    }
-
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    fn extra_mut(&mut self) -> &mut Extra {
-        self.extra
-            .get_or_insert_with(|| Box::new(Extra { delayed_eof: None }))
-    }
-
-    fn poll_eof(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<crate::Result<Bytes>>> {
-        match self.take_delayed_eof() {
-            #[cfg(any(feature = "http1", feature = "http2"))]
-            #[cfg(feature = "client")]
-            Some(DelayEof::NotEof(mut delay)) => match self.poll_inner(cx) {
-                ok @ Poll::Ready(Some(Ok(..))) | ok @ Poll::Pending => {
-                    self.extra_mut().delayed_eof = Some(DelayEof::NotEof(delay));
-                    ok
-                }
-                Poll::Ready(None) => match Pin::new(&mut delay).poll(cx) {
-                    Poll::Ready(Ok(never)) => match never {},
-                    Poll::Pending => {
-                        self.extra_mut().delayed_eof = Some(DelayEof::Eof(delay));
-                        Poll::Pending
-                    }
-                    Poll::Ready(Err(_done)) => Poll::Ready(None),
-                },
-                Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
-            },
-            #[cfg(any(feature = "http1", feature = "http2"))]
-            #[cfg(feature = "client")]
-            Some(DelayEof::Eof(mut delay)) => match Pin::new(&mut delay).poll(cx) {
-                Poll::Ready(Ok(never)) => match never {},
-                Poll::Pending => {
-                    self.extra_mut().delayed_eof = Some(DelayEof::Eof(delay));
-                    Poll::Pending
-                }
-                Poll::Ready(Err(_done)) => Poll::Ready(None),
-            },
-            #[cfg(any(
-                not(any(feature = "http1", feature = "http2")),
-                not(feature = "client")
-            ))]
-            Some(delay_eof) => match delay_eof {},
-            None => self.poll_inner(cx),
-        }
-    }
-
     #[cfg(feature = "ffi")]
     pub(crate) fn as_ffi_mut(&mut self) -> &mut crate::ffi::UserBody {
         match self.kind {
@@ -313,7 +224,7 @@ impl HttpBody for Body {
         mut self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-        self.poll_eof(cx)
+        self.poll_inner(cx)
     }
 
     fn poll_trailers(

src/client/client.rs

@@ -22,7 +22,7 @@ use super::super::dispatch;
 
 /// The sender side of an established connection.
 pub struct SendRequest<B> {
-    dispatch: dispatch::Sender<Request<B>, Response<Body>>,
+    dispatch: dispatch::UnboundedSender<Request<B>, Response<Body>>,
 }
 
 /// A future that processes all HTTP state for the IO object.
@@ -66,8 +66,12 @@ impl<B> SendRequest<B> {
     /// Polls to determine whether this sender can be used yet for a request.
     ///
     /// If the associated connection is closed, this returns an Error.
-    pub fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
-        self.dispatch.poll_ready(cx)
+    pub fn poll_ready(&mut self, _cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
+        if self.is_closed() {
+            Poll::Ready(Err(crate::Error::new_closed()))
+        } else {
+            Poll::Ready(Ok(()))
+        }
     }
 
     /*
@@ -83,11 +87,11 @@ impl<B> SendRequest<B> {
     pub(super) fn is_ready(&self) -> bool {
         self.dispatch.is_ready()
     }
+    */
 
     pub(super) fn is_closed(&self) -> bool {
         self.dispatch.is_closed()
     }
-    */
 }
 
 impl<B> SendRequest<B>
@@ -423,7 +427,7 @@ impl Builder {
                 proto::h2::client::handshake(io, rx, &opts.h2_builder, opts.exec)
                     .await?;
             Ok((
-                SendRequest { dispatch: tx },
+                SendRequest { dispatch: tx.unbound() },
                 Connection { inner: (PhantomData, h2) },
             ))
         }

src/client/conn/http2.rs

@@ -63,7 +63,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use bytes::Bytes;
-use futures_util::future::{self, Either, FutureExt as _};
+use futures_util::future;
 use httparse::ParserConfig;
 use pin_project_lite::pin_project;
 use tokio::io::{AsyncRead, AsyncWrite};
@@ -214,16 +214,6 @@ pub struct Parts<T> {
     _inner: (),
 }
 
-// ========== internal client api
-
-// A `SendRequest` that can be cloned to send HTTP2 requests.
-// private for now, probably not a great idea of a type...
-#[must_use = "futures do nothing unless polled"]
-#[cfg(feature = "http2")]
-pub(super) struct Http2SendRequest<B> {
-    dispatch: dispatch::UnboundedSender<Request<B>, Response<Body>>,
-}
-
 // ===== impl SendRequest
 
 impl<B> SendRequest<B> {
@@ -233,30 +223,6 @@ impl<B> SendRequest<B> {
     pub fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
         self.dispatch.poll_ready(cx)
     }
-
-    pub(super) async fn when_ready(self) -> crate::Result<Self> {
-        let mut me = Some(self);
-        future::poll_fn(move |cx| {
-            ready!(me.as_mut().unwrap().poll_ready(cx))?;
-            Poll::Ready(Ok(me.take().unwrap()))
-        })
-        .await
-    }
-
-    pub(super) fn is_ready(&self) -> bool {
-        self.dispatch.is_ready()
-    }
-
-    pub(super) fn is_closed(&self) -> bool {
-        self.dispatch.is_closed()
-    }
-
-    #[cfg(feature = "http2")]
-    pub(super) fn into_http2(self) -> Http2SendRequest<B> {
-        Http2SendRequest {
-            dispatch: self.dispatch.unbound(),
-        }
-    }
 }
 
 impl<B> SendRequest<B>
@@ -316,32 +282,6 @@ where
 
         ResponseFuture { inner }
     }
-
-    pub(super) fn send_request_retryable(
-        &mut self,
-        req: Request<B>,
-    ) -> impl Future<Output = Result<Response<Body>, (crate::Error, Option<Request<B>>)>> + Unpin
-    where
-        B: Send,
-    {
-        match self.dispatch.try_send(req) {
-            Ok(rx) => {
-                Either::Left(rx.then(move |res| {
-                    match res {
-                        Ok(Ok(res)) => future::ok(res),
-                        Ok(Err(err)) => future::err(err),
-                        // this is definite bug if it happens, but it shouldn't happen!
-                        Err(_) => panic!("dispatch dropped without returning error"),
-                    }
-                }))
-            }
-            Err(req) => {
-                debug!("connection was not ready");
-                let err = crate::Error::new_canceled().with("connection was not ready");
-                Either::Right(future::err((err, Some(req))))
-            }
-        }
-    }
 }
 
 impl<B> Service<Request<B>> for SendRequest<B>
@@ -367,67 +307,6 @@ impl<B> fmt::Debug for SendRequest<B> {
     }
 }
 
-// ===== impl Http2SendRequest
-
-#[cfg(feature = "http2")]
-impl<B> Http2SendRequest<B> {
-    pub(super) fn is_ready(&self) -> bool {
-        self.dispatch.is_ready()
-    }
-
-    pub(super) fn is_closed(&self) -> bool {
-        self.dispatch.is_closed()
-    }
-}
-
-#[cfg(feature = "http2")]
-impl<B> Http2SendRequest<B>
-where
-    B: HttpBody + 'static,
-{
-    pub(super) fn send_request_retryable(
-        &mut self,
-        req: Request<B>,
-    ) -> impl Future<Output = Result<Response<Body>, (crate::Error, Option<Request<B>>)>>
-    where
-        B: Send,
-    {
-        match self.dispatch.try_send(req) {
-            Ok(rx) => {
-                Either::Left(rx.then(move |res| {
-                    match res {
-                        Ok(Ok(res)) => future::ok(res),
-                        Ok(Err(err)) => future::err(err),
-                        // this is definite bug if it happens, but it shouldn't happen!
-                        Err(_) => panic!("dispatch dropped without returning error"),
-                    }
-                }))
-            }
-            Err(req) => {
-                debug!("connection was not ready");
-                let err = crate::Error::new_canceled().with("connection was not ready");
-                Either::Right(future::err((err, Some(req))))
-            }
-        }
-    }
-}
-
-#[cfg(feature = "http2")]
-impl<B> fmt::Debug for Http2SendRequest<B> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Http2SendRequest").finish()
-    }
-}
-
-#[cfg(feature = "http2")]
-impl<B> Clone for Http2SendRequest<B> {
-    fn clone(&self) -> Self {
-        Http2SendRequest {
-            dispatch: self.dispatch.clone(),
-        }
-    }
-}
-
 // ===== impl Connection
 
 impl<T, B> Connection<T, B>

src/client/conn/mod.rs

@@ -166,6 +166,7 @@ impl Connected {
         self.alpn == Alpn::H2
     }
 
+    /*
     // Don't public expose that `Connected` is `Clone`, unsure if we want to
     // keep that contract...
     #[cfg(feature = "http2")]
@@ -176,6 +177,7 @@ impl Connected {
             extra: self.extra.clone(),
         }
     }
+    */
 }
 
 // ===== impl Extra =====

src/client/connect/mod.rs

@@ -7,6 +7,7 @@ use tokio::sync::{mpsc, oneshot};
 use crate::common::Pin;
 use crate::common::{task, Poll};
 
+#[cfg(test)]
 pub(crate) type RetryPromise<T, U> = oneshot::Receiver<Result<U, (crate::Error, Option<T>)>>;
 pub(crate) type Promise<T> = oneshot::Receiver<Result<T, crate::Error>>;
 
@@ -58,13 +59,16 @@ impl<T, U> Sender<T, U> {
             .map_err(|_| crate::Error::new_closed())
     }
 
+    #[cfg(test)]
     pub(crate) fn is_ready(&self) -> bool {
         self.giver.is_wanting()
     }
 
+    /*
     pub(crate) fn is_closed(&self) -> bool {
         self.giver.is_canceled()
     }
+    */
 
     fn can_send(&mut self) -> bool {
         if self.giver.give() || !self.buffered_once {
@@ -79,6 +83,7 @@ impl<T, U> Sender<T, U> {
         }
     }
 
+    #[cfg(test)]
     pub(crate) fn try_send(&mut self, val: T) -> Result<RetryPromise<T, U>, T> {
         if !self.can_send() {
             return Err(val);
@@ -112,21 +117,32 @@ impl<T, U> Sender<T, U> {
 
 #[cfg(feature = "http2")]
 impl<T, U> UnboundedSender<T, U> {
+    /*
     pub(crate) fn is_ready(&self) -> bool {
         !self.giver.is_canceled()
     }
+    */
 
     pub(crate) fn is_closed(&self) -> bool {
         self.giver.is_canceled()
     }
 
+    #[cfg(test)]
     pub(crate) fn try_send(&mut self, val: T) -> Result<RetryPromise<T, U>, T> {
         let (tx, rx) = oneshot::channel();
         self.inner
             .send(Envelope(Some((val, Callback::Retry(tx)))))
             .map(move |_| rx)
             .map_err(|mut e| (e.0).0.take().expect("envelope not dropped").0)
     }
+
+    pub(crate) fn send(&mut self, val: T) -> Result<Promise<U>, T> {
+        let (tx, rx) = oneshot::channel();
+        self.inner
+            .send(Envelope(Some((val, Callback::NoRetry(tx)))))
+            .map(move |_| rx)
+            .map_err(|mut e| (e.0).0.take().expect("envelope not dropped").0)
+    }
 }
 
 #[cfg(feature = "http2")]
@@ -198,6 +214,7 @@ impl<T, U> Drop for Envelope<T, U> {
 }
 
 pub(crate) enum Callback<T, U> {
+    #[allow(unused)]
     Retry(oneshot::Sender<Result<U, (crate::Error, Option<T>)>>),
     NoRetry(oneshot::Sender<Result<U, crate::Error>>),
 }

src/client/dispatch.rs

@@ -1,31 +1,11 @@
 //! HTTP Client
 //!
-//! There are two levels of APIs provided for construct HTTP clients:
-//!
-//! - The higher-level [`Client`](Client) type.
-//! - The lower-level [`conn`](conn) module.
-//!
-//! # Client
-//!
-//! The [`Client`](Client) is the main way to send HTTP requests to a server.
-//! The default `Client` provides these things on top of the lower-level API:
-//!
-//! - A default **connector**, able to resolve hostnames and connect to
-//!   destinations over plain-text TCP.
-//! - A **pool** of existing connections, allowing better performance when
-//!   making multiple requests to the same hostname.
-//! - Automatic setting of the `Host` header, based on the request `Uri`.
-//! - Automatic request **retries** when a pooled connection is closed by the
-//!   server before any bytes have been written.
-//!
-//! Many of these features can configured, by making use of
-//! [`Client::builder`](Client::builder).
+//! hyper provides HTTP over a single connection. See the [`conn`](conn) module.
 //!
 //! ## Example
 //!
 //! For a small example program simply fetching a URL, take a look at the
 //! [full client example](https://github.com/hyperium/hyper/blob/master/examples/client.rs).
-//!
 
 pub mod connect;
 #[cfg(all(test, feature = "runtime"))]
@@ -34,10 +14,6 @@ mod tests;
 cfg_feature! {
     #![any(feature = "http1", feature = "http2")]
 
-    pub use self::client::{Builder, Client, ResponseFuture};
-
-    mod client;
     pub mod conn;
     pub(super) mod dispatch;
-    mod pool;
 }

src/client/mod.rs

@@ -13,16 +13,10 @@ pub(crate) mod date;
 #[cfg(any(feature = "http1", feature = "http2", feature = "server"))]
 pub(crate) mod exec;
 pub(crate) mod io;
-#[cfg(all(feature = "client", any(feature = "http1", feature = "http2")))]
-mod lazy;
 mod never;
-#[cfg(all(feature = "client", any(feature = "http1", feature = "http2")))]
-pub(crate) mod sync_wrapper;
 pub(crate) mod task;
 pub(crate) mod watch;
 
-#[cfg(all(feature = "client", any(feature = "http1", feature = "http2")))]
-pub(crate) use self::lazy::{lazy, Started as Lazy};
 #[cfg(any(feature = "http1", feature = "http2", feature = "runtime"))]
 pub(crate) use self::never::Never;
 pub(crate) use self::task::Poll;

src/client/pool.rs

@@ -34,9 +34,6 @@ pub(super) enum Kind {
     /// An `io::Error` that occurred while trying to read or write to a network stream.
     #[cfg(any(feature = "http1", feature = "http2"))]
     Io,
-    /// Error occurred while connecting.
-    #[allow(unused)]
-    Connect,
     /// Error creating a TcpListener.
     #[cfg(all(feature = "tcp", feature = "server"))]
     Listen,
@@ -101,22 +98,10 @@ pub(super) enum User {
     #[cfg(any(feature = "http1", feature = "http2"))]
     #[cfg(feature = "server")]
     UnexpectedHeader,
-    /// User tried to create a Request with bad version.
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    UnsupportedVersion,
-    /// User tried to create a CONNECT Request with the Client.
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    UnsupportedRequestMethod,
     /// User tried to respond with a 1xx (not 101) response code.
     #[cfg(feature = "http1")]
     #[cfg(feature = "server")]
     UnsupportedStatusCode,
-    /// User tried to send a Request with Client with non-absolute URI.
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    AbsoluteUriRequired,
 
     /// User tried polling for an upgrade that doesn't exist.
     NoUpgrade,
@@ -173,11 +158,6 @@ impl Error {
         matches!(self.inner.kind, Kind::ChannelClosed)
     }
 
-    /// Returns true if this was an error from `Connect`.
-    pub fn is_connect(&self) -> bool {
-        matches!(self.inner.kind, Kind::Connect)
-    }
-
     /// Returns true if the connection closed before a message could complete.
     pub fn is_incomplete_message(&self) -> bool {
         matches!(self.inner.kind, Kind::IncompleteMessage)
@@ -270,12 +250,6 @@ impl Error {
         Error::new(Kind::Listen).with(cause)
     }
 
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    pub(super) fn new_connect<E: Into<Cause>>(cause: E) -> Error {
-        Error::new(Kind::Connect).with(cause)
-    }
-
     pub(super) fn new_closed() -> Error {
         Error::new(Kind::ChannelClosed)
     }
@@ -309,30 +283,12 @@ impl Error {
         Error::new(Kind::HeaderTimeout)
     }
 
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    pub(super) fn new_user_unsupported_version() -> Error {
-        Error::new_user(User::UnsupportedVersion)
-    }
-
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    pub(super) fn new_user_unsupported_request_method() -> Error {
-        Error::new_user(User::UnsupportedRequestMethod)
-    }
-
     #[cfg(feature = "http1")]
     #[cfg(feature = "server")]
     pub(super) fn new_user_unsupported_status_code() -> Error {
         Error::new_user(User::UnsupportedStatusCode)
     }
 
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[cfg(feature = "client")]
-    pub(super) fn new_user_absolute_uri_required() -> Error {
-        Error::new_user(User::AbsoluteUriRequired)
-    }
-
     pub(super) fn new_user_no_upgrade() -> Error {
         Error::new_user(User::NoUpgrade)
     }
@@ -411,7 +367,6 @@ impl Error {
             #[cfg(feature = "http1")]
             Kind::UnexpectedMessage => "received unexpected message from connection",
             Kind::ChannelClosed => "channel closed",
-            Kind::Connect => "error trying to connect",
             Kind::Canceled => "operation was canceled",
             #[cfg(all(feature = "server", feature = "tcp"))]
             Kind::Listen => "error creating server listener",
@@ -436,20 +391,11 @@ impl Error {
             #[cfg(any(feature = "http1", feature = "http2"))]
             #[cfg(feature = "server")]
             Kind::User(User::UnexpectedHeader) => "user sent unexpected header",
-            #[cfg(any(feature = "http1", feature = "http2"))]
-            #[cfg(feature = "client")]
-            Kind::User(User::UnsupportedVersion) => "request has unsupported HTTP version",
-            #[cfg(any(feature = "http1", feature = "http2"))]
-            #[cfg(feature = "client")]
-            Kind::User(User::UnsupportedRequestMethod) => "request has unsupported HTTP method",
             #[cfg(feature = "http1")]
             #[cfg(feature = "server")]
             Kind::User(User::UnsupportedStatusCode) => {
                 "response has 1xx status code, not supported by server"
             }
-            #[cfg(any(feature = "http1", feature = "http2"))]
-            #[cfg(feature = "client")]
-            Kind::User(User::AbsoluteUriRequired) => "client requires absolute-form URIs",
             Kind::User(User::NoUpgrade) => "no upgrade available",
             #[cfg(feature = "http1")]
             Kind::User(User::ManualUpgrade) => "upgrade expected but low level API in use",

src/common/lazy.rs

@@ -93,9 +93,6 @@ cfg_feature! {
     #![feature = "client"]
 
     pub mod client;
-    #[cfg(any(feature = "http1", feature = "http2"))]
-    #[doc(no_inline)]
-    pub use crate::client::Client;
 }
 
 cfg_feature! {