dependency cleanup

Cargo.toml

@@ -1,17 +1,26 @@
 [workspace]
-members = ["arnold", "braid", "hyperdrive", "patron", "pidfile", "platter", "roomservice"]
+members = [
+    "arnold",
+    "braid",
+    "bridge",
+    "hyperdrive",
+    "patron",
+    "pidfile",
+    "platter",
+    "roomservice",
+]
 resolver = "2"
 
 [workspace.dependencies]
 arnold = { path = "./arnold" }
 braid = { path = "./braid" }
+bridge = { path = "./bridge" }
 patron = { path = "./patron" }
 pidfile = { path = "./pidfile" }
 
 bytes = "1"
-camino = { version = "1" }
+camino = { version = "1", default-features = false }
 dashmap = "5"
-futures = "0.3"
 futures-core = "0.3"
 futures-util = "0.3"
 http = { version = "1" }
@@ -23,15 +32,15 @@ hyper-util = { version = "0.1", features = ["client"] }
 ouroboros = "0.18"
 pem-rfc7468 = { version = "0.7", features = ["alloc"] }
 pin-project = { version = "1" }
-rustls = "^0.22"
+rustls = "^0.23"
 rustls-native-certs = "0.7.0"
 serde = { version = "1" }
 socket2 = "0.5"
 static-assertions = { version = "1", package = "static_assertions" }
 tempfile = "3"
 thiserror = { version = "1" }
 tokio = { version = "1" }
-tokio-rustls = "0.25"
+tokio-rustls = "0.26"
 tower = { version = "0.4" }
 tracing = { version = "^0.1" }
 tracing-test = { version = "*" }

braid/Cargo.toml

@@ -19,9 +19,6 @@ tokio-rustls.workspace = true
 tower.workspace = true
 tracing.workspace = true
 
-[dependencies.hyper-util]
-workspace = true
-features = ["client", "client-legacy", "tokio"]
 
 [dev-dependencies]
 futures-util.workspace = true

braid/src/tls/client/connector.rs

@@ -4,7 +4,6 @@ use std::sync::Arc;
 
 use futures_core::future::BoxFuture;
 use hyper::Uri;
-use hyper_util::client::legacy::connect::HttpConnector;
 use rustls::ClientConfig;
 use tokio::net::TcpStream;
 use tower::Service;
@@ -15,28 +14,24 @@ use super::TlsStream;
 ///
 /// This connector is a wrapper around `tokio_rustls::TlsConnector` and `hyper::client::HttpConnector`.
 #[derive(Clone, Debug)]
