Address review

Author: arjan-bal

grpc/Cargo.toml

@@ -6,11 +6,7 @@ authors = ["gRPC Authors"]
 license = "MIT"
 
 [package.metadata.cargo_check_external_types]
-allowed_external_types = [
-    "tonic::*",
-    "futures_core::stream::Stream",
-    "tokio::sync::oneshot::Sender",
-]
+allowed_external_types = ["tonic::*", "tokio::sync::oneshot::Sender"]
 
 [features]
 default = ["dns", "_runtime-tokio"]
@@ -22,16 +18,13 @@ dns = ["dep:hickory-resolver", "_runtime-tokio"]
 _runtime-tokio = [
     "tokio/rt",
     "tokio/net",
-    "dep:futures",
+    "tokio/time",
     "dep:socket2",
     "dep:tower",
 ]
 
 [dependencies]
 bytes = "1.10.1"
-futures = { version = "0.3.31", optional = true }
-futures-core = "0.3.31"
-futures-util = "0.3.31"
 hickory-resolver = { version = "0.25.1", optional = true }
 http = "1.1.0"
 http-body = "1.0.1"
@@ -43,8 +36,8 @@ rand = "0.9"
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.140"
 socket2 = { version = "0.5.10", optional = true }
-tokio = { version = "1.37.0", features = ["sync", "time", "macros"] }
-tokio-stream = "0.1.17"
+tokio = { version = "1.37.0", features = ["sync", "macros"] }
+tokio-stream = { version = "0.1.17", default-features = false }
 tonic = { version = "0.14.0", path = "../tonic", default-features = false, features = [
     "codegen",
 ] }

grpc/examples/inmemory.rs

@@ -1,8 +1,8 @@
 use std::any::Any;
 
-use futures_util::stream::StreamExt;
 use grpc::service::{Message, Request, Response, Service};
 use grpc::{client::ChannelOptions, inmemory};
+use tokio_stream::StreamExt;
 use tonic::async_trait;
 
 struct Handler {}

grpc/examples/multiaddr.rs

@@ -1,8 +1,8 @@
 use std::any::Any;
 
-use futures_util::StreamExt;
 use grpc::service::{Message, Request, Response, Service};
 use grpc::{client::ChannelOptions, inmemory};
