feat: unstable HttpClient Config

This adds an `unstable-config` feature, with a new `Config` struct, which can be used to configure any `HttpClient` which implements support for it.

Currently it supports two features - the most important and most generally supported:
- `timeout` (`Duration`)
- `no_delay` (`bool`)

Implementations are provided for async-h1, isahc, and hyper (partial, no `no_delay` support due to the tls connector).

No serious attempt has been made to add this to the wasm client at this point, since I don't understand well how to even build the wasm client or if it even works anymore with the state of rust wasm web build tools.

Author: Fishrock123

Cargo.toml

@@ -27,11 +27,13 @@ h1_client = ["async-h1", "async-std", "deadpool", "futures"]
 native_client = ["curl_client", "wasm_client"]
 curl_client = ["isahc", "async-std"]
 wasm_client = ["js-sys", "web-sys", "wasm-bindgen", "wasm-bindgen-futures", "futures"]
-hyper_client = ["hyper", "hyper-tls", "http-types/hyperium_http", "futures-util"]
+hyper_client = ["hyper", "hyper-tls", "http-types/hyperium_http", "futures-util", "tokio"]
 
 native-tls = ["async-native-tls"]
 rustls = ["async-tls"]
 
+unstable-config = []
+
 [dependencies]
 async-trait = "0.1.37"
 dashmap = "4.0.2"
@@ -53,6 +55,7 @@ async-tls = { version = "0.10.0", optional = true }
 hyper = { version = "0.13.6", features = ["tcp"], optional = true }
 hyper-tls = { version = "0.4.3", optional = true }
 futures-util = { version = "0.3.5", features = ["io"], optional = true }
+tokio = { version = "0.2", features = ["time"], optional = true }
 
 # curl_client
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]

src/config.rs

@@ -0,0 +1,47 @@
+//! Configuration for `HttpClient`s.
+
+use std::time::Duration;
+
+/// Configuration for `HttpClient`s.
+#[non_exhaustive]
+#[derive(Clone, Debug)]
+pub struct Config {
+    /// TCP `NO_DELAY`.
+    ///
+    /// Default: `false`.
+    pub no_delay: bool,
+    /// Connection timeout duration.
+    ///
+    /// Default: `Some(Duration::from_secs(60))`.
+    pub timeout: Option<Duration>,
+}
+
+impl Config {
+    /// Construct new empty config.
+    pub fn new() -> Self {
+        Self {
+            no_delay: false,
+            timeout: Some(Duration::from_secs(60)),
+        }
+    }
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Config {
+    /// Set TCP `NO_DELAY`.
+    pub fn set_no_delay(mut self, no_delay: bool) -> Self {
+        self.no_delay = no_delay;
+        self
+    }
+
+    /// Set connection timeout duration.
+    pub fn set_timeout(mut self, timeout: Option<Duration>) -> Self {
+        self.timeout = timeout;
+        self
+    }
+}

src/h1/mod.rs

@@ -1,5 +1,8 @@
 //! http-client implementation for async-h1, with connecton pooling ("Keep-Alive").
 
+#[cfg(feature = "unstable-config")]
+use std::convert::{Infallible, TryFrom};
+
 use std::fmt::Debug;
 use std::net::SocketAddr;
 
@@ -17,6 +20,8 @@ cfg_if::cfg_if! {
     }
 }
 
+use crate::Config;
+
 use super::{async_trait, Error, HttpClient, Request, Response};
 
 mod tcp;
@@ -40,6 +45,7 @@ pub struct H1Client {
     #[cfg(any(feature = "native-tls", feature = "rustls"))]
     https_pools: HttpsPool,
     max_concurrent_connections: usize,
+    config: Config,
 }
 
 impl Debug for H1Client {
@@ -75,6 +81,7 @@ impl Debug for H1Client {
                     .collect::<Vec<String>>(),
             )
             .field("https_pools", &https_pools)
+            .field("config", &self.config)
             .field(
                 "max_concurrent_connections",
                 &self.max_concurrent_connections,
@@ -97,6 +104,7 @@ impl H1Client {
             #[cfg(any(feature = "native-tls", feature = "rustls"))]
             https_pools: DashMap::new(),
             max_concurrent_connections: DEFAULT_MAX_CONCURRENT_CONNECTIONS,
+            config: Config::default(),
         }
     }
 
@@ -107,6 +115,7 @@ impl H1Client {
             #[cfg(any(feature = "native-tls", feature = "rustls"))]
             https_pools: DashMap::new(),
             max_concurrent_connections: max,
+            config: Config::default(),
         }
     }
 }
