pools wip

Author: seanmonstar

Cargo.toml

@@ -43,6 +43,7 @@ futures-util = { version = "0.3.16", default-features = false, features = ["allo
 http-body-util = "0.1.0"
 tokio = { version = "1", features = ["macros", "test-util", "signal"] }
 tokio-test = "0.4"
+tower = { version = "0.5", features = ["util"] }
 tower-test = "0.4"
 pretty_env_logger = "0.5"
 
@@ -56,7 +57,7 @@ system-configuration = { version = "0.6.1", optional = true }
 windows-registry = { version = "0.5", optional = true }
 
 [features]
-default = []
+default = ["client", "client-pool"]
 
 # Shorthand to enable everything
 full = [
@@ -80,6 +81,7 @@ client-legacy = ["client", "dep:socket2", "tokio/sync", "dep:libc", "dep:futures
 client-pool = ["dep:futures-util", "dep:tower-layer"]
 client-proxy = ["client", "dep:base64", "dep:ipnet", "dep:percent-encoding"]
 client-proxy-system = ["dep:system-configuration", "dep:windows-registry"]
+client-pool = ["tokio/sync", "dep:futures-util", "dep:tower-layer"]
 
 server = ["hyper/server"]
 server-auto = ["server", "http1", "http2"]
@@ -99,6 +101,10 @@ __internal_happy_eyeballs_tests = []
 
 [[example]]
 name = "client"
+required-features = ["client-legacy", "client-pool", "http1", "tokio"]
+
+[[example]]
+name = "client_legacy"
 required-features = ["client-legacy", "http1", "tokio"]
 
 [[example]]

examples/client.rs

@@ -1,37 +1,143 @@
-use std::env;
+use tower_service::Service;
+use tower::ServiceExt;
 
-use http_body_util::Empty;
-use hyper::Request;
-use hyper_util::client::legacy::{connect::HttpConnector, Client};
+use hyper_util::client::pool;
 
 #[tokio::main(flavor = "current_thread")]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let url = match env::args().nth(1) {
-        Some(url) => url,
-        None => {
-            eprintln!("Usage: client <url>");
-            return Ok(());
-        }
-    };
-
-    // HTTPS requires picking a TLS implementation, so give a better
-    // warning if the user tries to request an 'https' URL.
-    let url = url.parse::<hyper::Uri>()?;
-    if url.scheme_str() != Some("http") {
-        eprintln!("This example only works with 'http' URLs.");
-        return Ok(());
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    send_nego().await
+}
+
+async fn send_h1() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let tcp = hyper_util::client::legacy::connect::HttpConnector::new();
+
+    let http1 = tcp.and_then(|conn| Box::pin(async move {
+        let (mut tx, c) = hyper::client::conn::http1::handshake::<_, http_body_util::Empty<hyper::body::Bytes>>(conn).await?;
+        tokio::spawn(async move {
+            if let Err(e) = c.await {
+                eprintln!("connection error: {:?}", e);
+            }
+        });
+        let svc = tower::service_fn(move |req| { tx.send_request(req) });
+        Ok::<_, Box<dyn std::error::Error + Send + Sync>>(svc)
+    }));
+
+    let mut p = pool::Cache::new(http1).build();
+
+    let mut c = p.call(http::Uri::from_static("http://hyper.rs")).await?;
+    eprintln!("{:?}", c);
+
+    let req = http::Request::builder()
+        .header("host", "hyper.rs")
+        .body(http_body_util::Empty::new())
+        .unwrap();
+
+    c.ready().await?;
+    let resp = c.call(req).await?;
+    eprintln!("{:?}", resp);
+
+    Ok(())
+}
+
+async fn send_h2() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let tcp = hyper_util::client::legacy::connect::HttpConnector::new();
+
+    let http2 = tcp.and_then(|conn| Box::pin(async move {
+        let (mut tx, c) = hyper::client::conn::http2::handshake::<_, _, http_body_util::Empty<hyper::body::Bytes>>(
+            hyper_util::rt::TokioExecutor::new(),
+            conn,
+        ).await?;
+        println!("connected");
+        tokio::spawn(async move {
+            if let Err(e) = c.await {
+                eprintln!("connection error: {:?}", e);
+            }
+        });
+        let svc = tower::service_fn(move |req| { tx.send_request(req) });
+        Ok::<_, Box<dyn std::error::Error + Send + Sync>>(svc)
+    }));
+
+    let mut p = pool::Singleton::new(http2);
+
+    for _ in 0..5 {
+        let mut c = p.call(http::Uri::from_static("http://localhost:3000")).await?;
+        eprintln!("{:?}", c);
+
+        let req = http::Request::builder()
+            .header("host", "hyper.rs")
+            .body(http_body_util::Empty::new())
+            .unwrap();
+
+        c.ready().await?;
+        let resp = c.call(req).await?;
+        eprintln!("{:?}", resp);
     }
 
-    let client = Client::builder(hyper_util::rt::TokioExecutor::new()).build(HttpConnector::new());
+    Ok(())
+}
+
+async fn send_nego() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let tcp = hyper_util::client::legacy::connect::HttpConnector::new();
+
+    let http1 = tower::layer::layer_fn(|tcp| {
+        tower::service_fn(move |dst| {
+            let inner = tcp.call(dst);
+            async move {
+                let conn = inner.await?;
+                let (mut tx, c) = hyper::client::conn::http1::handshake::<_, http_body_util::Empty<hyper::body::Bytes>>(conn).await?;
+                tokio::spawn(async move {
+                    if let Err(e) = c.await {
+                        eprintln!("connection error: {:?}", e);
+                    }
+                });
+                let svc = tower::service_fn(move |req| { tx.send_request(req) });
+                Ok::<_, Box<dyn std::error::Error + Send + Sync>>(svc)
+            }
+        })
+    });
+
 
-    let req = Request::builder()
-        .uri(url)
-        .body(Empty::<bytes::Bytes>::new())?;
+    let http2 = tower::layer::layer_fn(|tcp| {
+        tower::service_fn(move |dst| {
+            let inner = tcp.call(dst);
+            async move {
+                let conn = inner.await?;
+                let (mut tx, c) = hyper::client::conn::http2::handshake::<_, _, http_body_util::Empty<hyper::body::Bytes>>(
+                    hyper_util::rt::TokioExecutor::new(),
+                    conn,
+                ).await?;
+                println!("connected");
+                tokio::spawn(async move {
+                    if let Err(e) = c.await {
+                        eprintln!("connection error: {:?}", e);
+                    }
+                });
+                let svc = tower::service_fn(move |req| { tx.send_request(req) });
+                Ok::<_, Box<dyn std::error::Error + Send + Sync>>(svc)
+            }
+        })
+    });
 
-    let resp = client.request(req).await?;
+    let mut svc = pool::negotiate(
+        tcp,
+        |_| false,
+        http1,
+        http2,
+    );
 
-    eprintln!("{:?} {:?}", resp.version(), resp.status());
-    eprintln!("{:#?}", resp.headers());
+    for _ in 0..5 {
+        let mut c = svc.call(http::Uri::from_static("http://localhost:3000")).await?;
+        eprintln!("{:?}", c);
+
+        let req = http::Request::builder()
+            .header("host", "hyper.rs")
+            .body(http_body_util::Empty::new())
+            .unwrap();
+
+        c.ready().await?;
+        let resp = c.call(req).await?;
+        eprintln!("{:?}", resp);
+    }
 
     Ok(())
 }