+use tokio_stream::StreamExt;
 use tonic::async_trait;
 
 struct Handler {

grpc/src/client/load_balancing/pick_first.rs

@@ -4,7 +4,6 @@ use std::{
     time::Duration,
 };
 
-use tokio::time::sleep;
 use tonic::metadata::MetadataMap;
 
 use crate::{
@@ -75,9 +74,10 @@ impl LbPolicy for PickFirstPolicy {
 
         self.next_addresses = addresses;
         let work_scheduler = self.work_scheduler.clone();
+        let runtime = self.runtime.clone();
         // TODO: Implement Drop that cancels this task.
         self.runtime.spawn(Box::pin(async move {
-            sleep(Duration::from_millis(200)).await;
+            runtime.sleep(Duration::from_millis(200)).await;
             work_scheduler.schedule_work();
         }));
         // TODO: return a picker that queues RPCs.

grpc/src/client/subchannel.rs

@@ -15,17 +15,15 @@ use crate::{
     service::{Request, Response, Service},
 };
 use core::panic;
+use std::time::{Duration, Instant};
 use std::{
     collections::BTreeMap,
     error::Error,
     fmt::{Debug, Display},
     ops::Sub,
     sync::{Arc, Mutex, RwLock, Weak},
 };
-use tokio::{
-    sync::{mpsc, oneshot, watch, Notify},
-    time::{Duration, Instant},
-};
+use tokio::sync::{mpsc, oneshot, watch, Notify};
 use tonic::async_trait;
 
 type SharedService = Arc<dyn Service>;
@@ -358,7 +356,7 @@ impl InternalSubchannel {
 
         let connect_task = self.runtime.spawn(Box::pin(async move {
             tokio::select! {
-                _ = tokio::time::sleep(min_connect_timeout) => {
+                _ = runtime.sleep(min_connect_timeout) => {
                     let _ = state_machine_tx.send(SubchannelStateMachineEvent::ConnectionTimedOut);
                 }
                 result = transport.connect(address.to_string().clone(), runtime, &transport_opts) => {
@@ -400,7 +398,7 @@ impl InternalSubchannel {
             // terminated? But what can we do with that error other than logging
             // it, which the transport can do as well?
             if let Err(e) = closed_rx.await {
-                eprintln!("Transport closed with error: {}", e.to_string())
+                eprintln!("Transport closed with error: {e}",)
             };
             let _ = state_machine_tx.send(SubchannelStateMachineEvent::ConnectionTerminated);
         }));
@@ -430,8 +428,11 @@ impl InternalSubchannel {
 
         let backoff_interval = self.backoff.backoff_until();
         let state_machine_tx = self.state_machine_event_sender.clone();
+        let runtime = self.runtime.clone();
         let backoff_task = self.runtime.spawn(Box::pin(async move {
-            tokio::time::sleep_until(backoff_interval).await;
+            runtime
+                .sleep(backoff_interval.saturating_duration_since(Instant::now()))
+                .await;
             let _ = state_machine_tx.send(SubchannelStateMachineEvent::BackoffExpired);
         }));
         let mut inner = self.inner.lock().unwrap();

grpc/src/client/transport/mod.rs

@@ -24,18 +24,18 @@ pub(crate) struct ConnectedTransport {
 // can hold config relevant to a particular transport.
 #[derive(Default)]
 pub(crate) struct TransportOptions {
-    pub init_stream_window_size: Option<u32>,
-    pub init_connection_window_size: Option<u32>,
-    pub http2_keep_alive_interval: Option<Duration>,
-    pub http2_keep_alive_timeout: Option<Duration>,
-    pub http2_keep_alive_while_idle: Option<bool>,
-    pub http2_max_header_list_size: Option<u32>,
-    pub http2_adaptive_window: Option<bool>,
-    pub concurrency_limit: Option<usize>,
-    pub rate_limit: Option<(u64, Duration)>,
-    pub tcp_keepalive: Option<Duration>,
-    pub tcp_nodelay: bool,
-    pub connect_deadline: Option<Instant>,
+    pub(crate) init_stream_window_size: Option<u32>,
+    pub(crate) init_connection_window_size: Option<u32>,
+    pub(crate) http2_keep_alive_interval: Option<Duration>,
+    pub(crate) http2_keep_alive_timeout: Option<Duration>,
+    pub(crate) http2_keep_alive_while_idle: Option<bool>,
+    pub(crate) http2_max_header_list_size: Option<u32>,
+    pub(crate) http2_adaptive_window: Option<bool>,
+    pub(crate) concurrency_limit: Option<usize>,
+    pub(crate) rate_limit: Option<(u64, Duration)>,
+    pub(crate) tcp_keepalive: Option<Duration>,
+    pub(crate) tcp_nodelay: bool,
+    pub(crate) connect_deadline: Option<Instant>,
 }
 
 #[async_trait]

grpc/src/client/transport/registry.rs

@@ -10,12 +10,12 @@ use std::{
 /// the address type they are intended to handle.
 #[derive(Default, Clone)]
 pub(crate) struct TransportRegistry {
-    m: Arc<Mutex<HashMap<String, Arc<dyn Transport>>>>,
+    inner: Arc<Mutex<HashMap<String, Arc<dyn Transport>>>>,
 }
 
 impl Debug for TransportRegistry {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let m = self.m.lock().unwrap();
+        let m = self.inner.lock().unwrap();
         for key in m.keys() {
             write!(f, "k: {key:?}")?
         }
@@ -31,15 +31,15 @@ impl TransportRegistry {
 
     /// Add a transport into the registry.
     pub(crate) fn add_transport(&self, address_type: &str, transport: impl Transport + 'static) {
-        self.m
+        self.inner
             .lock()
             .unwrap()
             .insert(address_type.to_string(), Arc::new(transport));
     }
 
     /// Retrieve a name resolver from the registry, or None if not found.
     pub(crate) fn get_transport(&self, address_type: &str) -> Result<Arc<dyn Transport>, String> {
-        self.m
+        self.inner
             .lock()
             .unwrap()
             .get(address_type)

grpc/src/client/transport/tonic/mod.rs

@@ -8,8 +8,6 @@ use crate::service::Message;
 use crate::service::Request as GrpcRequest;
 use crate::service::Response as GrpcResponse;
 use bytes::Bytes;
-use futures::stream::StreamExt;
-use futures::Stream;
 use http::uri::PathAndQuery;
 use http::Request as HttpRequest;
 use http::Response as HttpResponse;
@@ -27,6 +25,8 @@ use std::{
     sync::Arc,
     task::{Context, Poll},
 };
+use tokio_stream::Stream;
+use tokio_stream::StreamExt;
 use tonic::Request as TonicRequest;
 use tonic::Response as TonicResponse;
 use tonic::Streaming;
@@ -95,14 +95,14 @@ impl Service for TonicTransport {
 
 /// Helper function to create an error response stream.
 fn create_error_response(status: Status) -> GrpcResponse {
-    let stream = futures::stream::once(async { Err(status) });
+    let stream = tokio_stream::once(Err(status));
     TonicResponse::new(Box::pin(stream))
 }
 
 fn convert_request(req: GrpcRequest) -> TonicRequest<Pin<Box<dyn Stream<Item = Bytes> + Send>>> {
     let (metadata, extensions, stream) = req.into_parts();
 
-    let bytes_stream = Box::pin(stream.filter_map(|msg| async {
+    let bytes_stream = Box::pin(stream.filter_map(|msg| {
         if let Ok(bytes) = (msg as Box<dyn Any>).downcast::<Bytes>() {
             Some(*bytes)
         } else {
@@ -119,7 +119,7 @@ fn convert_response(res: Result<TonicResponse<Streaming<Bytes>>, Status>) -> Grp
     let response = match res {
         Ok(s) => s,
         Err(e) => {
-            let stream = futures::stream::once(async { Err(e) });
+            let stream = tokio_stream::once(Err(e));
             return TonicResponse::new(Box::pin(stream));
         }
     };

grpc/src/inmemory/mod.rs

@@ -128,6 +128,7 @@ static INMEMORY_NETWORK_TYPE: &str = "inmemory";
 
 pub fn reg() {
     GLOBAL_TRANSPORT_REGISTRY.add_transport(INMEMORY_NETWORK_TYPE, ClientTransport::new());
+    global_registry().add_builder(Box::new(InMemoryResolverBuilder));
 }
 
 struct InMemoryResolverBuilder;

grpc/src/service.rs

@@ -24,7 +24,7 @@
 
 use std::{any::Any, pin::Pin};
 
-use futures_core::Stream;
+use tokio_stream::Stream;
 use tonic::{async_trait, Request as TonicRequest, Response as TonicResponse, Status};
 
 pub type Request = TonicRequest<Pin<Box<dyn Stream<Item = Box<dyn Message>> + Send + Sync>>>;