feat(transport): Add service multiplexing/routing (#99)

* feat(transport): Add service multiplexing/routing This change introduces a new "router" built on top of `transport::Server` that allows one to run multiple gRPC services on the same socket. ```rust Server::builder() .add_service(greeter) .add_service(echo) .serve(addr) .await?; ``` There is also a new `multiplex` example showcasing server side service multiplexing and client side service multiplexing. BREAKING CHANGES: `Server::serve` is now crate private and all services must be added via `Server::add_service`. Codegen also returns just a `Service` now instead of a `MakeService` pair. Closes #29 Signed-off-by: Lucio Franco [email protected]
hyperium · Oct 29, 2019 · 5b4f468 · 5b4f468
1 parent a17049f
commit 5b4f468
Show file tree

Hide file tree

Showing 20 changed files with 473 additions and 176 deletions.
diff --git a/tonic-build/src/server.rs b/tonic-build/src/server.rs
@@ -7,63 +7,32 @@ use syn::{Ident, Lit, LitStr};
 pub(crate) fn generate(service: &Service, proto_path: &str) -> TokenStream {
     let methods = generate_methods(&service, proto_path);
 
-    let server_make_service = quote::format_ident!("{}Server", service.name);
-    let server_service = quote::format_ident!("{}ServerSvc", service.name);
+    let server_service = quote::format_ident!("{}Server", service.name);
     let server_trait = quote::format_ident!("{}", service.name);
     let generated_trait = generate_trait(service, proto_path, server_trait.clone());
     let service_doc = generate_doc_comments(&service.comments.leading);
-    let server_new_doc = generate_doc_comment(&format!(
-        "Create a new {} from a type that implements {}.",
-        server_make_service, server_trait
-    ));
+
+    // Transport based implementations
+    let path = format!("{}.{}", service.package, service.proto_name);
+    let transport = generate_transport(&server_service, &server_trait, &path);
 
     quote! {
         #generated_trait
 
         #service_doc
-        #[derive(Clone, Debug)]
-        pub struct #server_make_service<T: #server_trait> {
-            inner: Arc<T>,
-        }
-
-        #[derive(Clone, Debug)]
+        #[derive(Debug)]
         #[doc(hidden)]
         pub struct #server_service<T: #server_trait> {
             inner: Arc<T>,
         }
 
-        impl<T: #server_trait> #server_make_service<T> {
-            #server_new_doc
+        impl<T: #server_trait> #server_service<T> {
             pub fn new(inner: T) -> Self {
                 let inner = Arc::new(inner);
-                Self::from_shared(inner)
-            }
-
-            pub fn from_shared(inner: Arc<T>) -> Self {
                 Self { inner }
             }
         }
 
-        impl<T: #server_trait> #server_service<T> {
-            pub fn new(inner: Arc<T>) -> Self {
-                Self { inner }
-            }
-        }
-
-        impl<T: #server_trait, R> Service<R> for #server_make_service<T> {
-            type Response = #server_service<T>;
-            type Error = Never;
-            type Future = Ready<Result<Self::Response, Self::Error>>;
-
-            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-                Poll::Ready(Ok(()))
-            }
-
-            fn call(&mut self, _: R) -> Self::Future {
-                ok(#server_service::new(self.inner.clone()))
-            }
-        }
-
         impl<T: #server_trait> Service<http::Request<HyperBody>> for #server_service<T> {
             type Response = http::Response<tonic::body::BoxBody>;
             type Error = Never;
@@ -89,6 +58,15 @@ pub(crate) fn generate(service: &Service, proto_path: &str) -> TokenStream {
                 }
             }
         }
+
+        impl<T: #server_trait> Clone for #server_service<T> {
+            fn clone(&self) -> Self {
+                let inner = self.inner.clone();
+                Self { inner }
+            }
+        }
+
+        #transport
     }
 }
 
@@ -181,6 +159,30 @@ fn generate_trait_methods(service: &Service, proto_path: &str) -> TokenStream {
     stream
 }
 
