apollo_storage,apollo_rpc: migrate hyper to 1.x PART 10

Author: victorkstarkware

Cargo.lock

@@ -219,6 +219,8 @@ async-stream = "0.3.3"
 async-trait = "0.1.79"
 asynchronous-codec = "0.7.0"
 axum_08 = { package = "axum", version = "0.8" }
+# TODO(victork): finalise migration to hyper 1.x
+tower_05 = { package = "tower", version = "0.5" }
 base64 = "0.13.0"
 bench_tools.path = "crates/bench_tools"
 bitvec = "1.0.1"

Cargo.toml

@@ -30,8 +30,6 @@ serde_json = { workspace = true, features = ["arbitrary_precision"] }
 starknet-types-core.workspace = true
 starknet_api.workspace = true
 tokio = { workspace = true, features = ["full", "sync"] }
-# tower 0.4 is required for jsonrpsee 0.24 http middleware compatibility (jsonrpsee expects tower 0.4 types)
-tower = { version = "0.4.13", features = ["full"] }
 tower_05 = { workspace = true, features = ["full"] }
 tracing.workspace = true
 validator = { workspace = true, features = ["derive"] }

crates/apollo_rpc/Cargo.toml

@@ -33,7 +33,7 @@ use apollo_storage::state::StateStorageReader;
 use apollo_storage::{StorageReader, StorageScope, StorageTxn};
 use jsonrpsee::core::RpcResult;
 use jsonrpsee::server::middleware::rpc::RpcServiceBuilder;
-use jsonrpsee::server::{Server, ServerHandle};
+use jsonrpsee::server::{Server, ServerConfig, ServerHandle};
 use jsonrpsee::types::error::ErrorCode::InternalError;
 use jsonrpsee::types::error::INTERNAL_ERROR_MSG;
 use jsonrpsee::types::ErrorObjectOwned;
@@ -244,9 +244,12 @@ pub async fn run_server(
     );
     let addr;
     let handle;
-    let server_builder = Server::builder()
+    let server_config = ServerConfig::builder()
         .max_request_body_size(SERVER_MAX_BODY_SIZE)
-        .set_http_middleware(tower::ServiceBuilder::new().filter_async(proxy_rpc_request));
+        .build();
+    let server_builder = Server::builder()
+        .set_config(server_config)
+        .set_http_middleware(tower_05::ServiceBuilder::new().filter_async(proxy_rpc_request));
 
     let server_address = SocketAddr::new(config.ip, config.port);
 

crates/apollo_rpc/src/lib.rs

@@ -1,7 +1,7 @@
 use jsonrpsee::core::http_helpers::read_body;
 use jsonrpsee::server::HttpRequest;
 use regex::Regex;
-use tower::BoxError;
+use tower_05::BoxError;
 use tracing::debug;
 
 use crate::version_config::{VERSION_0_8, VERSION_PATTERN};

crates/apollo_rpc/src/middleware.rs

@@ -2,12 +2,9 @@
 mod rpc_metrics_test;
 
 use std::collections::HashSet;
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
 use std::time::Instant;
 
-use jsonrpsee::server::middleware::rpc::RpcServiceT;
+use jsonrpsee::server::middleware::rpc::{Batch, Notification, RpcServiceT};
 use jsonrpsee::server::MethodResponse;
 use jsonrpsee::types::Request;
 use jsonrpsee::Methods;
@@ -83,42 +80,48 @@ pub(crate) struct MetricLoggerService<S> {
     logger: MetricLogger,
 }
 
-impl<'a, S> RpcServiceT<'a> for MetricLoggerService<S>
+impl<S> RpcServiceT for MetricLoggerService<S>
 where
-    S: RpcServiceT<'a> + Send + Sync + Clone + 'static,
+    S: RpcServiceT<MethodResponse = MethodResponse, BatchResponse = MethodResponse, NotificationResponse = MethodResponse>
+        + Send
+        + Sync
+        + Clone
+        + 'static,
 {
-    type Future = MetricResponseFuture<S::Future>;
-
-    fn call(&self, request: Request<'a>) -> Self::Future {
+    type MethodResponse = MethodResponse;
+    type BatchResponse = MethodResponse;
+    type NotificationResponse = MethodResponse;
+
+    fn call<'a>(
+        &self,
+        request: Request<'a>,
+    ) -> impl std::future::Future<Output = Self::MethodResponse> + Send + 'a {
         let method_name = request.method_name().to_string();
-        MetricResponseFuture {
-            fut: Box::pin(self.service.call(request)),
-            method_name,
-            logger: self.logger.clone(),
-            started_at: Instant::now(),
+        let logger = self.logger.clone();
+        let service = self.service.clone();
+        let started_at = Instant::now();
+
+        async move {
+            let response = service.call(request).await;
+            logger.on_result(&method_name, response.is_success(), started_at);
+            response
         }
     }
-}
-
-/// Response future that records metrics when the response is ready.
-pub(crate) struct MetricResponseFuture<F> {
-    fut: Pin<Box<F>>,
-    method_name: String,
-    logger: MetricLogger,
-    started_at: Instant,
-}
 
-impl<F: Future<Output = MethodResponse>> Future for MetricResponseFuture<F> {
-    type Output = MethodResponse;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let res = self.fut.as_mut().poll(cx);
-
-        if let Poll::Ready(ref response) = res {
-            self.logger.on_result(&self.method_name, response.is_success(), self.started_at);
-        }
+    fn batch<'a>(
+        &self,
+        batch: Batch<'a>,
+    ) -> impl std::future::Future<Output = Self::BatchResponse> + Send + 'a {
+        let service = self.service.clone();
+        async move { service.batch(batch).await }
+    }
 
-        res
+    fn notification<'a>(
+        &self,
+        n: Notification<'a>,
+    ) -> impl std::future::Future<Output = Self::NotificationResponse> + Send + 'a {
+        let service = self.service.clone();
+        async move { service.notification(n).await }
     }
 }