examples/client_legacy.rs

@@ -0,0 +1,37 @@
+use std::env;
+
+use http_body_util::Empty;
+use hyper::Request;
+use hyper_util::client::legacy::{connect::HttpConnector, Client};
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let url = match env::args().nth(1) {
+        Some(url) => url,
+        None => {
+            eprintln!("Usage: client <url>");
+            return Ok(());
+        }
+    };
+
+    // HTTPS requires picking a TLS implementation, so give a better
+    // warning if the user tries to request an 'https' URL.
+    let url = url.parse::<hyper::Uri>()?;
+    if url.scheme_str() != Some("http") {
+        eprintln!("This example only works with 'http' URLs.");
+        return Ok(());
+    }
+
+    let client = Client::builder(hyper_util::rt::TokioExecutor::new()).build(HttpConnector::new());
+
+    let req = Request::builder()
+        .uri(url)
+        .body(Empty::<bytes::Bytes>::new())?;
+
+    let resp = client.request(req).await?;
+
+    eprintln!("{:?} {:?}", resp.version(), resp.status());
+    eprintln!("{:#?}", resp.headers());
+
+    Ok(())
+}

src/client/mod.rs

@@ -9,3 +9,5 @@ pub mod pool;
 
 #[cfg(feature = "client-proxy")]
 pub mod proxy;
+
+pub mod pool;

src/client/pool/expire.rs

@@ -0,0 +1,3 @@
+pub struct Expire {
+
+}