feat(grpc): Add tonic transport

codegen/src/main.rs

@@ -62,6 +62,20 @@ fn main() {
         false,
         false,
     );
+
+    // grpc
+    codegen(
+        &PathBuf::from(std::env!("CARGO_MANIFEST_DIR"))
+            .parent()
+            .unwrap()
+            .join("grpc"),
+        &["proto/echo/echo.proto"],
+        &["proto"],
+        &PathBuf::from("src/generated"),
+        &PathBuf::from("src/generated/echo_fds.rs"),
+        true,
+        true,
+    );
 }
 
 fn codegen(

grpc/Cargo.toml

@@ -7,6 +7,10 @@ license = "MIT"
 
 [dependencies]
 bytes = "1.10.1"
+futures = "0.3.31"
+tower = { version = "0.5.2", features = ["buffer", "limit", "util"] }
+tower-service = "0.3.3"
+socket2 = "0.5.10"
 futures-core = "0.3.31"
 futures-util = "0.3.31"
 hickory-resolver = { version = "0.25.1", optional = true }
@@ -20,19 +24,20 @@ pin-project-lite = "0.2.16"
 rand = "0.9"
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.140"
-socket2 = "0.5.10"
 tokio = { version = "1.37.0", features = ["sync", "rt", "net", "time", "macros"] }
 tokio-stream = "0.1.17"
 tonic = { version = "0.14.0", path = "../tonic", default-features = false, features = ["codegen", "transport"] }
-tower = "0.5.2"
-tower-service = "0.3.3"
 url = "2.5.0"
 
 [dev-dependencies]
 async-stream = "0.3.6"
 tonic = { version = "0.14.0", path = "../tonic", default-features = false, features = ["prost", "server", "router"] }
 hickory-server =  "0.25.2"
-prost = "0.13.5"
+prost = "0.14"
+
+[build-dependencies]
+tonic-build = { path = "../tonic-build" }
+prost = "0.14"
 
 [features]
 default = ["dns"]
@@ -43,5 +48,4 @@ allowed_external_types = [
     "tonic::*",
     "futures_core::stream::Stream",
     "tokio::sync::oneshot::Sender",
-    "once_cell::sync::Lazy",
 ]

grpc/examples/inmemory.rs

@@ -10,11 +10,8 @@ struct Handler {}
 #[derive(Debug)]
 struct MyReqMessage(String);
 
-impl Message for MyReqMessage {}
-
 #[derive(Debug)]
 struct MyResMessage(String);
-impl Message for MyResMessage {}
 
 #[async_trait]
 impl Service for Handler {

grpc/examples/multiaddr.rs

@@ -12,11 +12,8 @@ struct Handler {
 #[derive(Debug)]
 struct MyReqMessage(String);
 
-impl Message for MyReqMessage {}
-
 #[derive(Debug)]
 struct MyResMessage(String);
-impl Message for MyResMessage {}
 
 #[async_trait]
 impl Service for Handler {

grpc/proto/echo/echo.proto

@@ -0,0 +1,43 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+syntax = "proto3";
+
+package grpc.examples.echo;
+
+// EchoRequest is the request for echo.
+message EchoRequest {
+  string message = 1;
+}
+
+// EchoResponse is the response for echo.
+message EchoResponse {
+  string message = 1;
+}
+
+// Echo is the echo service.
+service Echo {
+  // UnaryEcho is unary echo.
+  rpc UnaryEcho(EchoRequest) returns (EchoResponse) {}
+  // ServerStreamingEcho is server side streaming.
+  rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}
+  // ClientStreamingEcho is client side streaming.
+  rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}
+  // BidirectionalStreamingEcho is bidi streaming.
+  rpc BidirectionalStreamingEcho(stream EchoRequest) returns (stream EchoResponse) {}
+}

grpc/src/client/channel.rs

@@ -262,6 +262,7 @@ impl ActiveChannel {
             tx.clone(),
             picker.clone(),
             connectivity_state.clone(),
+            runtime.clone(),
         );
 
         let resolver_helper = Box::new(tx.clone());
@@ -360,6 +361,7 @@ pub(crate) struct InternalChannelController {
     wqtx: WorkQueueTx,
     picker: Arc<Watcher<Arc<dyn Picker>>>,
     connectivity_state: Arc<Watcher<ConnectivityState>>,
+    runtime: Arc<dyn Runtime>,
 }
 
 impl InternalChannelController {
@@ -369,6 +371,7 @@ impl InternalChannelController {
         wqtx: WorkQueueTx,
         picker: Arc<Watcher<Arc<dyn Picker>>>,
         connectivity_state: Arc<Watcher<ConnectivityState>>,
+        runtime: Arc<dyn Runtime>,
     ) -> Self {
         let lb = Arc::new(GracefulSwitchBalancer::new(wqtx.clone()));
 
@@ -380,6 +383,7 @@ impl InternalChannelController {
             wqtx,
             picker,
             connectivity_state,
+            runtime,
         }
     }
 
@@ -429,6 +433,7 @@ impl load_balancing::ChannelController for InternalChannelController {
             Box::new(move |k: SubchannelKey| {
                 scp.unregister_subchannel(&k);
             }),
+            self.runtime.clone(),
         );
         let _ = self.subchannel_pool.register_subchannel(&key, isc.clone());
         self.new_esc_for_isc(isc)

grpc/src/client/mod.rs

@@ -28,9 +28,8 @@ pub mod channel;
 pub(crate) mod load_balancing;
 pub(crate) mod name_resolution;
 pub mod service_config;
-pub mod transport;
-
 mod subchannel;
+pub(crate) mod transport;
 pub use channel::Channel;
 pub use channel::ChannelOptions;
 

grpc/src/client/name_resolution/dns/test.rs

@@ -293,6 +293,14 @@ impl rt::Runtime for FakeRuntime {
     fn sleep(&self, duration: std::time::Duration) -> Pin<Box<dyn rt::Sleep>> {
         self.inner.sleep(duration)
     }
+
+    fn tcp_stream(
+        &self,
+        target: std::net::SocketAddr,
+        opts: rt::TcpOptions,
+    ) -> Pin<Box<dyn Future<Output = Result<Box<dyn rt::TcpStream>, String>> + Send>> {
+        self.inner.tcp_stream(target, opts)
+    }
 }
 
 #[tokio::test]

grpc/src/client/name_resolution/registry.rs

@@ -74,7 +74,7 @@ impl ResolverRegistry {
             .lock()
             .unwrap()
             .insert(scheme.to_string(), Arc::from(builder));
-        return Ok(());
+        Ok(())
     }
 
     /// Returns the resolver builder registered for the given scheme, if any.

grpc/src/client/subchannel.rs

@@ -2,11 +2,16 @@ use super::{
     channel::{InternalChannelController, WorkQueueTx},
     load_balancing::{self, ExternalSubchannel, Picker, Subchannel, SubchannelState},
     name_resolution::Address,
-    transport::{self, ConnectedTransport, Transport, TransportRegistry},
+    transport::{self, Transport, TransportRegistry},
     ConnectivityState,
 };
 use crate::{
-    client::{channel::WorkQueueItem, subchannel},
+    client::{
+        channel::WorkQueueItem,
+        subchannel,
+        transport::{ConnectedTransport, TransportOptions},
+    },
+    rt::{Runtime, TaskHandle},
     service::{Request, Response, Service},
 };
 use core::panic;
@@ -18,13 +23,13 @@ use std::{
     sync::{Arc, Mutex, RwLock, Weak},
 };
 use tokio::{
-    sync::{mpsc, watch, Notify},
+    sync::{mpsc, oneshot, watch, Notify},
     task::{AbortHandle, JoinHandle},
     time::{Duration, Instant},
 };
 use tonic::async_trait;
 
-type SharedService = Arc<dyn ConnectedTransport>;
+type SharedService = Arc<dyn Service>;
 
 pub trait Backoff: Send + Sync {
     fn backoff_until(&self) -> Instant;
@@ -52,11 +57,11 @@ enum InternalSubchannelState {
 }
 
 struct InternalSubchannelConnectingState {
-    abort_handle: Option<AbortHandle>,
+    abort_handle: Option<Box<dyn TaskHandle>>,
 }
 
 struct InternalSubchannelReadyState {
-    abort_handle: Option<AbortHandle>,
+    abort_handle: Option<Box<dyn TaskHandle>>,
     svc: SharedService,
 }
 
@@ -178,6 +183,7 @@ pub(crate) struct InternalSubchannel {
     unregister_fn: Option<Box<dyn FnOnce(SubchannelKey) + Send + Sync>>,
     state_machine_event_sender: mpsc::UnboundedSender<SubchannelStateMachineEvent>,
     inner: Mutex<InnerSubchannel>,
+    runtime: Arc<dyn Runtime>,
 }
 
 struct InnerSubchannel {
@@ -204,7 +210,7 @@ impl Service for InternalSubchannel {
 
 enum SubchannelStateMachineEvent {
     ConnectionRequested,
-    ConnectionSucceeded(SharedService),
+    ConnectionSucceeded(SharedService, oneshot::Receiver<Result<(), String>>),
     ConnectionTimedOut,
     ConnectionFailed(String),
     ConnectionTerminated,
@@ -214,7 +220,7 @@ impl Debug for SubchannelStateMachineEvent {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::ConnectionRequested => write!(f, "ConnectionRequested"),
-            Self::ConnectionSucceeded(_) => write!(f, "ConnectionSucceeded"),
+            Self::ConnectionSucceeded(_, _) => write!(f, "ConnectionSucceeded"),
             Self::ConnectionTimedOut => write!(f, "ConnectionTimedOut"),
             Self::ConnectionFailed(_) => write!(f, "ConnectionFailed"),
             Self::ConnectionTerminated => write!(f, "ConnectionTerminated"),
@@ -229,6 +235,7 @@ impl InternalSubchannel {
         transport: Arc<dyn Transport>,
         backoff: Arc<dyn Backoff>,
         unregister_fn: Box<dyn FnOnce(SubchannelKey) + Send + Sync>,
+        runtime: Arc<dyn Runtime>,
     ) -> Arc<InternalSubchannel> {
         println!("creating new internal subchannel for: {:?}", &key);
         let (tx, mut rx) = mpsc::unbounded_channel::<SubchannelStateMachineEvent>();
@@ -244,23 +251,24 @@ impl InternalSubchannel {
                 backoff_task: None,
                 disconnect_task: None,
             }),
+            runtime: runtime.clone(),
         });
 
         // This long running task implements the subchannel state machine. When
         // the subchannel is dropped, the channel from which this task reads is
         // closed, and therefore this task exits because rx.recv() returns None
         // in that case.
         let arc_to_self = Arc::clone(&isc);
-        tokio::task::spawn(async move {
+        runtime.spawn(Box::pin(async move {
             println!("starting subchannel state machine for: {:?}", &key);
             while let Some(m) = rx.recv().await {
                 println!("subchannel {:?} received event {:?}", &key, &m);
                 match m {
                     SubchannelStateMachineEvent::ConnectionRequested => {
                         arc_to_self.move_to_connecting();
                     }
-                    SubchannelStateMachineEvent::ConnectionSucceeded(svc) => {
-                        arc_to_self.move_to_ready(svc);
+                    SubchannelStateMachineEvent::ConnectionSucceeded(svc, rx) => {
+                        arc_to_self.move_to_ready(svc, rx);
                     }
                     SubchannelStateMachineEvent::ConnectionTimedOut => {
                         arc_to_self.move_to_transient_failure("connect timeout expired".into());
@@ -277,7 +285,7 @@ impl InternalSubchannel {
                 }
             }
             println!("exiting work queue task in subchannel");
-        });
+        }));
         isc
     }
 
@@ -345,30 +353,34 @@ impl InternalSubchannel {
         let transport = self.transport.clone();
         let address = self.address().address;
         let state_machine_tx = self.state_machine_event_sender.clone();
-        let connect_task = tokio::task::spawn(async move {
+        // TODO: All these options to be configured by users.
+        let transport_opts = TransportOptions::default();
+        let runtime = self.runtime.clone();
+
+        let connect_task = self.runtime.spawn(Box::pin(async move {
             tokio::select! {
                 _ = tokio::time::sleep(min_connect_timeout) => {
                     let _ = state_machine_tx.send(SubchannelStateMachineEvent::ConnectionTimedOut);
                 }
-                result = transport.connect(address.to_string().clone()) => {
+                result = transport.connect(address.to_string().clone(), runtime, &transport_opts) => {
                     match result {
                         Ok(s) => {
-                            let _ = state_machine_tx.send(SubchannelStateMachineEvent::ConnectionSucceeded(Arc::from(s)));
+                            let _ = state_machine_tx.send(SubchannelStateMachineEvent::ConnectionSucceeded(Arc::from(s.service), s.disconnection_listener));
                         }
                         Err(e) => {
                             let _ = state_machine_tx.send(SubchannelStateMachineEvent::ConnectionFailed(e));
                         }
                     }
                 },
             }
-        });
+        }));
         let mut inner = self.inner.lock().unwrap();
         inner.state = InternalSubchannelState::Connecting(InternalSubchannelConnectingState {
-            abort_handle: Some(connect_task.abort_handle()),
+            abort_handle: Some(connect_task),
         });
     }
 
-    fn move_to_ready(&self, svc: SharedService) {
+    fn move_to_ready(&self, svc: SharedService, closed_rx: oneshot::Receiver<Result<(), String>>) {
         let svc2 = svc.clone();
         {
             let mut inner = self.inner.lock().unwrap();
@@ -383,17 +395,19 @@ impl InternalSubchannel {
         });
 
         let state_machine_tx = self.state_machine_event_sender.clone();
-        let disconnect_task = tokio::task::spawn(async move {
+        let task_handle = self.runtime.spawn(Box::pin(async move {
             // TODO(easwars): Does it make sense for disconnected() to return an
             // error string containing information about why the connection
             // terminated? But what can we do with that error other than logging
             // it, which the transport can do as well?
-            svc.disconnected().await;
+            if let Err(e) = closed_rx.await {
+                eprintln!("Transport closed with error: {}", e.to_string())
+            };
             let _ = state_machine_tx.send(SubchannelStateMachineEvent::ConnectionTerminated);
-        });
+        }));
         let mut inner = self.inner.lock().unwrap();
         inner.state = InternalSubchannelState::Ready(InternalSubchannelReadyState {
-            abort_handle: Some(disconnect_task.abort_handle()),
+            abort_handle: Some(task_handle),
             svc: svc2.clone(),
         });
     }