From 1630547f976eeeb0ffeae5a8aef4723c67be7057 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Fri, 16 Aug 2019 14:46:18 -0400 Subject: [PATCH] Inital pass at route guide --- tonic-examples/Cargo.toml | 4 + tonic-examples/build.rs | 6 + .../proto/routeguide/route_guide.proto | 110 ++++++++++++++++ tonic-examples/src/helloworld/server.rs | 9 +- tonic-examples/src/routeguide/server.rs | 68 ++++++++++ tonic-macros/src/service.rs | 122 +++++++++++++++++- tonic/examples/server.rs | 104 --------------- tonic/src/body.rs | 1 + tonic/src/lib.rs | 2 + tonic/tests/server.rs | 19 +++ 10 files changed, 331 insertions(+), 114 deletions(-) create mode 100644 tonic-examples/proto/routeguide/route_guide.proto create mode 100644 tonic-examples/src/routeguide/server.rs delete mode 100644 tonic/examples/server.rs diff --git a/tonic-examples/Cargo.toml b/tonic-examples/Cargo.toml index f78ae81..d87a33e 100644 --- a/tonic-examples/Cargo.toml +++ b/tonic-examples/Cargo.toml @@ -14,6 +14,10 @@ path = "src/helloworld/server.rs" # name = "helloworld-client" # path = "src/helloworld/client.rs" +[[bin]] +name = "routeguide-server" +path = "src/routeguide/server.rs" + [dependencies] tonic = { path = "../tonic" } tower-h2 = { path = "../tower-h2" } diff --git a/tonic-examples/build.rs b/tonic-examples/build.rs index b6fc145..6760737 100644 --- a/tonic-examples/build.rs +++ b/tonic-examples/build.rs @@ -4,4 +4,10 @@ fn main() { &["proto/helloworld"], ) .unwrap(); + + tonic_build::compile_protos( + &["proto/routeguide/route_guide.proto"], + &["proto/routeguide"], + ) + .unwrap(); } diff --git a/tonic-examples/proto/routeguide/route_guide.proto b/tonic-examples/proto/routeguide/route_guide.proto new file mode 100644 index 0000000..fe21e43 --- /dev/null +++ b/tonic-examples/proto/routeguide/route_guide.proto @@ -0,0 +1,110 @@ +// Copyright 2015 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"; + +option java_multiple_files = true; +option java_package = "io.grpc.examples.routeguide"; +option java_outer_classname = "RouteGuideProto"; + +package routeguide; + +// Interface exported by the server. +service RouteGuide { + // A simple RPC. + // + // Obtains the feature at a given position. + // + // A feature with an empty name is returned if there's no feature at the given + // position. + rpc GetFeature(Point) returns (Feature) {} + + // A server-to-client streaming RPC. + // + // Obtains the Features available within the given Rectangle. Results are + // streamed rather than returned at once (e.g. in a response message with a + // repeated field), as the rectangle may cover a large area and contain a + // huge number of features. + rpc ListFeatures(Rectangle) returns (stream Feature) {} + + // A client-to-server streaming RPC. + // + // Accepts a stream of Points on a route being traversed, returning a + // RouteSummary when traversal is completed. + rpc RecordRoute(stream Point) returns (RouteSummary) {} + + // A Bidirectional streaming RPC. + // + // Accepts a stream of RouteNotes sent while a route is being traversed, + // while receiving other RouteNotes (e.g. from other users). + rpc RouteChat(stream RouteNote) returns (stream RouteNote) {} +} + +// Points are represented as latitude-longitude pairs in the E7 representation +// (degrees multiplied by 10**7 and rounded to the nearest integer). +// Latitudes should be in the range +/- 90 degrees and longitude should be in +// the range +/- 180 degrees (inclusive). +message Point { + int32 latitude = 1; + int32 longitude = 2; +} + +// A latitude-longitude rectangle, represented as two diagonally opposite +// points "lo" and "hi". +message Rectangle { + // One corner of the rectangle. + Point lo = 1; + + // The other corner of the rectangle. + Point hi = 2; +} + +// A feature names something at a given point. +// +// If a feature could not be named, the name is empty. +message Feature { + // The name of the feature. + string name = 1; + + // The point where the feature is detected. + Point location = 2; +} + +// A RouteNote is a message sent while at a given point. +message RouteNote { + // The location from which the message is sent. + Point location = 1; + + // The message to be sent. + string message = 2; +} + +// A RouteSummary is received in response to a RecordRoute rpc. +// +// It contains the number of individual points received, the number of +// detected features, and the total distance covered as the cumulative sum of +// the distance between each point. +message RouteSummary { + // The number of points received. + int32 point_count = 1; + + // The number of known features passed while traversing the route. + int32 feature_count = 2; + + // The distance covered in metres. + int32 distance = 3; + + // The duration of the traversal in seconds. + int32 elapsed_time = 4; +} diff --git a/tonic-examples/src/helloworld/server.rs b/tonic-examples/src/helloworld/server.rs index 8e91c72..b5f7f49 100644 --- a/tonic-examples/src/helloworld/server.rs +++ b/tonic-examples/src/helloworld/server.rs @@ -1,7 +1,7 @@ #![feature(async_await)] use std::time::Duration; -use tokio::{timer::Delay, net::TcpListener}; +use tokio::{net::TcpListener, timer::Delay}; use tonic::{Request, Response, Status}; use tower_h2::Server; @@ -16,7 +16,10 @@ pub struct MyGreeter { #[tonic::server(service = "helloworld.Greeter", proto = "hello_world")] impl MyGreeter { - pub async fn say_hello(&self, request: Request) -> Result, Status> { + pub async fn say_hello( + &self, + request: Request, + ) -> Result, Status> { println!("Got a request: {:?}", request); let string = &self.data; @@ -27,7 +30,7 @@ impl MyGreeter { println!("My data: {:?}", string); Delay::new(when).await; - + let reply = hello_world::HelloReply { message: "Zomg, it works!".into(), }; diff --git a/tonic-examples/src/routeguide/server.rs b/tonic-examples/src/routeguide/server.rs new file mode 100644 index 0000000..0c522cf --- /dev/null +++ b/tonic-examples/src/routeguide/server.rs @@ -0,0 +1,68 @@ +#![feature(async_await)] + +use futures::Stream; +use tokio::net::TcpListener; +use tonic::{Request, Response, Status}; +use tower_h2::Server; + +pub mod routeguide { + include!(concat!(env!("OUT_DIR"), "/routeguide.rs")); +} + +use routeguide::*; + +type BoxStream = Pin> + Send + 'static>>; + +#[derive(Default, Clone)] +pub struct RouteGuide { + data: String, +} + +#[tonic::server(service = "routeguide.RouteGuide", proto = "routeguide")] +impl RouteGuide { + pub async fn get_feature(&self, _req: Request) -> Result, Status> { + unimplemented!() + } + + pub async fn list_features( + &self, + _req: Request, + ) -> Result>, Status> { + unimplemented!() + } + + pub async fn record_route( + &self, + _req: Request>>, + ) -> Result, Status> { + unimplemented!() + } + + // pub async fn route_chat( + // &self, + // _req: Request>>, + // ) -> Result>>, Status> { + // unimplemented!() + // } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let addr = "[::1]:50051".parse().unwrap(); + let mut bind = TcpListener::bind(&addr)?; + + let route_guide = RouteGuide::default(); + let mut server = Server::new(RouteGuideServer::new(route_guide), Default::default()); + + while let Ok((sock, _addr)) = bind.accept().await { + if let Err(e) = sock.set_nodelay(true) { + return Err(e.into()); + } + + if let Err(e) = server.serve(sock).await { + println!("H2 ERROR: {}", e); + } + } + + Ok(()) +} diff --git a/tonic-macros/src/service.rs b/tonic-macros/src/service.rs index 40026db..9026215 100644 --- a/tonic-macros/src/service.rs +++ b/tonic-macros/src/service.rs @@ -126,12 +126,30 @@ fn generate_methods(service: &ServiceDef) -> TokenStream { ); let method_path = Lit::Str(LitStr::new(&path, Span::call_site())); - let method_stream = generate_unary( - method, - ident.clone(), - service.name.clone(), - &service.proto_path, - ); + let method_stream = match (method.client_streaming, method.server_streaming) { + (false, false) => generate_unary( + method, + ident.clone(), + service.name.clone(), + &service.proto_path, + ), + + (false, true) => generate_server_streaming( + method, + ident.clone(), + service.name.clone(), + &service.proto_path, + ), + + (true, false) => generate_client_streaming( + method, + ident.clone(), + service.name.clone(), + &service.proto_path, + ), + + _ => unimplemented!("method type"), + }; let method = quote! { #method_path => { @@ -160,7 +178,7 @@ fn generate_unary( struct #service_ident(pub std::sync::Arc<#service_impl>); impl tonic::server::UnaryService<#request> for #service_ident { - type Response =#response; + type Response = #response; type Future = BoxFuture, tonic::Status>; fn call(&mut self, request: tonic::Request<#request>) -> Self::Future { @@ -184,3 +202,93 @@ fn generate_unary( Box::pin(fut) } } + +fn generate_server_streaming( + method: &Method, + method_ident: Ident, + service_impl: Path, + proto_path: &str, +) -> TokenStream { + let service_ident = Ident::new(&method.proto_name, Span::call_site()); + + let request: Path = syn::parse_str(&format!("{}::{}", proto_path, method.input_type)).unwrap(); + let response: Path = + syn::parse_str(&format!("{}::{}", proto_path, method.output_type)).unwrap(); + + // TODO: parse response stream type, if it is a concrete type then use that + // as the ResponseStream type, if it is a impl Trait then we need to box. + quote! { + struct #service_ident(pub std::sync::Arc<#service_impl>); + + impl tonic::server::ServerStreamingService<#request> for #service_ident { + type Response = #response; + type ResponseStream = Pin> + Send>>; + type Future = BoxFuture, tonic::Status>; + + fn call(&mut self, request: tonic::Request<#request>) -> Self::Future { + let inner = self.0.clone(); + let fut = async move { + inner.#method_ident(request).await + + }; + Box::pin(fut) + } + } + + let inner = self.inner.clone(); + let fut = async move { + let method = #service_ident(inner); + let codec = tonic::codec::ProstCodec::new(); + let mut grpc = tonic::server::Grpc::new(codec); + let res = grpc.server_streaming(method, req).await; + Ok(res) + }; + + Box::pin(fut) + } +} + +fn generate_client_streaming( + method: &Method, + method_ident: Ident, + service_impl: Path, + proto_path: &str, +) -> TokenStream { + let service_ident = Ident::new(&method.proto_name, Span::call_site()); + + let request: Path = syn::parse_str(&format!("{}::{}", proto_path, method.input_type)).unwrap(); + let response: Path = + syn::parse_str(&format!("{}::{}", proto_path, method.output_type)).unwrap(); + + // TODO: parse response stream type, if it is a concrete type then use that + // as the ResponseStream type, if it is a impl Trait then we need to box. + quote! { + struct #service_ident(pub std::sync::Arc<#service_impl>); + + impl tonic::server::ClientStreamingService for #service_ident + where S: Stream> + Unpin + Send + 'static { + type Response = #response; + type Future = BoxFuture, tonic::Status>; + + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = self.0.clone(); + let fut = async move { + inner.#method_ident(request).await + + }; + Box::pin(fut) + } + } + + let inner = self.inner.clone(); + let fut = async move { + let method = #service_ident(inner); + let codec = tonic::codec::ProstCodec::new(); + let mut grpc = tonic::server::Grpc::new(codec); + let res = grpc.client_streaming(method, req).await; + Ok(res) + }; + + Box::pin(fut) + } +} diff --git a/tonic/examples/server.rs b/tonic/examples/server.rs deleted file mode 100644 index af5b228..0000000 --- a/tonic/examples/server.rs +++ /dev/null @@ -1,104 +0,0 @@ -#![feature(async_await, type_alias_impl_trait)] - -use futures_util::future; -use std::future::Future; -use std::task::{Context, Poll}; -use tokio::net::TcpListener; -use tonic::{ - body, - server::{Grpc, UnaryService}, - Request, Response, Status, -}; -use tower_h2::{RecvBody, Server}; -use tower_service::Service; - -#[derive(Clone, PartialEq, prost::Message)] -pub struct HelloRequest { - #[prost(string, tag = "1")] - pub name: std::string::String, -} -/// The response message containing the greetings -#[derive(Clone, PartialEq, prost::Message)] -pub struct HelloReply { - #[prost(string, tag = "1")] - pub message: std::string::String, -} - -struct SayHello; - -impl UnaryService for SayHello { - type Response = HelloReply; - type Future = impl Future, Status>>; - - fn call(&mut self, request: Request) -> Self::Future { - async move { - println!("REQUEST = {:?}", request); - - let reply = HelloReply { - message: "Zomg, it works!".to_string(), - }; - - Ok(Response::new(reply)) - } - } -} - -#[tokio::main] -async fn main() -> Result<(), Box> { - let addr = "[::1]:50051".parse().unwrap(); - let mut bind = TcpListener::bind(&addr)?; - - let mut server = Server::new(MakeSvc, Default::default()); - - while let Ok((sock, _addr)) = bind.accept().await { - if let Err(e) = sock.set_nodelay(true) { - return Err(e.into()); - } - - if let Err(e) = server.serve(sock).await { - println!("H2 ERROR: {}", e); - } - } - - Ok(()) -} - -#[derive(Debug)] -pub struct Svc; - -impl Service> for Svc { - type Response = http::Response; - type Error = tonic::error::Never; - type Future = impl Future>; - - fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { - Ok(()).into() - } - - fn call(&mut self, req: http::Request) -> Self::Future { - let fut = async move { - let codec = tonic::codec::ProstCodec::new(); - let mut grpc = Grpc::new(codec); - let response = grpc.unary(SayHello, req).await; - Ok(response) - }; - - Box::pin(fut) - } -} - -pub struct MakeSvc; - -impl Service<()> for MakeSvc { - type Response = Svc; - type Error = std::io::Error; - type Future = future::Ready>; - - fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { - Ok(()).into() - } - - fn call(&mut self, _: ()) -> Self::Future { - future::ok(Svc) - } -} diff --git a/tonic/src/body.rs b/tonic/src/body.rs index 25ba770..c27c897 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -98,6 +98,7 @@ impl Body for BoxAsyncBody { } } +// TODO: refactor this to accept an !Unpin stream #[derive(Debug)] pub struct AsyncBody { inner: S, diff --git a/tonic/src/lib.rs b/tonic/src/lib.rs index 1868840..eb7078c 100644 --- a/tonic/src/lib.rs +++ b/tonic/src/lib.rs @@ -43,6 +43,8 @@ pub mod _codegen { pub type BoxFuture = self::Pin> + Send + 'static>>; + pub type BoxStream = + self::Pin> + Send + 'static>>; pub mod http { pub use http::*; diff --git a/tonic/tests/server.rs b/tonic/tests/server.rs index ddd6750..05c23d2 100644 --- a/tonic/tests/server.rs +++ b/tonic/tests/server.rs @@ -6,6 +6,10 @@ use tokio_buf::BufStream; use tonic::codec::UnitCodec; use tonic::server::*; use tonic::{Request, Response, Status}; +use std::pin::Pin; +use futures_core::Stream; + +type BoxStream = Pin> + Send + 'static>>; struct SayHello; @@ -18,6 +22,18 @@ impl UnaryService<()> for SayHello { } } +struct SayHelloStream; + +impl ClientStreamingService for SayHelloStream +where S: Stream{ + type Response = (); + type Future = impl Future, Status>>; + + fn call(&mut self, _: Request) -> Self::Future { + async move { Ok(Response::new(())) } + } +} + #[tokio::test] async fn say_hello() { let codec = UnitCodec::default(); @@ -25,6 +41,9 @@ async fn say_hello() { let request = http::Request::new(Body(Vec::new())); grpc.unary(SayHello, request).await; + + let request = http::Request::new(Body(Vec::new())); + grpc.client_streaming(SayHelloStream, request).await; } #[derive(Debug, Default, Clone)]