Skip to content
This repository has been archived by the owner on Feb 4, 2021. It is now read-only.

Commit

Permalink
feat(transport): Add service multiplexing/routing (hyperium#99)
Browse files Browse the repository at this point in the history
* 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 hyperium#29

Signed-off-by: Lucio Franco luciofranco14@gmail.com
  • Loading branch information
LucioFranco authored and rabbitinspace committed Jan 1, 2020
1 parent 1b31f81 commit 18c6aa3
Show file tree
Hide file tree
Showing 20 changed files with 473 additions and 176 deletions.
78 changes: 40 additions & 38 deletions tonic-build/src/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
}
}

Expand Down Expand Up @@ -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();

Expand Down
8 changes: 8 additions & 0 deletions tonic-examples/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
6 changes: 4 additions & 2 deletions tonic-examples/helloworld-tutorial.md
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand Down Expand Up @@ -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(())
Expand Down
5 changes: 4 additions & 1 deletion tonic-examples/routeguide-tutorial.md
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
}
Expand Down
4 changes: 2 additions & 2 deletions tonic-examples/src/authentication/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand Down
3 changes: 2 additions & 1 deletion tonic-examples/src/helloworld/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand Down
4 changes: 3 additions & 1 deletion tonic-examples/src/load_balance/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
37 changes: 37 additions & 0 deletions tonic-examples/src/multiplex/client.rs
Original file line number Diff line number Diff line change
@@ -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(())
}
69 changes: 69 additions & 0 deletions tonic-examples/src/multiplex/server.rs
Original file line number Diff line number Diff line change
@@ -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>>;
}
2 changes: 1 addition & 1 deletion tonic-examples/src/routeguide/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
}
Expand Down
3 changes: 2 additions & 1 deletion tonic-examples/src/tls/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand Down
2 changes: 1 addition & 1 deletion tonic-examples/src/tls_client_auth/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions tonic-examples/src/tls_client_auth/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand Down
Loading

0 comments on commit 18c6aa3

Please sign in to comment.