+#[cfg(feature = "transport")]
+fn generate_transport(
+    server_service: &syn::Ident,
+    server_trait: &syn::Ident,
+    service_name: &str,
+) -> TokenStream {
+    let service_name = syn::LitStr::new(service_name, proc_macro2::Span::call_site());
+
+    quote! {
+        impl<T: #server_trait> tonic::transport::ServiceName for #server_service<T> {
+            const NAME: &'static str = #service_name;
+        }
+    }
+}
+
+#[cfg(not(feature = "transport"))]
+fn generate_transport(
+    _server_service: &syn::Ident,
+    _server_trait: &syn::Ident,
+    _service_name: &str,
+) -> TokenStream {
+    TokenStream::new()
+}
+
 fn generate_methods(service: &Service, proto_path: &str) -> TokenStream {
     let mut stream = TokenStream::new();
 

diff --git a/tonic-examples/Cargo.toml b/tonic-examples/Cargo.toml
@@ -52,6 +52,14 @@ path = "src/tls_client_auth/server.rs"
 name = "tls-client-auth-client"
 path = "src/tls_client_auth/client.rs"
 
+[[bin]]
+name = "multiplex-server"
+path = "src/multiplex/server.rs"
+
+[[bin]]
+name = "multiplex-client"
+path = "src/multiplex/client.rs"
+
 [[bin]]
 name = "gcp-client"
 path = "src/gcp/client.rs"

diff --git a/tonic-examples/helloworld-tutorial.md b/tonic-examples/helloworld-tutorial.md
@@ -191,7 +191,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let greeter = MyGreeter {};
 
     Server::builder()
-        .serve(addr, GreeterServer::new(greeter))
+        .add_service(GreeterServer::new(greeter))
+        .serve(addr)
         .await?;
 
     Ok(())
@@ -236,7 +237,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let greeter = MyGreeter {};
 
     Server::builder()
-        .serve(addr, GreeterServer::new(greeter))
+        .add_service(GreeterServer::new(greeter))
+        .serve(addr)
         .await?;
 
     Ok(())

diff --git a/tonic-examples/routeguide-tutorial.md b/tonic-examples/routeguide-tutorial.md
@@ -558,7 +558,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let svc = server::RouteGuideServer::new(route_guide);
 
-    Server::builder().serve(addr, svc).await?;
+    Server::builder()
+        .add_service(svc)
+        .serve(addr)
+        .await?;
 
     Ok(())
 }

diff --git a/tonic-examples/src/authentication/server.rs b/tonic-examples/src/authentication/server.rs
@@ -78,8 +78,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 }
             }
         })
-        .clone()
-        .serve(addr, pb::server::EchoServer::new(server))
+        .add_service(pb::server::EchoServer::new(server))
+        .serve(addr)
         .await?;
 
     Ok(())

diff --git a/tonic-examples/src/helloworld/server.rs b/tonic-examples/src/helloworld/server.rs
@@ -33,7 +33,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let greeter = MyGreeter::default();
 
     Server::builder()
-        .serve(addr, GreeterServer::new(greeter))
+        .add_service(GreeterServer::new(greeter))
+        .serve(addr)
         .await?;
 
     Ok(())

diff --git a/tonic-examples/src/load_balance/server.rs b/tonic-examples/src/load_balance/server.rs
@@ -60,7 +60,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         let mut tx = tx.clone();
 
         let server = EchoServer { addr };
