feat(client): add 1.0 compatible client conn API (#3155)

This patch backports client/conn/http1 and http2 modules from 1.0 to
ease transition. It allows code still using 0.14.x to switch to the
per-version Connection types available in 1.0.

Closes #3053 

Co-authored-by: KOVACS Tamas <ktamas@fastmail.fm>

Author: seanmonstar

.github/workflows/CI.yml

@@ -62,11 +62,11 @@ jobs:
 
         include:
           - rust: stable
-            features: "--features full"
+            features: "--features full,backports"
           - rust: beta
-            features: "--features full"
+            features: "--features full,backports"
           - rust: nightly
-            features: "--features full,nightly"
+            features: "--features full,nightly,backports"
             benches: true
 
     runs-on: ${{ matrix.os }}

Cargo.toml

@@ -109,6 +109,9 @@ tcp = [
 # C-API support (currently unstable (no semver))
 ffi = ["libc"]
 
+# enable 1.0 backports
+backports = []
+
 # internal features used in CI
 nightly = []
 __internal_happy_eyeballs_tests = []

src/client/conn.rs

@@ -54,6 +54,11 @@
 //! # }
 //! ```
 
+#[cfg(all(feature = "backports", feature = "http1"))]
+pub mod http1;
+#[cfg(all(feature = "backports", feature = "http2"))]
+pub mod http2;
+
 use std::error::Error as StdError;
 use std::fmt;
 #[cfg(not(all(feature = "http1", feature = "http2")))]

src/client/conn/http1.rs

@@ -0,0 +1,427 @@
+//! HTTP/2 client connections
+
+use std::error::Error as StdError;
+use std::fmt;
+use std::marker::PhantomData;
+use std::sync::Arc;
+use std::time::Duration;
+
+use http::{Request, Response};
+use tokio::io::{AsyncRead, AsyncWrite};
+
+use super::super::dispatch;
+use crate::body::{HttpBody as Body, Body as IncomingBody};
+use crate::common::{
+    exec::{BoxSendFuture, Exec},
+    task, Future, Pin, Poll,
+};
+use crate::proto;
+use crate::rt::Executor;
+
+/// The sender side of an established connection.
+pub struct SendRequest<B> {
+    dispatch: dispatch::UnboundedSender<Request<B>, Response<IncomingBody>>,
+}
+
+impl<B> Clone for SendRequest<B> {
+    fn clone(&self) -> SendRequest<B> {
+        SendRequest { dispatch: self.dispatch.clone() }
+    }
+}
+
+/// 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
+/// can process incoming and outgoing messages, notice hangups, and the like.
+#[must_use = "futures do nothing unless polled"]
+pub struct Connection<T, B>
+where
+    T: AsyncRead + AsyncWrite + Send + 'static,
+    B: Body + 'static,
+{
+    inner: (PhantomData<T>, proto::h2::ClientTask<B>),
+}
+
+/// A builder to configure an HTTP connection.
+///
+/// After setting options, the builder is used to create a handshake future.
+#[derive(Clone, Debug)]
+pub struct Builder {
+    pub(super) exec: Exec,
+    h2_builder: proto::h2::client::Config,
+}
+
+/// Returns a handshake future over some IO.
+///
+/// This is a shortcut for `Builder::new().handshake(io)`.
+/// See [`client::conn`](crate::client::conn) for more.
+pub async fn handshake<E, T, B>(
+    exec: E,
+    io: T,
+) -> crate::Result<(SendRequest<B>, Connection<T, B>)>
+where
+    E: Executor<BoxSendFuture> + Send + Sync + 'static,
+    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    B: Body + 'static,
+    B::Data: Send,
+    B::Error: Into<Box<dyn StdError + Send + Sync>>,
+{
+    Builder::new(exec).handshake(io).await
+}
+
+// ===== impl SendRequest
+
+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<()>> {
+        if self.is_closed() {
+            Poll::Ready(Err(crate::Error::new_closed()))
+        } else {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    /// 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);
+        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()
+    }
+}
+
+impl<B> SendRequest<B>
+where
+    B: Body + 'static,
+{
+    /// Sends a `Request` on the associated connection.
+    ///
+    /// Returns a future that if successful, yields the `Response`.
+    ///
+    /// # Note
+    ///
+    /// There are some key differences in what automatic things the `Client`
+    /// does for you that will not be done here:
+    ///
+    /// - `Client` requires absolute-form `Uri`s, since the scheme and
+    ///   authority are needed to connect. They aren't required here.
+    /// - Since the `Client` requires absolute-form `Uri`s, it can add
+    ///   the `Host` header based on it. You must add a `Host` header yourself
+    ///   before calling this method.
+    /// - Since absolute-form `Uri`s are not required, if received, they will
+    ///   be serialized as-is.
+    pub fn send_request(
+        &mut self,
+        req: Request<B>,
+    ) -> impl Future<Output = crate::Result<Response<IncomingBody>>> {
+        let sent = self.dispatch.send(req);
+
+        async move {
+            match sent {
+                Ok(rx) => match rx.await {
+                    Ok(Ok(resp)) => Ok(resp),
+                    Ok(Err(err)) => Err(err),
+                    // this is definite bug if it happens, but it shouldn't happen!
+                    Err(_canceled) => panic!("dispatch dropped without returning error"),
+                },
+                Err(_req) => {
+                    tracing::debug!("connection was not ready");
+
+                    Err(crate::Error::new_canceled().with("connection was not ready"))
+                }
+            }
+        }
+    }
+
+    /*
+    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) => {
+                tracing::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> fmt::Debug for SendRequest<B> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("SendRequest").finish()
+    }
+}
+
+// ===== impl Connection
+
+impl<T, B> Connection<T, B>
+where
+    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    B: Body + Unpin + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<Box<dyn StdError + Send + Sync>>,
+{
+    /// Returns whether the [extended CONNECT protocol][1] is enabled or not.
+    ///
+    /// This setting is configured by the server peer by sending the
+    /// [`SETTINGS_ENABLE_CONNECT_PROTOCOL` parameter][2] in a `SETTINGS` frame.
+    /// This method returns the currently acknowledged value received from the
+    /// remote.
+    ///
+    /// [1]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
+    /// [2]: https://datatracker.ietf.org/doc/html/rfc8441#section-3
+    pub fn is_extended_connect_protocol_enabled(&self) -> bool {
+        self.inner.1.is_extended_connect_protocol_enabled()
+    }
+}
+
+impl<T, B> fmt::Debug for Connection<T, B>
+where
+    T: AsyncRead + AsyncWrite + fmt::Debug + Send + 'static,
+    B: Body + 'static,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Connection").finish()
+    }
+}
+
+impl<T, B> Future for Connection<T, B>
+where
+    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    B: Body + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<Box<dyn StdError + Send + Sync>>,
+{
+    type Output = crate::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+        match ready!(Pin::new(&mut self.inner.1).poll(cx))? {
+            proto::Dispatched::Shutdown => Poll::Ready(Ok(())),
+            #[cfg(feature = "http1")]
+            proto::Dispatched::Upgrade(_pending) => unreachable!("http2 cannot upgrade"),
+        }
+    }
+}
+
+// ===== impl Builder
+
+impl Builder {
+    /// Creates a new connection builder.
+    #[inline]
+    pub fn new<E>(exec: E) -> Builder 
+    where
+        E: Executor<BoxSendFuture> + Send + Sync + 'static,
+    {
+        use std::sync::Arc;
+        Builder {
+            exec: Exec::Executor(Arc::new(exec)),
+            h2_builder: Default::default(),
+        }
+    }
+
+    /// Provide an executor to execute background HTTP2 tasks.
+    pub fn executor<E>(&mut self, exec: E) -> &mut Builder
+    where
+        E: Executor<BoxSendFuture> + Send + Sync + 'static,
+    {
+        self.exec = Exec::Executor(Arc::new(exec));
+        self
+    }
+
+    /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
+    /// stream-level flow control.
+    ///
+    /// Passing `None` will do nothing.
+    ///
+    /// If not set, hyper will use a default.
+    ///
+    /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_INITIAL_WINDOW_SIZE
+    pub fn initial_stream_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
+        if let Some(sz) = sz.into() {
+            self.h2_builder.adaptive_window = false;
+            self.h2_builder.initial_stream_window_size = sz;
+        }
+        self
+    }
+
+    /// Sets the max connection-level flow control for HTTP2
+    ///
+    /// Passing `None` will do nothing.
+    ///
+    /// If not set, hyper will use a default.
+    pub fn initial_connection_window_size(
+        &mut self,
+        sz: impl Into<Option<u32>>,
+    ) -> &mut Self {
+        if let Some(sz) = sz.into() {
+            self.h2_builder.adaptive_window = false;
+            self.h2_builder.initial_conn_window_size = sz;
+        }
+        self
+    }
+
+    /// Sets whether to use an adaptive flow control.
+    ///
+    /// Enabling this will override the limits set in
+    /// `initial_stream_window_size` and
+    /// `initial_connection_window_size`.
+    pub fn adaptive_window(&mut self, enabled: bool) -> &mut Self {
+        use proto::h2::SPEC_WINDOW_SIZE;
+
+        self.h2_builder.adaptive_window = enabled;
+        if enabled {
+            self.h2_builder.initial_conn_window_size = SPEC_WINDOW_SIZE;
+            self.h2_builder.initial_stream_window_size = SPEC_WINDOW_SIZE;
+        }
+        self
+    }
+
+    /// Sets the maximum frame size to use for HTTP2.
+    ///
+    /// Passing `None` will do nothing.
+    ///
+    /// If not set, hyper will use a default.
+    pub fn max_frame_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
+        if let Some(sz) = sz.into() {
+            self.h2_builder.max_frame_size = sz;
+        }
+        self
+    }
+
+    /// Sets an interval for HTTP2 Ping frames should be sent to keep a
+    /// connection alive.
+    ///
+    /// Pass `None` to disable HTTP2 keep-alive.
+    ///
+    /// Default is currently disabled.
+    #[cfg(feature = "runtime")]
+    pub fn keep_alive_interval(
+        &mut self,
+        interval: impl Into<Option<Duration>>,
+    ) -> &mut Self {
+        self.h2_builder.keep_alive_interval = interval.into();
+        self
+    }
+
+    /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
+    ///
+    /// If the ping is not acknowledged within the timeout, the connection will
+    /// be closed. Does nothing if `keep_alive_interval` is disabled.
+    ///
+    /// Default is 20 seconds.
+    #[cfg(feature = "runtime")]
+    pub fn keep_alive_timeout(&mut self, timeout: Duration) -> &mut Self {
+        self.h2_builder.keep_alive_timeout = timeout;
+        self
+    }
+
+    /// Sets whether HTTP2 keep-alive should apply while the connection is idle.
+    ///
+    /// If disabled, keep-alive pings are only sent while there are open
+    /// request/responses streams. If enabled, pings are also sent when no
+    /// streams are active. Does nothing if `keep_alive_interval` is
+    /// disabled.
+    ///
+    /// Default is `false`.
+    #[cfg(feature = "runtime")]
+    pub fn keep_alive_while_idle(&mut self, enabled: bool) -> &mut Self {
+        self.h2_builder.keep_alive_while_idle = enabled;
+        self
+    }
+
+    /// Sets the maximum number of HTTP2 concurrent locally reset streams.
+    ///
+    /// See the documentation of [`h2::client::Builder::max_concurrent_reset_streams`] for more
+    /// details.
+    ///
+    /// The default value is determined by the `h2` crate.
+    ///
+    /// [`h2::client::Builder::max_concurrent_reset_streams`]: https://docs.rs/h2/client/struct.Builder.html#method.max_concurrent_reset_streams
+    pub fn max_concurrent_reset_streams(&mut self, max: usize) -> &mut Self {
+        self.h2_builder.max_concurrent_reset_streams = Some(max);
+        self
+    }
+
+    /// Set the maximum write buffer size for each HTTP/2 stream.
+    ///
+    /// Default is currently 1MB, but may change.
+    ///
+    /// # Panics
+    ///
+    /// 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);
+        self.h2_builder.max_send_buffer_size = max;
+        self
+    }
+
+    /// Constructs a connection with the configured options and IO.
+    /// See [`client::conn`](crate::client::conn) for more.
+    ///
+    /// Note, if [`Connection`] is not `await`-ed, [`SendRequest`] will
+    /// do nothing.
+    pub fn handshake<T, B>(
+        &self,
+        io: T,
+    ) -> impl Future<Output = crate::Result<(SendRequest<B>, Connection<T, B>)>>
+    where
+        T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+        B: Body + 'static,
+        B::Data: Send,
+        B::Error: Into<Box<dyn StdError + Send + Sync>>,
+    {
+        let opts = self.clone();
+
+        async move {
+            tracing::trace!("client handshake HTTP/1");
+
+            let (tx, rx) = dispatch::channel();
+            let h2 = proto::h2::client::handshake(io, rx, &opts.h2_builder, opts.exec)
+                .await?;
+            Ok((
+                SendRequest {
+                    dispatch: tx.unbound(),
+                },
+                Connection {
+                    inner: (PhantomData, h2),
+                },
+            ))
+        }
+    }
+}

src/client/conn/http2.rs

@@ -128,6 +128,15 @@ impl<T, U> UnboundedSender<T, U> {
             .map(move |_| rx)
             .map_err(|mut e| (e.0).0.take().expect("envelope not dropped").0)
     }
+
+    #[cfg(all(feature = "backports", feature = "http2"))]
+    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(Some(tx))))))
+            .map(move |_| rx)
+            .map_err(|mut e| (e.0).0.take().expect("envelope not dropped").0)
+    }
 }
 
 #[cfg(feature = "http2")]

src/client/dispatch.rs

@@ -2246,6 +2246,130 @@ mod conn {
         future::join(server, client).await;
     }
 
+    #[cfg(feature = "backports")]
+    mod backports {
+        use super::*;
+        #[tokio::test]
+        async fn get() {
+            let _ = ::pretty_env_logger::try_init();
+            let listener = TkTcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))
+                .await
+                .unwrap();
+            let addr = listener.local_addr().unwrap();
+
+            let server = async move {
+                let mut sock = listener.accept().await.unwrap().0;
+                let mut buf = [0; 4096];
+                let n = sock.read(&mut buf).await.expect("read 1");
+
+                // Notably:
+                // - Just a path, since just a path was set
+                // - No host, since no host was set
+                let expected = "GET /a HTTP/1.1\r\n\r\n";
+                assert_eq!(s(&buf[..n]), expected);
+
+                sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
+                    .await
+                    .unwrap();
+            };
+
+            let client = async move {
+                let tcp = tcp_connect(&addr).await.expect("connect");
+                let (mut client, conn) = conn::http1::handshake(tcp).await.expect("handshake");
+
+                tokio::task::spawn(async move {
+                    conn.await.expect("http conn");
+                });
+
+                let req: Request<Body> = Request::builder()
+                    .uri("/a")
+                    .body(Default::default())
+                    .unwrap();
+                let mut res = client.send_request(req).await.expect("send_request");
+                assert_eq!(res.status(), hyper::StatusCode::OK);
+                assert!(res.body_mut().next().await.is_none());
+            };
+
+            future::join(server, client).await;
+        }
+
+        #[tokio::test]
+        async fn http2_detect_conn_eof() {
+            use futures_util::future;
+            use hyper::service::{make_service_fn, service_fn};
+            use hyper::{Response, Server};
+
+            let _ = pretty_env_logger::try_init();
+
+            let server = Server::bind(&([127, 0, 0, 1], 0).into())
+                .http2_only(true)
+                .serve(make_service_fn(|_| async move {
+                    Ok::<_, hyper::Error>(service_fn(|_req| {
+                        future::ok::<_, hyper::Error>(Response::new(Body::empty()))
+                    }))
+                }));
+            let addr = server.local_addr();
+            let (shdn_tx, shdn_rx) = oneshot::channel();
+            tokio::task::spawn(async move {
+                server
+                    .with_graceful_shutdown(async move {
+                        let _ = shdn_rx.await;
+                    })
+                    .await
+                    .expect("server")
+            });
+
+            struct TokioExec;
+            impl<F> hyper::rt::Executor<F> for TokioExec
+            where
+                F: std::future::Future + Send + 'static,
+                F::Output: Send + 'static,
+            {
+                fn execute(&self, fut: F) {
+                    tokio::spawn(fut);
+                }
+            }
+
+            let io = tcp_connect(&addr).await.expect("tcp connect");
+            let (mut client, conn) = conn::http2::Builder::new(TokioExec)
+                .handshake::<_, Body>(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 req = Request::builder()
+                .uri(format!("http://{}/", addr))
+                .body(Body::empty())
+                .expect("request builder");
+
+            client.send_request(req).await.expect("req1 send");
+
+            // Sanity check that client is STILL ready
+            future::poll_fn(|ctx| client.poll_ready(ctx))
+                .await
+                .expect("client poll ready after");
+
+            // Trigger the server shutdown...
+            let _ = shdn_tx.send(());
+
+            // Allow time for graceful shutdown roundtrips...
+            tokio::time::sleep(Duration::from_millis(100)).await;
+
+            // After graceful shutdown roundtrips, the client should be closed...
+            future::poll_fn(|ctx| client.poll_ready(ctx))
+                .await
+                .expect_err("client should be closed");
+        }
+    }
+
     #[tokio::test]
     async fn get_obsolete_line_folding() {
         let _ = ::pretty_env_logger::try_init();