-pub struct TlsConnector {
-    http: HttpConnector,
+pub struct TlsConnector<P> {
+    protocol: P,
     tls: Arc<ClientConfig>,
 }
 
-impl TlsConnector {
+impl<P> TlsConnector<P> {
     /// Create a new connector with the given TLS configuration.
-    pub fn new(tls: Arc<ClientConfig>) -> Self {
-        Self {
-            http: HttpConnector::new(),
-            tls,
-        }
-    }
-
-    /// Set the http connector to use for TCP connections.
-    pub fn with_http(self, mut http: HttpConnector) -> Self {
-        http.enforce_http(false);
-        Self { http, ..self }
+    pub fn new(tls: Arc<ClientConfig>, protocol: P) -> Self {
+        Self { protocol, tls }
     }
 }
 
-impl Service<Uri> for TlsConnector {
+impl<P> Service<Uri> for TlsConnector<P>
+where
+    P: Service<Uri, Response = TcpStream> + Clone + Send + 'static,
+    P::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+    P::Future: Send + 'static,
+{
     type Response = TlsStream<TcpStream>;
 
     type Error = Box<dyn std::error::Error + Send + Sync>;
@@ -47,7 +42,7 @@ impl Service<Uri> for TlsConnector {
         &mut self,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Result<(), Self::Error>> {
-        self.http.poll_ready(cx).map_err(|err| err.into())
+        self.protocol.poll_ready(cx).map_err(|err| err.into())
     }
 
     fn call(&mut self, req: Uri) -> Self::Future {
@@ -57,12 +52,11 @@ impl Service<Uri> for TlsConnector {
             .and_then(|host| host.to_owned().try_into());
 
         let tls = self.tls.clone();
-        let conn = self.http.call(req);
+        let conn = self.protocol.call(req);
 
         let fut = async move {
-            let stream = conn.await?;
-            let connect =
-                tokio_rustls::TlsConnector::from(tls).connect(domain?, stream.into_inner());
+            let stream = conn.await.map_err(Into::into)?;
+            let connect = tokio_rustls::TlsConnector::from(tls).connect(domain?, stream);
             Ok::<_, Box<dyn std::error::Error + Send + Sync>>(TlsStream::from(connect))
         };
         Box::pin(fut)

bridge/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "bridge"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+pin-project.workspace = true
+hyper.workspace = true
+tokio.workspace = true

bridge/licenses/hyper-util.license

@@ -0,0 +1,19 @@
+Copyright (c) 2023 Sean McArthur
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

bridge/src/io.rs

@@ -0,0 +1,93 @@
+use std::{
+    io::Error,
+    ops::{Deref, DerefMut},
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use hyper::rt::Read;
+use hyper::rt::Write;
+
+#[derive(Debug)]
+#[pin_project::pin_project]
+pub struct TokioIo<T> {
+    #[pin]
+    inner: T,
+}
+
+impl<T> TokioIo<T> {
+    pub fn new(inner: T) -> Self {
+        Self { inner }
+    }
+}
+
+impl<T> Deref for TokioIo<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl<T> DerefMut for TokioIo<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
+impl<T> Read for TokioIo<T>
+where
+    T: tokio::io::AsyncRead,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        mut buf: hyper::rt::ReadBufCursor<'_>,
+    ) -> Poll<Result<(), Error>> {
+        let n = unsafe {
+            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
+            match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
+                Poll::Ready(Ok(())) => tbuf.filled().len(),
+                other => return other,
+            }
+        };
+
+        unsafe {
+            buf.advance(n);
+        }
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<T> Write for TokioIo<T>
+where
+    T: tokio::io::AsyncWrite,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, Error>> {
+        tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        tokio::io::AsyncWrite::is_write_vectored(&self.inner)
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[std::io::IoSlice<'_>],
+    ) -> Poll<Result<usize, Error>> {
+        tokio::io::AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
+    }
+}

bridge/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod io;
+pub mod rt;

bridge/src/rt.rs

@@ -0,0 +1,20 @@
+use hyper::rt::Executor;
+
+#[derive(Debug, Default, Clone, Copy)]
+pub struct TokioExecutor;
+
+impl TokioExecutor {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl<F> Executor<F> for TokioExecutor
+where
+    F: std::future::Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    fn execute(&self, future: F) {
+        tokio::spawn(future);
+    }
+}

patron/Cargo.toml

@@ -8,17 +8,17 @@ edition = "2021"
 [dependencies]
 arnold.workspace = true
 braid.workspace = true
+bridge.workspace = true
 futures-util.workspace = true
 http.workspace = true
-hyper-util.workspace = true
 hyper.workspace = true
 pin-project.workspace = true
 rustls-native-certs.workspace = true
 rustls.workspace = true
 socket2.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
-tower.workspace = true
+tower = { workspace = true, features = ["util"] }
 tracing.workspace = true
 
 [dev-dependencies]

patron/src/conn/http.rs

@@ -2,9 +2,10 @@
 
 use ::http::{Response, Version};
 use braid::client::Stream;
+use bridge::io::TokioIo;
+use bridge::rt::TokioExecutor;
 use futures_util::future::BoxFuture;
 use hyper::body::Incoming;
-use hyper_util::rt::{TokioExecutor, TokioIo};
 use std::fmt;
 use thiserror::Error;
 use tracing::trace;

patron/src/pool/mod.rs

@@ -26,7 +26,27 @@ use self::idle::IdleConnections;
 pub(crate) use self::key::Key;
 use self::weakopt::WeakOpt;
 
-/// A pool of connections to a database.
+/// A pool of connections to remote hosts.
+///
+/// This connection pool is specifically designed with HTTP connections in mind. It separates the treatment of the
+/// connection (e.g. HTTP/1.1, HTTP/2, etc) from the transport (e.g. TCP, TLS, etc). This allows the pool to be used
+/// with any type of connection, as long as it implements the `PoolableConnection` trait, and any type of transport,
+/// as long as it implements the `PoolableTransport` trait. This also allows the pool to be used with upgradeable
+/// connections, such as HTTP/1.1 connections that can be upgraded to HTTP/2, where the pool will have new HTTP/2
+/// connections wait for in-progress upgrades from HTTP/1.1 connections to complete and use those, rather than creating
+/// new connections.
+///
+/// The pool makes use of a `Checkout` to represent a connection that is being checked out of the pool. The `Checkout`
+/// type requires a `Connector` to be provided, which provides a future that will create a new connection to the remote
+/// host, and a future that will perform the handshake for the connection. The `Checkout` ensures that in-progress
+/// connection state is correctly managed, and that duplicate connections are not made unnecessarily.
+///
+/// The pool also provides a `Pooled` type, which is a wrapper around a connection that will return the connection to
+/// the pool when dropped, if the connection is still open and has not been marked as reusable (reusable connections
+/// are always kept in the pool - there is no need to return dropped copies).
+///
+/// Pool configuration happens in the `Config` type, which allows for setting the maximum idle duration of a connection,
+/// and the maximum number of idle connections per host.
 #[derive(Debug)]
 pub(crate) struct Pool<T: PoolableConnection> {
     inner: Arc<Mutex<PoolInner<T>>>,

patron/tests/client.rs

@@ -1,6 +1,6 @@
+use bridge::io::TokioIo;
 use futures_util::StreamExt;
 use http::StatusCode;
-use hyper_util::rt::TokioIo;
 use std::pin::pin;
 
 use patron::conn::duplex::DuplexTransport;

pidfile/Cargo.toml

@@ -9,7 +9,7 @@ edition = "2021"
 libc = "0.2"
 camino.workspace = true
 tracing.workspace = true
-tracing-test.workspace = true
 
 [dev-dependencies]
 tempfile.workspace = true
+tracing-test.workspace = true

platter/Cargo.toml

@@ -13,6 +13,7 @@ http-body-util.workspace = true
 hyper-util = { workspace = true, features = [
     "server",
     "server-auto",
+    "tokio",
     "service",
     "http1",
     "http2",