Introduce tower service into clients

Signed-off-by: Paul Scoropan <[email protected]>

Co-authored-by: Evaline Ju <[email protected]>

Cargo.toml

@@ -30,6 +30,7 @@ http-body-util = "0.1.2"
 http-serde = "2.1.1"
 hyper = { version = "1.4.1", features = ["http1", "http2", "server"] }
 hyper-rustls = { version = "0.27.3", features = ["ring"]}
+hyper-timeout = "0.5.2"
 hyper-util = { version = "0.1.7", features = ["server-auto", "server-graceful", "tokio"] }
 mime = "0.3.17"
 mio = "1.0.2"
@@ -52,14 +53,15 @@ tokio = { version = "1.39.2", features = ["rt", "rt-multi-thread", "parking_lot"
 tokio-rustls = { version = "0.26.0", features = ["ring"]}
 tokio-stream = { version = "0.1.15", features = ["sync"] }
 tonic = { version = "0.12.1", features = ["tls", "tls-roots", "tls-webpki-roots"] }
+tower = "0.5.1"
 tower-http = { version = "0.5.2", features = ["trace"] }
 tower-service = "0.3"
+tower-timeout = "0.3.0"
 tracing = "0.1.40"
 tracing-opentelemetry = "0.25.0"
 tracing-subscriber = { version = "0.3.18", features = ["json", "env-filter"] }
 url = "2.5.2"
 uuid = { version = "1.10.0", features = ["v4", "fast-rng"] }
-hyper-timeout = "0.5.2"
 
 [build-dependencies]
 tonic-build = "0.12.1"

src/clients.rs

@@ -30,6 +30,8 @@ use ginepro::LoadBalancedChannel;
 use hyper_timeout::TimeoutConnector;
 use hyper_util::rt::TokioExecutor;
 use tonic::{metadata::MetadataMap, Request};
+use tower::ServiceBuilder;
+use tower_timeout::TimeoutLayer;
 use tracing::{debug, instrument, Span};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 use url::Url;
@@ -244,7 +246,10 @@ pub async fn create_http_client(
 
     let client =
         hyper_util::client::legacy::Client::builder(TokioExecutor::new()).build(timeout_conn);
-    Ok(HttpClient::new(base_url, request_timeout, client))
+    let client = ServiceBuilder::new()
+        .layer(TimeoutLayer::new(request_timeout))
+        .service(client);
+    Ok(HttpClient::new(base_url, client))
 }
 
 #[instrument(skip_all, fields(hostname = service_config.hostname))]
@@ -306,6 +311,8 @@ pub async fn create_grpc_client<C>(
         .channel()
         .await
         .unwrap_or_else(|error| panic!("error creating grpc client: {error}"));
+
+    let channel = ServiceBuilder::new().service(channel);
     new(channel)
 }