-        let serve = Server::builder().serve(addr, pb::server::EchoServer::new(server));
+        let serve = Server::builder()
+            .add_service(pb::server::EchoServer::new(server))
+            .serve(addr);
 
         tokio::spawn(async move {
             if let Err(e) = serve.await {

diff --git a/tonic-examples/src/multiplex/client.rs b/tonic-examples/src/multiplex/client.rs
@@ -0,0 +1,37 @@
+pub mod hello_world {
+    tonic::include_proto!("helloworld");
+}
+
+pub mod echo {
+    tonic::include_proto!("grpc.examples.echo");
+}
+
+use echo::{client::EchoClient, EchoRequest};
+use hello_world::{client::GreeterClient, HelloRequest};
+use tonic::transport::Endpoint;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let channel = Endpoint::from_static("http://[::1]:50051").channel();
+
+    let mut greeter_client = GreeterClient::new(channel.clone());
+    let mut echo_client = EchoClient::new(channel);
+
+    let request = tonic::Request::new(HelloRequest {
+        name: "Tonic".into(),
+    });
+
+    let response = greeter_client.say_hello(request).await?;
+
+    println!("GREETER RESPONSE={:?}", response);
+
+    let request = tonic::Request::new(EchoRequest {
+        message: "hello".into(),
+    });
+
+    let response = echo_client.unary_echo(request).await?;
+
+    println!("ECHO RESPONSE={:?}", response);
+
+    Ok(())
+}
diff --git a/tonic-examples/src/multiplex/server.rs b/tonic-examples/src/multiplex/server.rs
@@ -0,0 +1,69 @@
+use std::collections::VecDeque;
+use tonic::{transport::Server, Request, Response, Status};
+
+pub mod hello_world {
+    tonic::include_proto!("helloworld");
+}
+
+pub mod echo {
+    tonic::include_proto!("grpc.examples.echo");
+}
+
+use hello_world::{
+    server::{Greeter, GreeterServer},
+    HelloReply, HelloRequest,
+};
+
+use echo::{
+    server::{Echo, EchoServer},
+    EchoRequest, EchoResponse,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:50051".parse().unwrap();
+
+    let greeter = GreeterServer::new(MyGreeter::default());
+    let echo = EchoServer::new(MyEcho::default());
+
+    Server::builder()
+        .add_service(greeter)
+        .add_service(echo)
+        .serve(addr)
+        .await?;
+
+    Ok(())
+}
+
+#[derive(Default)]
+pub struct MyGreeter {}
+
+#[tonic::async_trait]
+impl Greeter for MyGreeter {
+    async fn say_hello(
+        &self,
+        request: Request<HelloRequest>,
+    ) -> Result<Response<HelloReply>, Status> {
+        let reply = hello_world::HelloReply {
+            message: format!("Hello {}!", request.into_inner().name).into(),
+        };
+        Ok(Response::new(reply))
+    }
+}
+
+#[derive(Default)]
+pub struct MyEcho;
+
+#[tonic::async_trait]
+impl Echo for MyEcho {
+    async fn unary_echo(
+        &self,
+        request: Request<EchoRequest>,
+    ) -> Result<Response<EchoResponse>, Status> {
+        let message = request.into_inner().message;
+        Ok(Response::new(EchoResponse { message }))
+    }
+
+    type ServerStreamingEchoStream = VecDeque<Result<EchoResponse, Status>>;
+    type BidirectionalStreamingEchoStream = VecDeque<Result<EchoResponse, Status>>;
+}
diff --git a/tonic-examples/src/routeguide/server.rs b/tonic-examples/src/routeguide/server.rs
@@ -155,7 +155,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let svc = server::RouteGuideServer::new(route_guide);
 
-    Server::builder().serve(addr, svc).await?;
+    Server::builder().add_service(svc).serve(addr).await?;
 
     Ok(())
 }

diff --git a/tonic-examples/src/tls/server.rs b/tonic-examples/src/tls/server.rs
@@ -61,7 +61,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     Server::builder()
         .tls_config(ServerTlsConfig::with_rustls().identity(identity))
         .clone()
-        .serve(addr, pb::server::EchoServer::new(server))
+        .add_service(pb::server::EchoServer::new(server))
+        .serve(addr)
         .await?;
 
     Ok(())

diff --git a/tonic-examples/src/tls_client_auth/client.rs b/tonic-examples/src/tls_client_auth/client.rs
@@ -13,7 +13,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let client_key = tokio::fs::read("tonic-examples/data/tls/client1.key").await?;
     let client_identity = Identity::from_pem(client_cert, client_key);
 
-    let tls = ClientTlsConfig::with_openssl()
+    let tls = ClientTlsConfig::with_rustls()
         .domain_name("localhost")
         .ca_certificate(server_root_ca_cert)
         .identity(client_identity)

diff --git a/tonic-examples/src/tls_client_auth/server.rs b/tonic-examples/src/tls_client_auth/server.rs
@@ -44,8 +44,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     Server::builder()
         .tls_config(&tls)
-        .clone()
-        .serve(addr, pb::server::EchoServer::new(server))
+        .add_service(pb::server::EchoServer::new(server))
+        .serve(addr)
         .await?;
 
     Ok(())