Add HTTP/2 multiplexed connection with concurrency control

## Summary
- Introduce Connection<C>, a Tower Service-based HTTP/2 connection that multiplexes requests over a single H2 session with semaphore-backed concurrency control
- Add TcpConnector service and ConnectionInfo/IntoConnectionInfo abstractions for URI-based TCP connection establishment

Author: muhamadazmy

Cargo.lock

@@ -28,23 +28,29 @@ bytestring = { workspace = true }
 futures = { workspace = true }
 h2 = "0.4.12"
 http = { workspace = true }
+http-body = { workspace = true }
 http-body-util = { workspace = true }
 http-serde = { workspace = true }
 hyper = { workspace = true, features = ["http1", "http2", "client"] }
 hyper-rustls = { workspace = true }
 hyper-util = { workspace = true, features = ["client-legacy"] }
 jsonwebtoken = { workspace = true }
 pem = { version = "3.0.6" }
+pin-project = { workspace = true }
 ring = { version = "0.17.14" }
 rustls = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_with = { workspace = true }
 thiserror = { workspace = true }
-tower-service = { version = "0.3" }
+tokio = { workspace = true }
+tokio-util = { workspace = true }
+tokio-rustls = "0.26"
+tower = { workspace = true }
 tracing = { workspace = true }
 zstd = { workspace = true }
 
 [dev-dependencies]
 googletest = { workspace = true }
-tempfile = { workspace = true }
+tempfile = { workspace = true }
+tokio-stream = {workspace = true}

crates/service-client/Cargo.toml

@@ -10,6 +10,7 @@
 
 use super::proxy::ProxyConnector;
 
+use crate::pool;
 use crate::utils::ErrorExt;
 
 use bytes::Bytes;
@@ -30,6 +31,7 @@ use std::fmt::Debug;
 use std::future::Future;
 use std::sync::{Arc, LazyLock};
 use std::{fmt, future};
+use tower::Layer;
 
 type ProxiedHttpsConnector = ProxyConnector<HttpsConnector<HttpConnector>>;
 
@@ -107,6 +109,11 @@ impl HttpClient {
             .enable_http2()
             .wrap_connector(http_connector.clone());
 
+        {
+            let _connector = pool::tls::TlsConnectorLayer::new(TLS_CLIENT_CONFIG.clone())
+                .layer(pool::TcpConnector);
+        }
+
         HttpClient {
             alpn_client: builder.clone().build::<_, BoxBody>(ProxyConnector::new(
                 options.http_proxy.clone(),

crates/service-client/src/http.rs

@@ -36,6 +36,7 @@ use std::sync::Arc;
 
 mod http;
 mod lambda;
+pub mod pool;
 mod proxy;
 mod request_identity;
 mod utils;