Add server docs

This commit is contained in:
Lucio Franco
2019-09-04 16:38:46 -04:00
parent 2a01ab229a
commit 77a6c89ae9
8 changed files with 107 additions and 13 deletions
-1
View File
@@ -24,7 +24,6 @@ path = "src/routeguide/client.rs"
[dependencies] [dependencies]
tonic = { path = "../tonic" } tonic = { path = "../tonic" }
hyper = { git = "https://github.com/hyperium/hyper" }
futures-preview = { version = "=0.3.0-alpha.18", default-features = false, features = ["alloc"]} futures-preview = { version = "=0.3.0-alpha.18", default-features = false, features = ["alloc"]}
tokio = "=0.2.0-alpha.4" tokio = "=0.2.0-alpha.4"
prost = "0.5" prost = "0.5"
+3 -4
View File
@@ -1,12 +1,12 @@
mod data; mod data;
use futures::{Stream, StreamExt}; use futures::{Stream, StreamExt};
use hyper::Server;
use std::collections::HashMap; use std::collections::HashMap;
use std::hash::{Hash, Hasher}; use std::hash::{Hash, Hasher};
use std::sync::Arc; use std::sync::Arc;
use std::time::Instant; use std::time::Instant;
use tokio::sync::{mpsc, Lock}; use tokio::sync::{mpsc, Lock};
use tonic::transport::Server;
use tonic::{Request, Response, Status}; use tonic::{Request, Response, Status};
pub mod routeguide { pub mod routeguide {
@@ -159,9 +159,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}, },
}; };
Server::bind(&addr) Server::builder()
.http2_only(true) .serve(addr, RouteGuideServer::new(route_guide))
.serve(RouteGuideServer::new(route_guide))
.await?; .await?;
Ok(()) Ok(())
+2 -2
View File
@@ -106,7 +106,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream {
} }
} }
impl Service<http::Request<hyper::Body>> for #server_service { impl Service<http::Request<tonic::_codegen::HyperBody>> for #server_service {
type Response = http::Response<tonic::BoxBody>; type Response = http::Response<tonic::BoxBody>;
type Error = tonic::error::Never; type Error = tonic::error::Never;
type Future = BoxFuture<Self::Response, Self::Error>; type Future = BoxFuture<Self::Response, Self::Error>;
@@ -115,7 +115,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream {
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
} }
fn call(&mut self, req: http::Request<hyper::Body>) -> Self::Future { fn call(&mut self, req: http::Request<tonic::_codegen::HyperBody>) -> Self::Future {
let inner = self.inner.clone(); let inner = self.inner.clone();
match req.uri().path() { match req.uri().path() {
+1 -1
View File
@@ -21,7 +21,7 @@ http-body = "0.2.0-alpha.1"
pin-project = "0.4.0-alpha.7" pin-project = "0.4.0-alpha.7"
# optional # optional
hyper = { git = "https://github.com/hyperium/hyper", optional = true} hyper = { version = "=0.13.0-alpha.1", optional = true}
tokio = { version = "=0.2.0-alpha.4", default-features = false, features = ["tcp"], optional = true } tokio = { version = "=0.2.0-alpha.4", default-features = false, features = ["tcp"], optional = true }
tower-make = "=0.1.0-alpha.2" tower-make = "=0.1.0-alpha.2"
tower-reconnect = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-reconnect-buffer", optional = true } tower-reconnect = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-reconnect-buffer", optional = true }
+3
View File
@@ -37,6 +37,9 @@ pub mod _codegen {
pub use std::task::{Context, Poll}; pub use std::task::{Context, Poll};
pub use tower_service::Service; pub use tower_service::Service;
#[cfg(feature = "transport")]
pub use hyper::Body as HyperBody;
pub type BoxFuture<T, E> = pub type BoxFuture<T, E> =
self::Pin<Box<dyn self::Future<Output = Result<T, E>> + Send + 'static>>; self::Pin<Box<dyn self::Future<Output = Result<T, E>> + Send + 'static>>;
pub type BoxStream<T> = pub type BoxStream<T> =
+15 -3
View File
@@ -1,14 +1,22 @@
use crate::{ use crate::{
body::BoxBody,
codec::{encode_server, Codec, Streaming}, codec::{encode_server, Codec, Streaming},
server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService}, server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService},
Code, Request, Response, Status, BoxBody, Code, Request, Response, Status,
}; };
use bytes::Bytes; use bytes::Bytes;
use futures_core::TryStream; use futures_core::TryStream;
use futures_util::{future, stream, TryStreamExt}; use futures_util::{future, stream, TryStreamExt};
use http_body::Body; use http_body::Body;
/// A gRPC Server handler.
///
/// This will wrap some inner [`Codec`] and provide utilities to handle
/// inbound unary, client side streaming, server side streaming, and
/// bi-directional streaming.
///
/// Each request handler method accepts some service that implements the
/// corresponding service trait and a http request that contains some body that
/// implements some [`Body`].
pub struct Grpc<T> { pub struct Grpc<T> {
codec: T, codec: T,
} }
@@ -21,10 +29,12 @@ where
T::Encoder: Send + 'static, T::Encoder: Send + 'static,
T::Encode: Send + Unpin + 'static, T::Encode: Send + Unpin + 'static,
{ {
/// Creates a new gRPC client with the provided [`Codec`].
pub fn new(codec: T) -> Self { pub fn new(codec: T) -> Self {
Self { codec } Self { codec }
} }
/// Handle a single unary gRPC request.
pub async fn unary<S, B>( pub async fn unary<S, B>(
&mut self, &mut self,
mut service: S, mut service: S,
@@ -54,6 +64,7 @@ where
self.map_response(response) self.map_response(response)
} }
// Handle a server side streaming request.
pub async fn server_streaming<S, B>( pub async fn server_streaming<S, B>(
&mut self, &mut self,
mut service: S, mut service: S,
@@ -78,7 +89,7 @@ where
self.map_response(response) self.map_response(response)
} }
//BoxStream<T::Decode>, /// Handle a client side streaming gRPC request.
pub async fn client_streaming<S, B>( pub async fn client_streaming<S, B>(
&mut self, &mut self,
mut service: S, mut service: S,
@@ -100,6 +111,7 @@ where
self.map_response(response) self.map_response(response)
} }
/// Handle a bi-directional streaming gRPC request.
pub async fn streaming<S, B>( pub async fn streaming<S, B>(
&mut self, &mut self,
mut service: S, mut service: S,
+10
View File
@@ -1,3 +1,13 @@
//! gRPC over HTTP2 server implementation.
//!
//! This module contains the low level components to build a gRPC server. It
//! provides a codec agnostic gRPC server handler.
//!
//! The items in this module are generally desgined to be used by some codegen
//! tool that will provide the user some custom way to implement the server that
//! will implement the proper gRPC service. Thusly, they are a bit hard to use
//! by hand.
mod grpc; mod grpc;
mod service; mod service;
+73 -2
View File
@@ -1,7 +1,12 @@
use crate::{Request, Response, Status}; use crate::{Request, Response, Status};
use futures_core::Stream; use futures_core::Stream;
use std::future::Future; use std::future::Future;
use tower_service::Service;
/// A specialization of tower_service::Service.
///
/// Existing tower_service::Service implementations with the correct form will
/// automatically implement `UnaryService`.
pub trait UnaryService<R> { pub trait UnaryService<R> {
/// Protobuf response message type /// Protobuf response message type
type Response; type Response;
@@ -13,12 +18,28 @@ pub trait UnaryService<R> {
fn call(&mut self, request: Request<R>) -> Self::Future; fn call(&mut self, request: Request<R>) -> Self::Future;
} }
impl<T, M1, M2> UnaryService<M1> for T
where
T: Service<Request<M1>, Response = Response<M2>, Error = crate::Status>,
{
type Response = M2;
type Future = T::Future;
fn call(&mut self, request: Request<M1>) -> Self::Future {
Service::call(self, request)
}
}
/// A specialization of tower_service::Service.
///
/// Existing tower_service::Service implementations with the correct form will
/// automatically implement `ServerStreamingService`.
pub trait ServerStreamingService<R> { pub trait ServerStreamingService<R> {
/// Protobuf response message type /// Protobuf response message type
type Response; type Response;
/// Stream of outbound response messages /// Stream of outbound response messages
type ResponseStream: Stream<Item = Result<Self::Response, Status>> + Unpin; type ResponseStream: Stream<Item = Result<Self::Response, Status>>;
/// Response future /// Response future
type Future: Future<Output = Result<Response<Self::ResponseStream>, Status>>; type Future: Future<Output = Result<Response<Self::ResponseStream>, Status>>;
@@ -27,6 +48,24 @@ pub trait ServerStreamingService<R> {
fn call(&mut self, request: Request<R>) -> Self::Future; fn call(&mut self, request: Request<R>) -> Self::Future;
} }
impl<T, S, M1, M2> ServerStreamingService<M1> for T
where
T: Service<Request<M1>, Response = Response<S>, Error = crate::Status>,
S: Stream<Item = Result<M2, crate::Status>>,
{
type Response = M2;
type ResponseStream = S;
type Future = T::Future;
fn call(&mut self, request: Request<M1>) -> Self::Future {
Service::call(self, request)
}
}
/// A specialization of tower_service::Service.
///
/// Existing tower_service::Service implementations with the correct form will
/// automatically implement `ClientStreamingService`.
pub trait ClientStreamingService<RequestStream> { pub trait ClientStreamingService<RequestStream> {
/// Protobuf response message type /// Protobuf response message type
type Response; type Response;
@@ -38,12 +77,29 @@ pub trait ClientStreamingService<RequestStream> {
fn call(&mut self, request: Request<RequestStream>) -> Self::Future; fn call(&mut self, request: Request<RequestStream>) -> Self::Future;
} }
impl<T, M1, M2, S> ClientStreamingService<S> for T
where
T: Service<Request<S>, Response = Response<M2>, Error = crate::Status>,
S: Stream<Item = Result<M1, crate::Status>>,
{
type Response = M2;
type Future = T::Future;
fn call(&mut self, request: Request<S>) -> Self::Future {
Service::call(self, request)
}
}
/// A specialization of tower_service::Service.
///
/// Existing tower_service::Service implementations with the correct form will
/// automatically implement `StreamingService`.
pub trait StreamingService<RequestStream> { pub trait StreamingService<RequestStream> {
/// Protobuf response message type /// Protobuf response message type
type Response; type Response;
/// Stream of outbound response messages /// Stream of outbound response messages
type ResponseStream: Stream<Item = Result<Self::Response, Status>> + Unpin; type ResponseStream: Stream<Item = Result<Self::Response, Status>>;
/// Response future /// Response future
type Future: Future<Output = Result<Response<Self::ResponseStream>, Status>>; type Future: Future<Output = Result<Response<Self::ResponseStream>, Status>>;
@@ -51,3 +107,18 @@ pub trait StreamingService<RequestStream> {
/// Call the service /// Call the service
fn call(&mut self, request: Request<RequestStream>) -> Self::Future; fn call(&mut self, request: Request<RequestStream>) -> Self::Future;
} }
impl<T, S1, S2, M1, M2> StreamingService<S1> for T
where
T: Service<Request<S1>, Response = Response<S2>, Error = crate::Status>,
S1: Stream<Item = Result<M1, crate::Status>>,
S2: Stream<Item = Result<M2, crate::Status>>,
{
type Response = M2;
type ResponseStream = S2;
type Future = T::Future;
fn call(&mut self, request: Request<S1>) -> Self::Future {
Service::call(self, request)
}
}