talksik/tonic
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add more docs

Browse Source
This commit is contained in:
Lucio Franco
2019-09-03 23:53:42 -04:00
parent e48131bfbe
commit e00ddc920f
7 changed files with 113 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tonic-macros/src/client.rs
+4 -4
View File
@@ -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::Response<#response>, 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::Response<tonic::codec::Streaming<#response>>, 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::Response<#response>, tonic::Status>
where S: tonic::_codegen::Stream<Item = Result<#request, tonic::Status>> + 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::Response<tonic::codec::Streaming<#response>>, tonic::Status>
where S: tonic::_codegen::Stream<Item = Result<#request, tonic::Status>> + 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));
tonic-macros/src/lib.rs
+5 -2
View File
@@ -26,8 +26,9 @@ pub fn client(attr: TokenStream) -> TokenStream {
}
impl<T> #service_ident <T>
where T: tonic::GrpcService<tonic::body::BoxBody>,
where T: tonic::client::GrpcService<tonic::BoxBody>,
T::ResponseBody: tonic::body::Body + tonic::_codegen::HttpBody + Send + 'static,
T::Error: Into<tonic::error::Error>,
<T::ResponseBody as tonic::_codegen::HttpBody>::Error: Into<tonic::error::Error> + Send,
<T::ResponseBody as tonic::_codegen::HttpBody>::Data: Into<bytes::Bytes> + 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
tonic/src/client/grpc.rs
+43 -26
View File
@@ -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<T> {
inner: T,
}
impl<T> Grpc<T> {
/// 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<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::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<M1, M2, C>(
&mut self,
request: Request<M1>,
@@ -47,18 +59,18 @@ impl<T> Grpc<T> {
where
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
<T::ResponseBody as HttpBody>::Data: Into<Bytes>,
C: Codec<Encode = M1, Decode = M2>,
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<S, M1, M2, C>(
&mut self,
request: Request<S>,
@@ -68,14 +80,13 @@ impl<T> Grpc<T> {
where
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
<T::ResponseBody as HttpBody>::Data: Into<Bytes>,
S: Stream<Item = Result<M1, Status>> + Send + 'static,
C: Codec<Encode = M1, Decode = M2>,
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<T> Grpc<T> {
Ok(Response::from_parts(parts, message))
}
/// Send a server side streaming gRPC request.
pub async fn server_streaming<M1, M2, C>(
&mut self,
request: Request<M1>,
@@ -102,18 +114,18 @@ impl<T> Grpc<T> {
where
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
<T::ResponseBody as HttpBody>::Data: Into<Bytes>,
C: Codec<Encode = M1, Decode = M2>,
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<S, M1, M2, C>(
&mut self,
request: Request<S>,
@@ -123,14 +135,13 @@ impl<T> Grpc<T> {
where
T: GrpcService<BoxBody>,
T::ResponseBody: Body + HttpBody + Send + 'static,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error> + Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes> + Send,
<T::ResponseBody as HttpBody>::Data: Into<Bytes>,
<T::ResponseBody as HttpBody>::Error: Into<crate::Error>,
S: Stream<Item = Result<M1, Status>> + Send + 'static,
C: Codec<Encode = M1, Decode = M2>,
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<T> Grpc<T> {
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<T: Clone> Clone for Grpc<T> {
}
}
}
impl<T> fmt::Debug for Grpc<T> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Grpc").finish()
}
}
tonic/src/client/mod.rs
+11
View File
@@ -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;
tonic/src/client/service.rs
+48
View File
@@ -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<ReqBody> {
/// Responses body given by the service.
type ResponseBody: Body + HttpBody;
/// Errors produced by the service.
type Error: Into<crate::Error>;
/// The future response value.
type Future: Future<Output = Result<http::Response<Self::ResponseBody>, 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<Result<(), Self::Error>>;
/// Process the request and return the response asynchronously.
///
/// Reference [`Service::call`].
fn call(&mut self, request: http::Request<ReqBody>) -> Self::Future;
}
impl<T, ReqBody, ResBody> GrpcService<ReqBody> for T
where
T: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
T::Error: Into<crate::Error>,
ResBody: Body + HttpBody,
<ResBody as HttpBody>::Error: Into<crate::Error>,
{
type ResponseBody = ResBody;
type Error = T::Error;
type Future = T::Future;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Service::poll_ready(self, cx)
}
fn call(&mut self, request: http::Request<ReqBody>) -> Self::Future {
Service::call(self, request)
}
}
tonic/src/lib.rs
+1 -37
View File
@@ -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<ReqBody> {
type ResponseBody: Body + HttpBody;
type Error: Into<crate::Error>;
type Future: Future<Output = Result<http::Response<Self::ResponseBody>, Self::Error>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
fn call(&mut self, request: http::Request<ReqBody>) -> Self::Future;
}
impl<T, ReqBody, ResBody> GrpcService<ReqBody> for T
where
T: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
T::Error: Into<crate::Error>,
ResBody: Body + HttpBody,
<ResBody as HttpBody>::Error: Into<crate::Error>,
{
type ResponseBody = ResBody;
type Error = T::Error;
type Future = T::Future;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Service::poll_ready(self, cx)
}
fn call(&mut self, request: http::Request<ReqBody>) -> Self::Future {
Service::call(self, request)
}
}
#[doc(hidden)]
pub mod _codegen {
tonic/src/transport/channel.rs
+1 -1
View File
@@ -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};