From e00ddc920fc59c23ce4c1928f6136a63ba4f4217 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Tue, 3 Sep 2019 23:53:42 -0400 Subject: [PATCH] Add more docs --- tonic-macros/src/client.rs | 8 ++-- tonic-macros/src/lib.rs | 7 +++- tonic/src/client/grpc.rs | 69 +++++++++++++++++++++------------- tonic/src/client/mod.rs | 11 ++++++ tonic/src/client/service.rs | 48 +++++++++++++++++++++++ tonic/src/lib.rs | 38 +------------------ tonic/src/transport/channel.rs | 2 +- 7 files changed, 113 insertions(+), 70 deletions(-) create mode 100644 tonic/src/client/service.rs diff --git a/tonic-macros/src/client.rs b/tonic-macros/src/client.rs index 7eea05a..fa1ceec 100644 --- a/tonic-macros/src/client.rs +++ b/tonic-macros/src/client.rs @@ -33,7 +33,7 @@ fn generate_unary(method: &Method, proto: &str, path: String) -> TokenStream { quote! { pub async fn #ident (&mut self, request: tonic::Request<#request>) -> Result, tonic::Status> { - self.inner.ready().await?; + self.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); self.inner.unary(request, path, codec).await @@ -49,7 +49,7 @@ fn generate_server_streaming(method: &Method, proto: &str, path: String) -> Toke quote! { pub async fn #ident (&mut self, request: tonic::Request<#request>) -> Result>, tonic::Status> { - self.inner.ready().await?; + self.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); self.inner.server_streaming(request, path, codec).await @@ -67,7 +67,7 @@ fn generate_client_streaming(method: &Method, proto: &str, path: String) -> Toke -> Result, tonic::Status> where S: tonic::_codegen::Stream> + Send + 'static, { - self.inner.ready().await?; + self.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); let request = request.map(|s| Box::pin(s)); @@ -86,7 +86,7 @@ fn generate_streaming(method: &Method, proto: &str, path: String) -> TokenStream -> Result>, tonic::Status> where S: tonic::_codegen::Stream> + Send + 'static, { - self.inner.ready().await?; + self.ready().await?; let codec = tonic::codec::ProstCodec::new(); let path = http::uri::PathAndQuery::from_static(#path); let request = request.map(|s| Box::pin(s)); diff --git a/tonic-macros/src/lib.rs b/tonic-macros/src/lib.rs index eca2d0f..47dd335 100644 --- a/tonic-macros/src/lib.rs +++ b/tonic-macros/src/lib.rs @@ -26,8 +26,9 @@ pub fn client(attr: TokenStream) -> TokenStream { } impl #service_ident - where T: tonic::GrpcService, + where T: tonic::client::GrpcService, T::ResponseBody: tonic::body::Body + tonic::_codegen::HttpBody + Send + 'static, + T::Error: Into, ::Error: Into + Send, ::Data: Into + Send, { pub fn new(inner: T) -> Self { @@ -36,7 +37,9 @@ pub fn client(attr: TokenStream) -> TokenStream { } pub async fn ready(&mut self) -> Result<(), tonic::Status> { - self.inner.ready().await + self.inner.ready().await.map_err(|e| { + tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())) + }) } #methods diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index 2c11e71..5b5cfec 100644 --- a/tonic/src/client/grpc.rs +++ b/tonic/src/client/grpc.rs @@ -1,7 +1,8 @@ use crate::{ body::{Body, BoxBody}, + client::GrpcService, codec::{encode_client, Codec, Streaming}, - Code, GrpcService, Request, Response, Status, + Code, Request, Response, Status, }; use bytes::Bytes; use futures_core::Stream; @@ -11,33 +12,44 @@ use http::{ uri::{Parts, PathAndQuery, Uri}, }; use http_body::Body as HttpBody; +use std::fmt; +/// A gRPC client dispatcher. +/// +/// This will wrap some inner [`GrpcService`] and will encode/decode +/// messages via the provided codec. +/// +/// Each request method takes a [`Request`], a [`PathAndQuery`], and a +/// [`Codec`]. The request contains the message to send via the +/// [`Codec::encoder`]. The path determines the fully qualified path +/// that will be appened to the outgoing uri. The path must follow +/// the convetions explained in the [gRPC protocol definition] under `Path →`. An +/// example of this path could look like `/greeter.Greeter/SayHello`. +/// +/// [gRPC protocol definition]: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests pub struct Grpc { inner: T, } impl Grpc { + /// Creates a new gRPC client with the provided [`GrpcService`]. pub fn new(inner: T) -> Self { Self { inner } } - pub async fn ready(&mut self) -> Result<(), Status> + /// Check if the inner [`GrpcService`] is able to accept a new request. + /// + /// This will call [`GrpcService::poll_ready`] until it returns ready or + /// an error. If this returns ready the inner [`GrpcService`] is ready to + /// accept one more request. + pub async fn ready(&mut self) -> Result<(), T::Error> where T: GrpcService, - T::ResponseBody: Body + HttpBody + Send + 'static, - ::Error: Into + Send, - ::Data: Send, { - futures_util::future::poll_fn(|cx| self.inner.poll_ready(cx)) - .await - .map_err(|e| { - Status::new( - Code::Unknown, - format!("Unexpected connection error: {}", e.into()), - ) - }) + future::poll_fn(|cx| self.inner.poll_ready(cx)).await } + /// Send a single unary gRPC request. pub async fn unary( &mut self, request: Request, @@ -47,18 +59,18 @@ impl Grpc { where T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, - ::Error: Into + Send, - ::Data: Into + Send, + ::Error: Into, + ::Data: Into, C: Codec, C::Encoder: Send + 'static, C::Decoder: Send + 'static, M1: Send + 'static, - M2: Send + Unpin + 'static, { let request = request.map(|m| stream::once(future::ok(m))); self.client_streaming(request, path, codec).await } + /// Send a client side streaming gRPC request. pub async fn client_streaming( &mut self, request: Request, @@ -68,14 +80,13 @@ impl Grpc { where T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, - ::Error: Into + Send, - ::Data: Into + Send, + ::Error: Into, + ::Data: Into, S: Stream> + Send + 'static, C: Codec, C::Encoder: Send + 'static, C::Decoder: Send + 'static, M1: Send, - M2: Send + Unpin + 'static, { let (mut parts, body) = self.streaming(request, path, codec).await?.into_parts(); @@ -93,6 +104,7 @@ impl Grpc { Ok(Response::from_parts(parts, message)) } + /// Send a server side streaming gRPC request. pub async fn server_streaming( &mut self, request: Request, @@ -102,18 +114,18 @@ impl Grpc { where T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, - ::Error: Into + Send, - ::Data: Into + Send, + ::Error: Into, + ::Data: Into, C: Codec, C::Encoder: Send + 'static, C::Decoder: Send + 'static, M1: Send + 'static, - M2: Send + Unpin + 'static, { let request = request.map(|m| stream::once(future::ok(m))); self.streaming(request, path, codec).await } + /// Send a bi-directional streaming gRPC request. pub async fn streaming( &mut self, request: Request, @@ -123,14 +135,13 @@ impl Grpc { where T: GrpcService, T::ResponseBody: Body + HttpBody + Send + 'static, - ::Error: Into + Send, - ::Data: Into + Send, + ::Data: Into, + ::Error: Into, S: Stream> + Send + 'static, C: Codec, C::Encoder: Send + 'static, C::Decoder: Send + 'static, M1: Send, - M2: Send + Unpin + 'static, { let mut parts = Parts::default(); parts.path_and_query = Some(path); @@ -138,7 +149,7 @@ impl Grpc { let uri = Uri::from_parts(parts).expect("path_and_query only is valid Uri"); let request = request - .map(|s| encode_client(codec.encoder(), Box::pin(s))) + .map(|s| encode_client(codec.encoder(), s)) .map(BoxBody::new); let mut request = request.into_http(uri); @@ -194,3 +205,9 @@ impl Clone for Grpc { } } } + +impl fmt::Debug for Grpc { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.debug_struct("Grpc").finish() + } +} diff --git a/tonic/src/client/mod.rs b/tonic/src/client/mod.rs index d9a4942..2a59b7e 100644 --- a/tonic/src/client/mod.rs +++ b/tonic/src/client/mod.rs @@ -1,3 +1,14 @@ +//! gRPC over HTTP2 client implementation. +//! +//! This module contains the low level components to build a gRPC client. It +//! provides a codec agnostic gRPC client dispatcher and a decorated tower +//! service trait. +//! +//! This client is generally used by some code generation tool to provide stubs +//! for the gRPC service. Thusly, they are a bit cumbersome to use by hand. + mod grpc; +mod service; pub use self::grpc::Grpc; +pub use self::service::GrpcService; diff --git a/tonic/src/client/service.rs b/tonic/src/client/service.rs new file mode 100644 index 0000000..cd66706 --- /dev/null +++ b/tonic/src/client/service.rs @@ -0,0 +1,48 @@ +use crate::body::Body; +use http_body::Body as HttpBody; +use std::future::Future; +use std::task::{Context, Poll}; +use tower_service::Service; + +/// Definition of the gRPC trait alias for [`tower_service::Service`]. +/// +/// This trait enforces that all tower services provided to [`Grpc`] implements +/// the correct traits. +pub trait GrpcService { + /// Responses body given by the service. + type ResponseBody: Body + HttpBody; + /// Errors produced by the service. + type Error: Into; + /// The future response value. + type Future: Future, Self::Error>>; + + /// Returns `Ready` when the service is able to process requests. + /// + /// Reference [`Service::poll_ready`]. + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll>; + + /// Process the request and return the response asynchronously. + /// + /// Reference [`Service::call`]. + fn call(&mut self, request: http::Request) -> Self::Future; +} + +impl GrpcService for T +where + T: Service, Response = http::Response>, + T::Error: Into, + ResBody: Body + HttpBody, + ::Error: Into, +{ + type ResponseBody = ResBody; + type Error = T::Error; + type Future = T::Future; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + Service::poll_ready(self, cx) + } + + fn call(&mut self, request: http::Request) -> Self::Future { + Service::call(self, request) + } +} diff --git a/tonic/src/lib.rs b/tonic/src/lib.rs index 4ed7a36..84977a9 100644 --- a/tonic/src/lib.rs +++ b/tonic/src/lib.rs @@ -17,6 +17,7 @@ mod request; mod response; mod status; +#[doc(inline)] pub use body::BoxBody; pub use request::Request; pub use response::Response; @@ -25,43 +26,6 @@ pub use tonic_macros::{client, server}; pub(crate) use error::Error; -use crate::body::Body; -use http_body::Body as HttpBody; -use std::future::Future; -use std::task::{Context, Poll}; -use tower_service::Service; - -pub trait GrpcService { - type ResponseBody: Body + HttpBody; - type Error: Into; - - type Future: Future, Self::Error>>; - - fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll>; - - fn call(&mut self, request: http::Request) -> Self::Future; -} - -impl GrpcService for T -where - T: Service, Response = http::Response>, - T::Error: Into, - ResBody: Body + HttpBody, - ::Error: Into, -{ - type ResponseBody = ResBody; - type Error = T::Error; - type Future = T::Future; - - fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { - Service::poll_ready(self, cx) - } - - fn call(&mut self, request: http::Request) -> Self::Future { - Service::call(self, request) - } -} - #[doc(hidden)] pub mod _codegen { diff --git a/tonic/src/transport/channel.rs b/tonic/src/transport/channel.rs index ac42d2b..778fc8b 100644 --- a/tonic/src/transport/channel.rs +++ b/tonic/src/transport/channel.rs @@ -2,7 +2,7 @@ use super::{ service::{BoxService, Connection, ServiceList}, Endpoint, }; -use crate::{BoxBody, GrpcService}; +use crate::{client::GrpcService, BoxBody}; use futures_util::try_future::{MapErr, TryFutureExt}; use http::Uri; use hyper::{Request, Response};