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

feat(transport): Add service multiplexing/routing (#99)

Browse Source 
* feat(transport): Add service multiplexing/routing

This change introduces a new "router" built on top of
`transport::Server` that allows one to run multiple
gRPC services on the same socket.

```rust
Server::builder()
    .add_service(greeter)
    .add_service(echo)
    .serve(addr)
    .await?;
```

There is also a new `multiplex` example showcasing
server side service multiplexing and client side
service multiplexing.

BREAKING CHANGES: `Server::serve` is now crate private
and all services must be added via `Server::add_service`.
Codegen also returns just a `Service` now instead of a
`MakeService` pair.

Closes #29

Signed-off-by: Lucio Franco [email protected]
This commit is contained in:
Lucio Franco
2019-10-29 16:32:04 -04:00
committed by GitHub
parent a17049f1f7
commit 5b4f4689a2
20 changed files with 473 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tonic/src/transport/mod.rs
+20 -4
View File
@@ -50,7 +50,23 @@
//! # use futures_util::future::{err, ok};
//! # #[cfg(feature = "rustls")]
//! # async fn do_thing() -> Result<(), Box<dyn std::error::Error>> {
//! # let my_svc = service_fn(|_| ok::<_, tonic::Status>(service_fn(|req| err(tonic::Status::unimplemented("")))));
//! # #[derive(Clone)]
//! # pub struct Svc;
//! # impl Service<hyper::Request<hyper::Body>> for Svc {
//! # type Response = hyper::Response<tonic::body::BoxBody>;
//! # type Error = tonic::Status;
//! # type Future = futures_util::future::Ready<Result<Self::Response, Self::Error>>;
//! # fn poll_ready(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
//! # Ok(()).into()
//! # }
//! # fn call(&mut self, _req: hyper::Request<hyper::Body>) -> Self::Future {
//! # unimplemented!()
//! # }
//! # }
//! # impl tonic::transport::ServiceName for Svc {
//! # const NAME: &'static str = "some_svc";
//! # }
//! # let my_svc = Svc;
//! let cert = std::fs::read_to_string("server.pem")?;
//! let key = std::fs::read_to_string("server.key")?;
//!
@@ -64,8 +80,8 @@
//! println!("Request: {:?}", req);
//! svc.call(req)
//! })
//! .clone()
//! .serve(addr, my_svc)
//! .add_service(my_svc)
//! .serve(addr)
//! .await?;
//!
//! # Ok(())
@@ -88,7 +104,7 @@ pub use self::channel::Channel;
pub use self::endpoint::Endpoint;
pub use self::error::Error;
#[doc(inline)]
pub use self::server::Server;
pub use self::server::{Server, ServiceName};
pub use self::tls::{Certificate, Identity};
pub use hyper::Body;
tonic/src/transport/server.rs
+138 -22
View File
@@ -1,6 +1,6 @@
//! Server implementation and builder.
use super::service::{layer_fn, BoxedIo, ServiceBuilderExt};
use super::service::{layer_fn, BoxedIo, Or, Routes, ServiceBuilderExt};
#[cfg(feature = "tls")]
use super::{
service::TlsAcceptor,
@@ -9,7 +9,7 @@ use super::{
};
use crate::body::BoxBody;
use futures_core::Stream;
use futures_util::{ready, try_future::MapErr, TryFutureExt, TryStreamExt};
use futures_util::{future, ready, try_future::MapErr, TryFutureExt, TryStreamExt};
use http::{Request, Response};
use hyper::{
server::{accept::Accept, conn},
@@ -31,7 +31,6 @@ use tower::{
Service,
ServiceBuilder,
};
use tower_make::MakeService;
#[cfg(feature = "tls")]
use tracing::error;
@@ -58,6 +57,22 @@ pub struct Server {
max_concurrent_streams: Option<u32>,
}
/// A stack based `Service` router.
#[derive(Debug)]
pub struct Router<A, B> {
server: Server,
routes: Routes<A, B, Request<Body>>,
}
/// A trait to provide a static reference to the service's
/// name. This is used for routing service's within the router.
pub trait ServiceName {
/// The `Service-Name` as described [here].
///
/// [here]: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
const NAME: &'static str;
}
impl Server {
/// Create a new server builder that can configure a [`Server`].
pub fn builder() -> Self {
@@ -149,14 +164,26 @@ impl Server {
self
}
/// Consume this [`Server`] creating a future that will execute the server
/// on [`tokio`]'s default executor.
pub async fn serve<M, S>(self, addr: SocketAddr, svc: M) -> Result<(), super::Error>
/// Create a router with the `S` typed service as the first service.
///
/// This will clone the `Server` builder and create a router that will
/// route around different services.
pub fn add_service<S>(&mut self, svc: S) -> Router<S, Unimplemented>
where
M: Service<(), Response = S>,
M::Error: Into<crate::Error> + Send + 'static,
M::Future: Send + 'static,
S: Service<Request<Body>, Response = Response<BoxBody>> + Send + 'static,
S: Service<Request<Body>, Response = Response<BoxBody>>
+ ServiceName
+ Clone
+ Send
+ 'static,
S::Future: Send + 'static,
S::Error: Into<crate::Error> + Send,
{
Router::new(self.clone(), svc)
}
pub(crate) async fn serve<S>(self, addr: SocketAddr, svc: S) -> Result<(), super::Error>
where
S: Service<Request<Body>, Response = Response<BoxBody>> + Clone + Send + 'static,
S::Future: Send + 'static,
S::Error: Into<crate::Error> + Send,
{
@@ -210,6 +237,74 @@ impl Server {
}
}
impl<S> Router<S, Unimplemented> {
pub(crate) fn new(server: Server, svc: S) -> Self
where
S: Service<Request<Body>, Response = Response<BoxBody>>
+ ServiceName
+ Clone
+ Send
+ 'static,
S::Future: Send + 'static,
S::Error: Into<crate::Error> + Send,
{
let svc_name = <S as ServiceName>::NAME;
let svc_route = format!("/{}", svc_name);
let pred = move |req: &Request<Body>| {
let path = req.uri().path();
path.starts_with(&svc_route)
};
Self {
server,
routes: Routes::new(pred, svc, Unimplemented::default()),
}
}
}
impl<A, B> Router<A, B>
where
A: Service<Request<Body>, Response = Response<BoxBody>> + Clone + Send + 'static,
A::Future: Send + 'static,
A::Error: Into<crate::Error> + Send,
B: Service<Request<Body>, Response = Response<BoxBody>> + Clone + Send + 'static,
B::Future: Send + 'static,
B::Error: Into<crate::Error> + Send,
{
/// Add a new service to this router.
pub fn add_service<S>(self, svc: S) -> Router<S, Or<A, B, Request<Body>>>
where
S: Service<Request<Body>, Response = Response<BoxBody>>
+ ServiceName
+ Clone
+ Send
+ 'static,
S::Future: Send + 'static,
S::Error: Into<crate::Error> + Send,
{
let Self { routes, server } = self;
let svc_name = <S as ServiceName>::NAME;
let svc_route = format!("/{}", svc_name);
let pred = move |req: &Request<Body>| {
let path = req.uri().path();
path.starts_with(&svc_route)
};
let routes = routes.push(pred, svc);
Router { server, routes }
}
/// Consume this [`Server`] creating a future that will execute the server
/// on [`tokio`]'s default executor.
///
/// [`Server`]: struct.Server.html
pub async fn serve(self, addr: SocketAddr) -> Result<(), super::Error> {
self.server.serve(addr, self.routes).await
}
}
fn map_err(e: impl Into<crate::Error>) -> super::Error {
super::Error::from_source(super::ErrorKind::Server, e.into())
}
@@ -371,19 +466,16 @@ where
}
}
struct MakeSvc<M> {
struct MakeSvc<S> {
interceptor: Option<Interceptor>,
concurrency_limit: Option<usize>,
// timeout: Option<Duration>,
inner: M,
inner: S,
}
impl<M, S, T> Service<T> for MakeSvc<M>
impl<S, T> Service<T> for MakeSvc<S>
where
M: Service<(), Response = S>,
M::Error: Into<crate::Error> + Send,
M::Future: Send + 'static,
S: Service<Request<Body>, Response = Response<BoxBody>> + Send + 'static,
S: Service<Request<Body>, Response = Response<BoxBody>> + Clone + Send + 'static,
S::Future: Send + 'static,
S::Error: Into<crate::Error> + Send,
{
@@ -392,19 +484,17 @@ where
type Future =
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
MakeService::poll_ready(&mut self.inner, cx).map_err(Into::into)
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Ok(()).into()
}
fn call(&mut self, _: T) -> Self::Future {
let interceptor = self.interceptor.clone();
let make = self.inner.make_service(());
let svc = self.inner.clone();
let concurrency_limit = self.concurrency_limit;
// let timeout = self.timeout.clone();
Box::pin(async move {
let svc = make.await.map_err(Into::into)?;
let svc = ServiceBuilder::new()
.optional_layer(concurrency_limit.map(ConcurrencyLimitLayer::new))
// .optional_layer(timeout.map(TimeoutLayer::new))
@@ -421,3 +511,29 @@ where
})
}
}
#[derive(Default, Clone, Debug)]
#[doc(hidden)]
pub struct Unimplemented {
_p: (),
}
impl Service<Request<Body>> for Unimplemented {
type Response = Response<BoxBody>;
type Error = crate::Error;
type Future = future::Ready<Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Ok(()).into()
}
fn call(&mut self, _req: Request<Body>) -> Self::Future {
future::ok(
http::Response::builder()
.status(200)
.header("grpc-status", "12")
.body(BoxBody::empty())
.unwrap(),
)
}
}
tonic/src/transport/service/mod.rs
+2
View File
@@ -5,6 +5,7 @@ mod discover;
mod either;
mod io;
mod layer;
mod router;
#[cfg(feature = "tls")]
mod tls;
@@ -14,5 +15,6 @@ pub(crate) use self::connector::connector;
pub(crate) use self::discover::ServiceList;
pub(crate) use self::io::BoxedIo;
pub(crate) use self::layer::{layer_fn, ServiceBuilderExt};
pub(crate) use self::router::{Or, Routes};
#[cfg(feature = "tls")]
pub(crate) use self::tls::{TlsAcceptor, TlsConnector};
tonic/src/transport/service/router.rs
+129
View File
@@ -0,0 +1,129 @@
use futures_util::{
future::Either,
try_future::{MapErr, TryFutureExt},
};
use std::{
fmt,
sync::Arc,
task::{Context, Poll},
};
use tower_service::Service;
#[derive(Debug)]
pub(crate) struct Routes<A, B, Request> {
routes: Or<A, B, Request>,
}
impl<A, B, Request> Routes<A, B, Request> {
pub(crate) fn new(
predicate: impl Fn(&Request) -> bool + Send + Sync + 'static,
a: A,
b: B,
) -> Self {
let routes = Or::new(predicate, a, b);
Self { routes }
}
}
impl<A, B, Request> Routes<A, B, Request> {
pub(crate) fn push<C>(
self,
predicate: impl Fn(&Request) -> bool + Send + Sync + 'static,
route: C,
) -> Routes<C, Or<A, B, Request>, Request> {
let routes = Or::new(predicate, route, self.routes);
Routes { routes }
}
}
impl<A, B, Request> Service<Request> for Routes<A, B, Request>
where
A: Service<Request>,
A::Future: Send + 'static,
A::Error: Into<crate::Error>,
B: Service<Request, Response = A::Response>,
B::Future: Send + 'static,
B::Error: Into<crate::Error>,
{
type Response = A::Response;
type Error = crate::Error;
type Future = <Or<A, B, Request> as Service<Request>>::Future;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Ok(()).into()
}
fn call(&mut self, req: Request) -> Self::Future {
self.routes.call(req)
}
}
impl<A: Clone, B: Clone, Request> Clone for Routes<A, B, Request> {
fn clone(&self) -> Self {
Self {
routes: self.routes.clone(),
}
}
}
#[doc(hidden)]
pub struct Or<A, B, Request> {
predicate: Arc<dyn Fn(&Request) -> bool + Send + Sync + 'static>,
a: A,
b: B,
}
impl<A, B, Request> Or<A, B, Request> {
pub(crate) fn new<F>(predicate: F, a: A, b: B) -> Self
where
F: Fn(&Request) -> bool + Send + Sync + 'static,
{
let predicate = Arc::new(predicate);
Self { predicate, a, b }
}
}
impl<A, B, Request> Service<Request> for Or<A, B, Request>
where
A: Service<Request>,
A::Future: Send + 'static,
A::Error: Into<crate::Error>,
B: Service<Request, Response = A::Response>,
B::Future: Send + 'static,
B::Error: Into<crate::Error>,
{
type Response = A::Response;
type Error = crate::Error;
type Future = Either<
MapErr<A::Future, fn(A::Error) -> crate::Error>,
MapErr<B::Future, fn(B::Error) -> crate::Error>,
>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Ok(()).into()
}
fn call(&mut self, req: Request) -> Self::Future {
if (self.predicate)(&req) {
Either::Left(self.a.call(req).map_err(|e| e.into()))
} else {
Either::Right(self.b.call(req).map_err(|e| e.into()))
}
}
}
impl<A: Clone, B: Clone, Request> Clone for Or<A, B, Request> {
fn clone(&self) -> Self {
Self {
predicate: self.predicate.clone(),
a: self.a.clone(),
b: self.b.clone(),
}
}
}
impl<A, B, Request> fmt::Debug for Or<A, B, Request> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "Or {{ .. }}")
}
}
tonic/src/transport/service/tls.rs
+1 -2
View File
@@ -14,7 +14,6 @@ use tokio_rustls::{
webpki::DNSNameRef,
TlsAcceptor as RustlsAcceptor, TlsConnector as RustlsConnector,
};
use tracing::trace;
/// h2 alpn in wire format for openssl.
#[cfg(feature = "openssl")]
@@ -137,7 +136,7 @@ impl TlsConnector {
let tls = tokio_openssl::connect(config, &self.domain, io).await?;
match tls.ssl().selected_alpn_protocol() {
Some(b) if b == b"h2" => trace!("HTTP/2 succesfully negotiated."),
Some(b) if b == b"h2" => tracing::trace!("HTTP/2 succesfully negotiated."),
_ => return Err(TlsError::H2NotNegotiated.into()),
};