diff --git a/tonic-examples/Cargo.toml b/tonic-examples/Cargo.toml index 09953c4..d8617d7 100644 --- a/tonic-examples/Cargo.toml +++ b/tonic-examples/Cargo.toml @@ -24,7 +24,6 @@ path = "src/routeguide/client.rs" [dependencies] tonic = { path = "../tonic" } -hyper = { git = "https://github.com/hyperium/hyper" } futures-preview = { version = "=0.3.0-alpha.18", default-features = false, features = ["alloc"]} tokio = "=0.2.0-alpha.4" prost = "0.5" diff --git a/tonic-examples/src/routeguide/server.rs b/tonic-examples/src/routeguide/server.rs index 645e97f..323e287 100644 --- a/tonic-examples/src/routeguide/server.rs +++ b/tonic-examples/src/routeguide/server.rs @@ -1,12 +1,12 @@ mod data; use futures::{Stream, StreamExt}; -use hyper::Server; use std::collections::HashMap; use std::hash::{Hash, Hasher}; use std::sync::Arc; use std::time::Instant; use tokio::sync::{mpsc, Lock}; +use tonic::transport::Server; use tonic::{Request, Response, Status}; pub mod routeguide { @@ -159,9 +159,8 @@ async fn main() -> Result<(), Box> { }, }; - Server::bind(&addr) - .http2_only(true) - .serve(RouteGuideServer::new(route_guide)) + Server::builder() + .serve(addr, RouteGuideServer::new(route_guide)) .await?; Ok(()) diff --git a/tonic-macros/src/service.rs b/tonic-macros/src/service.rs index 553cfe4..fa108a1 100644 --- a/tonic-macros/src/service.rs +++ b/tonic-macros/src/service.rs @@ -106,7 +106,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream { } } - impl Service> for #server_service { + impl Service> for #server_service { type Response = http::Response; type Error = tonic::error::Never; type Future = BoxFuture; @@ -115,7 +115,7 @@ pub(crate) fn generate(service: ServiceDef) -> TokenStream { Poll::Ready(Ok(())) } - fn call(&mut self, req: http::Request) -> Self::Future { + fn call(&mut self, req: http::Request) -> Self::Future { let inner = self.inner.clone(); match req.uri().path() { diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 7cc68c8..a0ff2fd 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -21,7 +21,7 @@ http-body = "0.2.0-alpha.1" pin-project = "0.4.0-alpha.7" # 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 } tower-make = "=0.1.0-alpha.2" tower-reconnect = { git = "https://github.com/tower-rs/tower", branch = "lucio/update-reconnect-buffer", optional = true } diff --git a/tonic/src/lib.rs b/tonic/src/lib.rs index 84977a9..fc60bf5 100644 --- a/tonic/src/lib.rs +++ b/tonic/src/lib.rs @@ -37,6 +37,9 @@ pub mod _codegen { pub use std::task::{Context, Poll}; pub use tower_service::Service; + #[cfg(feature = "transport")] + pub use hyper::Body as HyperBody; + pub type BoxFuture = self::Pin> + Send + 'static>>; pub type BoxStream = diff --git a/tonic/src/server/grpc.rs b/tonic/src/server/grpc.rs index a0639da..b55ee38 100644 --- a/tonic/src/server/grpc.rs +++ b/tonic/src/server/grpc.rs @@ -1,14 +1,22 @@ use crate::{ - body::BoxBody, codec::{encode_server, Codec, Streaming}, server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService}, - Code, Request, Response, Status, + BoxBody, Code, Request, Response, Status, }; use bytes::Bytes; use futures_core::TryStream; use futures_util::{future, stream, TryStreamExt}; 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 { codec: T, } @@ -21,10 +29,12 @@ where T::Encoder: Send + 'static, T::Encode: Send + Unpin + 'static, { + /// Creates a new gRPC client with the provided [`Codec`]. pub fn new(codec: T) -> Self { Self { codec } } + /// Handle a single unary gRPC request. pub async fn unary( &mut self, mut service: S, @@ -54,6 +64,7 @@ where self.map_response(response) } + // Handle a server side streaming request. pub async fn server_streaming( &mut self, mut service: S, @@ -78,7 +89,7 @@ where self.map_response(response) } - //BoxStream, + /// Handle a client side streaming gRPC request. pub async fn client_streaming( &mut self, mut service: S, @@ -100,6 +111,7 @@ where self.map_response(response) } + /// Handle a bi-directional streaming gRPC request. pub async fn streaming( &mut self, mut service: S, diff --git a/tonic/src/server/mod.rs b/tonic/src/server/mod.rs index da4c64f..319f5ff 100644 --- a/tonic/src/server/mod.rs +++ b/tonic/src/server/mod.rs @@ -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 service; diff --git a/tonic/src/server/service.rs b/tonic/src/server/service.rs index a9b956e..39f5ace 100644 --- a/tonic/src/server/service.rs +++ b/tonic/src/server/service.rs @@ -1,7 +1,12 @@ use crate::{Request, Response, Status}; use futures_core::Stream; 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 { /// Protobuf response message type type Response; @@ -13,12 +18,28 @@ pub trait UnaryService { fn call(&mut self, request: Request) -> Self::Future; } +impl UnaryService for T +where + T: Service, Response = Response, Error = crate::Status>, +{ + type Response = M2; + type Future = T::Future; + + fn call(&mut self, request: Request) -> 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 { /// Protobuf response message type type Response; /// Stream of outbound response messages - type ResponseStream: Stream> + Unpin; + type ResponseStream: Stream>; /// Response future type Future: Future, Status>>; @@ -27,6 +48,24 @@ pub trait ServerStreamingService { fn call(&mut self, request: Request) -> Self::Future; } +impl ServerStreamingService for T +where + T: Service, Response = Response, Error = crate::Status>, + S: Stream>, +{ + type Response = M2; + type ResponseStream = S; + type Future = T::Future; + + fn call(&mut self, request: Request) -> 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 { /// Protobuf response message type type Response; @@ -38,12 +77,29 @@ pub trait ClientStreamingService { fn call(&mut self, request: Request) -> Self::Future; } +impl ClientStreamingService for T +where + T: Service, Response = Response, Error = crate::Status>, + S: Stream>, +{ + type Response = M2; + type Future = T::Future; + + fn call(&mut self, request: Request) -> 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 { /// Protobuf response message type type Response; /// Stream of outbound response messages - type ResponseStream: Stream> + Unpin; + type ResponseStream: Stream>; /// Response future type Future: Future, Status>>; @@ -51,3 +107,18 @@ pub trait StreamingService { /// Call the service fn call(&mut self, request: Request) -> Self::Future; } + +impl StreamingService for T +where + T: Service, Response = Response, Error = crate::Status>, + S1: Stream>, + S2: Stream>, +{ + type Response = M2; + type ResponseStream = S2; + type Future = T::Future; + + fn call(&mut self, request: Request) -> Self::Future { + Service::call(self, request) + } +}