@@ -152,7 +161,7 @@ impl HttpClient for H1Client {
                     let pool_ref = if let Some(pool_ref) = self.http_pools.get(&addr) {
                         pool_ref
                     } else {
-                        let manager = TcpConnection::new(addr);
+                        let manager = TcpConnection::new(addr, self.config.clone());
                         let pool = Pool::<TcpStream, std::io::Error>::new(
                             manager,
                             self.max_concurrent_connections,
@@ -168,19 +177,28 @@ impl HttpClient for H1Client {
                     let stream = match pool.get().await {
                         Ok(s) => s,
                         Err(_) if has_another_addr => continue,
-                        Err(e) => return Err(Error::from_str(400, e.to_string()))?,
+                        Err(e) => return Err(Error::from_str(400, e.to_string())),
                     };
 
                     req.set_peer_addr(stream.peer_addr().ok());
                     req.set_local_addr(stream.local_addr().ok());
-                    return client::connect(TcpConnWrapper::new(stream), req).await;
+
+                    let tcp_conn = client::connect(TcpConnWrapper::new(stream), req);
+                    #[cfg(feature = "unstable-config")]
+                    return if let Some(timeout) = self.config.timeout {
+                        async_std::future::timeout(timeout, tcp_conn).await?
+                    } else {
+                        tcp_conn.await
+                    };
+                    #[cfg(not(feature = "unstable-config"))]
+                    return tcp_conn.await;
                 }
                 #[cfg(any(feature = "native-tls", feature = "rustls"))]
                 "https" => {
                     let pool_ref = if let Some(pool_ref) = self.https_pools.get(&addr) {
                         pool_ref
                     } else {
-                        let manager = TlsConnection::new(host.clone(), addr);
+                        let manager = TlsConnection::new(host.clone(), addr, self.config.clone());
                         let pool = Pool::<TlsStream<TcpStream>, Error>::new(
                             manager,
                             self.max_concurrent_connections,
@@ -196,13 +214,21 @@ impl HttpClient for H1Client {
                     let stream = match pool.get().await {
                         Ok(s) => s,
                         Err(_) if has_another_addr => continue,
-                        Err(e) => return Err(Error::from_str(400, e.to_string()))?,
+                        Err(e) => return Err(Error::from_str(400, e.to_string())),
                     };
 
                     req.set_peer_addr(stream.get_ref().peer_addr().ok());
                     req.set_local_addr(stream.get_ref().local_addr().ok());
 
-                    return client::connect(TlsConnWrapper::new(stream), req).await;
+                    let tls_conn = client::connect(TlsConnWrapper::new(stream), req);
+                    #[cfg(feature = "unstable-config")]
+                    return if let Some(timeout) = self.config.timeout {
+                        async_std::future::timeout(timeout, tls_conn).await?
+                    } else {
+                        tls_conn.await
+                    };
+                    #[cfg(not(feature = "unstable-config"))]
+                    return tls_conn.await;
                 }
                 _ => unreachable!(),
             }
@@ -213,6 +239,37 @@ impl HttpClient for H1Client {
             "missing valid address",
         ))
     }
+
+    #[cfg(feature = "unstable-config")]
+    /// Override the existing configuration with new configuration.
+    ///
+    /// Config options may not impact existing connections.
+    fn set_config(&mut self, config: Config) -> http_types::Result<()> {
+        self.config = config;
+
+        Ok(())
+    }
+
+    #[cfg(feature = "unstable-config")]
+    /// Get the current configuration.
+    fn config(&self) -> &Config {
+        &self.config
+    }
+}
+
+#[cfg(feature = "unstable-config")]
+impl TryFrom<Config> for H1Client {
+    type Error = Infallible;
+
+    fn try_from(config: Config) -> Result<Self, Self::Error> {
+        Ok(Self {
+            http_pools: DashMap::new(),
+            #[cfg(any(feature = "native-tls", feature = "rustls"))]
+            https_pools: DashMap::new(),
+            max_concurrent_connections: DEFAULT_MAX_CONCURRENT_CONNECTIONS,
+            config,
+        })
+    }
 }
 
 #[cfg(test)]

src/h1/tcp.rs

@@ -8,13 +8,17 @@ use deadpool::managed::{Manager, Object, RecycleResult};
 use futures::io::{AsyncRead, AsyncWrite};
 use futures::task::{Context, Poll};
 
+use crate::Config;
+
 #[derive(Clone, Debug)]
 pub(crate) struct TcpConnection {
     addr: SocketAddr,
+    config: Config,
 }
+
 impl TcpConnection {
-    pub(crate) fn new(addr: SocketAddr) -> Self {
-        Self { addr }
+    pub(crate) fn new(addr: SocketAddr, config: Config) -> Self {
+        Self { addr, config }
     }
 }
 
@@ -58,12 +62,21 @@ impl AsyncWrite for TcpConnWrapper {
 #[async_trait]
 impl Manager<TcpStream, std::io::Error> for TcpConnection {
     async fn create(&self) -> Result<TcpStream, std::io::Error> {
-        TcpStream::connect(self.addr).await
+        let tcp_stream = TcpStream::connect(self.addr).await?;
+
+        #[cfg(feature = "unstable-config")]
+        tcp_stream.set_nodelay(self.config.no_delay)?;
+
+        Ok(tcp_stream)
     }
 
     async fn recycle(&self, conn: &mut TcpStream) -> RecycleResult<std::io::Error> {
         let mut buf = [0; 4];
         let mut cx = Context::from_waker(futures::task::noop_waker_ref());
+
+        #[cfg(feature = "unstable-config")]
+        conn.set_nodelay(self.config.no_delay)?;
+
         match Pin::new(conn).poll_read(&mut cx, &mut buf) {
             Poll::Ready(Err(error)) => Err(error),
             Poll::Ready(Ok(bytes)) if bytes == 0 => Err(std::io::Error::new(

src/h1/tls.rs

@@ -16,16 +16,18 @@ cfg_if::cfg_if! {
     }
 }
 
-use crate::Error;
+use crate::{Config, Error};
 
 #[derive(Clone, Debug)]
 pub(crate) struct TlsConnection {
     host: String,
     addr: SocketAddr,
+    config: Config,
 }
+
 impl TlsConnection {
-    pub(crate) fn new(host: String, addr: SocketAddr) -> Self {
-        Self { host, addr }
+    pub(crate) fn new(host: String, addr: SocketAddr, config: Config) -> Self {
+        Self { host, addr, config }
     }
 }
 
@@ -70,13 +72,23 @@ impl AsyncWrite for TlsConnWrapper {
 impl Manager<TlsStream<TcpStream>, Error> for TlsConnection {
     async fn create(&self) -> Result<TlsStream<TcpStream>, Error> {
         let raw_stream = async_std::net::TcpStream::connect(self.addr).await?;
+
+        #[cfg(feature = "unstable-config")]
+        raw_stream.set_nodelay(self.config.no_delay)?;
+
         let tls_stream = add_tls(&self.host, raw_stream).await?;
         Ok(tls_stream)
     }
 
     async fn recycle(&self, conn: &mut TlsStream<TcpStream>) -> RecycleResult<Error> {
         let mut buf = [0; 4];
         let mut cx = Context::from_waker(futures::task::noop_waker_ref());
+
+        #[cfg(feature = "unstable-config")]
+        conn.get_ref()
+            .set_nodelay(self.config.no_delay)
+            .map_err(Error::from)?;
+
         match Pin::new(conn).poll_read(&mut cx, &mut buf) {
             Poll::Ready(Err(error)) => Err(error),
             Poll::Ready(Ok(bytes)) if bytes == 0 => Err(std::io::Error::new(
@@ -86,6 +98,7 @@ impl Manager<TlsStream<TcpStream>, Error> for TlsConnection {
             _ => Ok(()),
         }
         .map_err(Error::from)?;
+
         Ok(())
     }
 }