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

Add docs to transport and refactor some api's

Browse Source
This commit is contained in:
Lucio Franco
2019-09-29 17:04:42 -04:00
parent 1d9e3daa07
commit a35ef96a03
16 changed files with 485 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/test.yml
+1 -1
View File
@@ -28,4 +28,4 @@ jobs:
- name: Check with transport w/ rustls
run: cargo check -p tonic --features rustls
- name: Run tests
run: cargo test --all
run: cargo test --all --all-features
tonic-build/src/client.rs
+2 -8
View File
@@ -1,4 +1,4 @@
use crate::{generate_doc_comment, generate_doc_comments};
use crate::generate_doc_comments;
use proc_macro2::TokenStream;
use prost_build::{Method, Service};
use quote::{format_ident, quote};
@@ -52,12 +52,6 @@ pub(crate) fn generate(service: &Service, proto: &str) -> TokenStream {
#[cfg(feature = "transport")]
fn generate_connect(service_ident: &syn::Ident) -> TokenStream {
let doc_example = format!(
"let client = {}::connect(\"http://[::1]:50051\")?;",
service_ident
);
let doc_example = generate_doc_comment(&doc_example);
quote! {
impl #service_ident<tonic::transport::Channel> {
/// Attempt to create a new client by connecting to a given endpoint.
@@ -66,7 +60,7 @@ fn generate_connect(service_ident: &syn::Ident) -> TokenStream {
D: std::convert::TryInto<tonic::transport::Endpoint>,
D::Error: Into<StdError>,
{
tonic::transport::Channel::builder().build(dst).map(|c| Self::new(c))
tonic::transport::Endpoint::new(dst).map(|c| Self::new(c.channel()))
}
}
}
tonic-interop/src/bin/client.rs
+1 -1
View File
@@ -36,7 +36,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
endpoint.openssl_tls(ca, Some("foo.test.google.fr".into()));
}
let channel = endpoint.channel()?;
let channel = endpoint.channel();
let mut client = client::TestClient::new(channel.clone());
let mut unimplemented_client = client::UnimplementedClient::new(channel);
tonic-interop/src/client.rs
-24
View File
@@ -15,30 +15,6 @@ const TEST_STATUS_MESSAGE: &'static str = "test status message";
const SPECIAL_TEST_STATUS_MESSAGE: &'static str =
"\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";
pub async fn create(origin: http::Uri) -> Result<TestClient, Box<dyn std::error::Error>> {
// let ca = tokio::fs::read("tonic-interop/data/ca.pem").await?;
let svc = Channel::builder()
// .tls(ca)
// .tls_override_domain("foo.test.google.fr")
.build(origin)?;
Ok(TestServiceClient::new(svc))
}
pub async fn create_unimplemented(
origin: http::Uri,
) -> Result<UnimplementedClient, Box<dyn std::error::Error>> {
// let ca = tokio::fs::read("tonic-interop/data/ca.pem").await?;
let svc = Channel::builder()
// .tls(ca)
// .tls_override_domain("foo.test.google.fr")
.build(origin)?;
Ok(UnimplementedServiceClient::new(svc))
}
pub async fn empty_unary(client: &mut TestClient, assertions: &mut Vec<TestAssertion>) {
let result = client.empty_call(Request::new(Empty {})).await;
tonic/src/transport/channel.rs
+101 -79
View File
@@ -1,32 +1,32 @@
//! Client implementation and builder.
use super::{
service::{BoxService, Connection, ServiceList},
service::{Connection, ServiceList},
Endpoint,
};
use crate::{body::BoxBody, client::GrpcService};
use futures_util::try_future::{MapErr, TryFutureExt};
use hyper::{Request, Response};
use bytes::Bytes;
use http::{
uri::{InvalidUriBytes, Uri},
Request, Response,
};
use std::{
convert::TryInto,
fmt,
future::Future,
pin::Pin,
task::{Context, Poll},
};
use tower::buffer::{future::ResponseFuture, Buffer};
use tower::discover::Discover;
use tower::{
buffer::{self, Buffer},
discover::Discover,
util::{BoxService, Either},
Service,
};
use tower_balance::p2c::Balance;
use tower_service::Service;
type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
type Inner = Box<
dyn Service<
Request<BoxBody>,
Response = Response<hyper::Body>,
Error = crate::Error,
Future = BoxFuture<'static, Result<Response<hyper::Body>, crate::Error>>,
> + Send
+ 'static,
>;
type Svc = Either<Connection, BoxService<Request<BoxBody>, Response<hyper::Body>, crate::Error>>;
const DEFAULT_BUFFER_SIZE: usize = 1024;
/// A default batteries included `transport` channel.
///
@@ -34,67 +34,73 @@ type Inner = Box<
/// and `tower` services.
#[derive(Clone)]
pub struct Channel {
svc: Buffer<Inner, Request<BoxBody>>,
svc: Buffer<Svc, Request<BoxBody>>,
}
/// A future that resolves to an HTTP response.
///
/// This is returned by the `Service::call` on [`Channel`].
pub struct ResponseFuture {
inner: buffer::future::ResponseFuture<<Svc as Service<Request<BoxBody>>>::Future>,
}
impl Channel {
/// Create a [`Builder`] that can create a [`Channel`].
pub fn builder() -> Builder {
Builder::new()
}
}
impl GrpcService<BoxBody> for Channel {
type ResponseBody = hyper::Body;
type Error = super::Error;
type Future = MapErr<
ResponseFuture<BoxFuture<'static, Result<Response<Self::ResponseBody>, crate::Error>>>,
fn(crate::Error) -> super::Error,
>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
GrpcService::poll_ready(&mut self.svc, cx)
.map_err(|e| super::Error::from((super::ErrorKind::Client, e)))
/// Create a [`Endpoint`] builder that can create a [`Channel`]'s.
pub fn builder(uri: Uri) -> Endpoint {
Endpoint::from(uri)
}
fn call(&mut self, request: Request<BoxBody>) -> Self::Future {
GrpcService::call(&mut self.svc, request)
.map_err(|e| super::Error::from((super::ErrorKind::Client, e)))
}
}
#[derive(Debug)]
pub struct Builder<D = ServiceList> {
ca: Option<Vec<u8>>,
override_domain: Option<String>,
buffer_size: usize,
balance: Option<D>,
}
impl Builder {
fn new() -> Self {
Self {
ca: None,
override_domain: None,
buffer_size: 1024,
balance: None,
}
/// Create an `Endpoint` from a static string.
///
/// ```
/// # use tonic::transport::Channel;
/// Channel::from_static("https://example.com");
/// ```
pub fn from_static(s: &'static str) -> Endpoint {
let uri = Uri::from_static(s);
Self::builder(uri)
}
/// Set the buffer size for when the inner client applies back pressure and
/// can no longer accept requests. Defaults to `1024`.
pub fn buffer(&mut self, size: usize) -> &mut Self {
self.buffer_size = size;
self
/// Create an `Endpoint` from shared bytes.
///
/// ```
/// # use tonic::transport::Channel;
/// Channel::from_shared("https://example.com");
/// ```
pub fn from_shared(s: impl Into<Bytes>) -> Result<Endpoint, InvalidUriBytes> {
let uri = Uri::from_shared(s.into())?;
Ok(Self::builder(uri))
}
pub fn balance_list(&mut self, list: Vec<Endpoint>) -> Result<Channel, super::Error> {
/// Balance a list of [`Endpoint`]'s.
///
/// This creates a [`Channel`] that will load balance accross all the
/// provided endpoints.
pub fn balance_list(list: impl Iterator<Item = Endpoint>) -> Self {
let list = list.collect::<Vec<_>>();
let buffer_size = list
.iter()
.next()
.and_then(|e| e.buffer_size)
.unwrap_or(DEFAULT_BUFFER_SIZE);
let discover = ServiceList::new(list);
self.balance(discover)
Self::balance(discover, buffer_size)
}
fn balance<D>(&mut self, discover: D) -> Result<Channel, super::Error>
pub(crate) fn connect(endpoint: Endpoint) -> Self {
let buffer_size = endpoint.buffer_size.clone().unwrap_or(DEFAULT_BUFFER_SIZE);
let svc = Connection::new(endpoint);
let svc = Buffer::new(Either::A(svc), buffer_size);
Channel { svc }
}
pub(crate) fn balance<D>(discover: D, buffer_size: usize) -> Self
where
D: Discover<Service = Connection> + Unpin + Send + 'static,
D::Error: Into<crate::Error>,
@@ -103,25 +109,35 @@ impl Builder {
let svc = Balance::from_entropy(discover);
let svc = BoxService::new(svc);
let svc = Buffer::new(Box::new(svc) as Inner, 100);
let svc = Buffer::new(Either::B(svc), buffer_size);
Ok(Channel { svc })
Channel { svc }
}
}
impl GrpcService<BoxBody> for Channel {
type ResponseBody = hyper::Body;
type Error = super::Error;
type Future = ResponseFuture;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
GrpcService::poll_ready(&mut self.svc, cx)
.map_err(|e| super::Error::from_source(super::ErrorKind::Client, e))
}
pub fn connect(&mut self, endpoint: Endpoint) -> Result<Channel, super::Error> {
self.balance_list(vec![endpoint])
fn call(&mut self, request: Request<BoxBody>) -> Self::Future {
let inner = GrpcService::call(&mut self.svc, request);
ResponseFuture { inner }
}
}
pub fn build<T>(&mut self, uri: T) -> Result<Channel, super::Error>
where
T: TryInto<Endpoint>,
T::Error: Into<crate::Error>,
{
let uri = uri
.try_into()
.map_err(|e| super::Error::from((super::ErrorKind::Client, e.into())))?;
impl Future for ResponseFuture {
type Output = Result<Response<hyper::Body>, super::Error>;
self.balance_list(vec![uri.into()])
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let val = futures_util::ready!(Pin::new(&mut self.inner).poll(cx))
.map_err(|e| super::Error::from_source(super::ErrorKind::Client, e))?;
Ok(val).into()
}
}
@@ -130,3 +146,9 @@ impl fmt::Debug for Channel {
f.debug_struct("Channel").finish()
}
}
impl fmt::Debug for ResponseFuture {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ResponseFuture").finish()
}
}
tonic/src/transport/endpoint.rs
+93 -9
View File
@@ -3,8 +3,14 @@ use super::channel::Channel;
use super::{service::TlsConnector, tls::Certificate};
use bytes::Bytes;
use http::uri::{InvalidUriBytes, Uri};
use std::{convert::TryFrom, time::Duration};
use std::{
convert::{TryFrom, TryInto},
time::Duration,
};
/// Channel builder.
///
/// This struct is used to build and configure HTTP/2 channels.
#[derive(Debug, Clone)]
pub struct Endpoint {
pub(super) uri: Uri,
@@ -13,54 +19,131 @@ pub struct Endpoint {
pub(super) rate_limit: Option<(u64, Duration)>,
#[cfg(feature = "tls")]
pub(super) tls: Option<TlsConnector>,
pub(super) buffer_size: Option<usize>,
}
impl Endpoint {
// TODO: determine if we want to expose this or not. This is really
// just used in codegen for a shortcut.
#[doc(hidden)]
pub fn new<D>(dst: D) -> Result<Self, super::Error>
where
D: TryInto<Self>,
D::Error: Into<crate::Error>,
{
let me = dst
.try_into()
.map_err(|e| super::Error::from_source(super::ErrorKind::Client, e.into()))?;
Ok(me)
}
/// Convert an `Endpoint` from a static string.
///
/// ```
/// # use tonic::transport::Endpoint;
/// Endpoint::from_static("https://example.com");
/// ```
pub fn from_static(s: &'static str) -> Self {
let uri = Uri::from_static(s);
Self::from(uri)
}
/// Convert an `Endpoint` from shared bytes.
///
/// ```
/// # use tonic::transport::Endpoint;
/// Endpoint::from_shared("https://example.com".to_string());
/// ```
pub fn from_shared(s: impl Into<Bytes>) -> Result<Self, InvalidUriBytes> {
let uri = Uri::from_shared(s.into())?;
Ok(Self::from(uri))
}
/// Apply a timeout to each request.
///
/// ```
/// # use tonic::transport::Endpoint;
/// # use std::time::Duration;
/// # let mut builder = Endpoint::from_static("https://example.com");
/// builder.timeout(Duration::from_secs(5));
/// ```
pub fn timeout(&mut self, dur: Duration) -> &mut Self {
self.timeout = Some(dur);
self
}
/// Apply a concurrency limit to each request.
///
/// ```
/// # use tonic::transport::Endpoint;
/// # let mut builder = Endpoint::from_static("https://example.com");
/// builder.concurrency_limit(256);
/// ```
pub fn concurrency_limit(&mut self, limit: usize) -> &mut Self {
self.concurrency_limit = Some(limit);
self
}
/// Apply a rate limit to each request.
///
/// ```
/// # use tonic::transport::Endpoint;
/// # use std::time::Duration;
/// # let mut builder = Endpoint::from_static("https://example.com");
/// builder.rate_limit(32, Duration::from_secs(1));
/// ```
pub fn rate_limit(&mut self, limit: u64, duration: Duration) -> &mut Self {
self.rate_limit = Some((limit, duration));
self
}
/// ```no_run
/// # use tonic::transport::{Certificate, Endpoint};
/// # fn dothing() -> Result<(), Box<dyn std::error::Error>> {
/// # let mut builder = Endpoint::from_static("https://example.com");
/// let ca = std::fs::read_to_string("ca.pem")?;
///
/// let ca = Certificate::from_pem(ca);
///
/// builder.openssl_tls(ca, "example.com".to_string());
/// # Ok(())
/// # }
/// ```
#[cfg(feature = "openssl")]
pub fn openssl_tls(&mut self, ca: Certificate, domain: Option<String>) -> &mut Self {
let domain = domain.unwrap_or_else(|| self.uri.clone().to_string());
pub fn openssl_tls(&mut self, ca: Certificate, domain: impl Into<Option<String>>) -> &mut Self {
let domain = domain
.into()
.unwrap_or_else(|| self.uri.clone().to_string());
let tls = TlsConnector::new_with_openssl(ca, domain).unwrap();
self.tls = Some(tls);
self
}
/// ```no_run
/// # use tonic::transport::{Certificate, Endpoint};
/// # fn dothing() -> Result<(), Box<dyn std::error::Error>> {
/// # let mut builder = Endpoint::from_static("https://example.com");
/// let ca = std::fs::read_to_string("ca.pem")?;
///
/// let ca = Certificate::from_pem(ca);
///
/// builder.rustls_tls(ca, "example.com".to_string());
/// # Ok(())
/// # }
/// ```
#[cfg(feature = "rustls")]
pub fn rustls_tls(&mut self, ca: Certificate, domain: Option<String>) -> &mut Self {
let domain = domain.unwrap_or_else(|| self.uri.clone().to_string());
pub fn rustls_tls(&mut self, ca: Certificate, domain: impl Into<Option<String>>) -> &mut Self {
let domain = domain
.into()
.unwrap_or_else(|| self.uri.clone().to_string());
let tls = TlsConnector::new_with_rustls(ca, domain).unwrap();
self.tls = Some(tls);
self
}
// pub fn metadata_interceptor(f: impl Fn(MetadataMap) ->)
pub fn channel(&self) -> Result<Channel, super::Error> {
Channel::builder().connect(self.clone())
/// Create a channel from this config.
pub fn channel(&self) -> Channel {
Channel::connect(self.clone())
}
}
@@ -73,6 +156,7 @@ impl From<Uri> for Endpoint {
timeout: None,
#[cfg(feature = "tls")]
tls: None,
buffer_size: None,
}
}
}
tonic/src/transport/error.rs
+10 -18
View File
@@ -1,34 +1,26 @@
use std::{error, fmt};
/// Error's that originate from the client or server;
pub struct Error {
kind: ErrorKind,
source: Option<crate::Error>,
}
impl Error {
pub(crate) fn from_source(kind: ErrorKind, source: crate::Error) -> Self {
Self {
kind,
source: Some(source),
}
}
}
#[derive(Debug)]
pub(crate) enum ErrorKind {
Client,
Server,
}
impl From<ErrorKind> for Error {
fn from(t: ErrorKind) -> Self {
Self {
kind: t,
source: None,
}
}
}
impl From<(ErrorKind, crate::Error)> for Error {
fn from(t: (ErrorKind, crate::Error)) -> Self {
Self {
kind: t.0,
source: Some(t.1),
}
}
}
impl fmt::Debug for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let mut f = f.debug_tuple("Error");
tonic/src/transport/mod.rs
+75 -3
View File
@@ -1,6 +1,76 @@
#![allow(missing_docs)]
//! TODO: write transport docs.
//! Batteries included server and client.
//!
//! This module provides a set of batteries included, fully featured and
//! fast set of HTTP/2 server and client's. These components each provide either an
//! `openssl` or `rustls` tls backend when the respective feature flags are enabled.
//!They also provide may configurable knobs that can be used to tune how they work.
//!
//! # Features
//!
//! - TLS support via either [OpenSSL] or [rustls].
//! - Load balancing
//! - Timeouts
//! - Concurrency Limits
//! - Rate limiting
//! - gRPC Interceptors
//!
//! # Examples
//!
//! ## Client
//!
//! ```no_run
//! # use tonic::transport::{Channel, Certificate};
//! # use std::time::Duration;
//! # use tonic::body::BoxBody;
//! # use tonic::client::GrpcService;;
//! # use http::Request;
//! # #[cfg(feature = "rustls")]
//! # async fn do_thing() -> Result<(), Box<dyn std::error::Error>> {
//! let cert = std::fs::read_to_string("ca.pem")?;
//!
//! let mut channel = Channel::from_static("https://example.com")
//! .rustls_tls(Certificate::from_pem(&cert), "example.com".to_string())
//! .timeout(Duration::from_secs(5))
//! .rate_limit(5, Duration::from_secs(1))
//! .concurrency_limit(256)
//! .channel();
//!
//! channel.call(Request::new(BoxBody::empty())).await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Server
//!
//! ```no_run
//! # use tonic::transport::{Server, Identity};
//! # use tower::{Service, service_fn};
//! # 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("")))));
//! let cert = std::fs::read_to_string("server.pem")?;
//! let key = std::fs::read_to_string("server.key")?;
//!
//! let addr = "[::1]:50051".parse()?;
//!
//! Server::builder()
//! .rustls_tls(Identity::from_pem(&cert, &key))
//! .concurrency_limit_per_connection(256)
//! .interceptor_fn(|svc, req| {
//! println!("Request: {:?}", req);
//! svc.call(req)
//! })
//! .clone()
//! .serve(addr, my_svc)
//! .await?;
//!
//! # Ok(())
//! # }
//! ```
//!
//! [OpenSSL]: https://www.openssl.org/
//! [rustls]: https://docs.rs/rustls/0.16.0/rustls/
pub mod channel;
pub mod server;
@@ -10,9 +80,11 @@ mod error;
mod service;
mod tls;
#[doc(inline)]
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::tls::{Certificate, Identity};
pub use hyper::Body;
tonic/src/transport/server.rs
+105 -36
View File
@@ -1,12 +1,16 @@
use super::service::{layer_fn, BoxedIo};
//! Server implementation and builder.
use super::service::{layer_fn, BoxedIo, ServiceBuilderExt};
#[cfg(feature = "tls")]
use super::{service::TlsAcceptor, tls::Identity};
use crate::body::BoxBody;
use futures_core::Stream;
use futures_util::{ready, try_future::MapErr, TryFutureExt, TryStreamExt};
use http::{Request, Response};
use hyper::server::{accept::Accept, conn};
use hyper::Body;
use hyper::{
server::{accept::Accept, conn},
Body,
};
use std::{
fmt,
future::Future,
@@ -14,12 +18,16 @@ use std::{
pin::Pin,
sync::Arc,
task::{Context, Poll},
// time::Duration,
};
use tower::{
layer::{util::Stack, Layer},
limit::concurrency::ConcurrencyLimitLayer,
// timeout::TimeoutLayer,
Service,
ServiceBuilder,
};
use tower::layer::util::Stack;
use tower::layer::Layer;
use tower::util::Either;
use tower_make::MakeService;
use tower_service::Service;
type BoxService = tower::util::BoxService<Request<Body>, Response<BoxBody>, crate::Error>;
type Interceptor = Arc<dyn Layer<BoxService, Service = BoxService> + Send + Sync + 'static>;
@@ -27,38 +35,43 @@ type Interceptor = Arc<dyn Layer<BoxService, Service = BoxService> + Send + Sync
/// A default batteries included `transport` server.
///
/// This is a wrapper around [`hyper::Server`] and provides an easy builder
/// pattern style [`Builder`]. This builder exposes easy configuration parameters
/// pattern style builder [`Server`]. This builder exposes easy configuration parameters
/// for providing a fully featured http2 based gRPC server. This should provide
/// a very good out of the box http2 server for use with tonic but is also a
/// reference implementation that should be a good starting point for anyone
/// wanting to create a more complex and/or specific implementation.
#[derive(Debug)]
#[derive(Default, Clone)]
pub struct Server {
_p: (),
}
impl Server {
/// Create a new [`Builder`] that can configure a Server.
pub fn builder() -> Builder {
Builder::new()
}
}
///
#[derive(Default)]
pub struct Builder {
interceptor: Option<Interceptor>,
// concurrency_limit: Option<usize>,
concurrency_limit: Option<usize>,
// timeout: Option<Duration>,
#[cfg(feature = "tls")]
tls: Option<TlsAcceptor>,
}
impl Builder {
fn new() -> Self {
impl Server {
/// Create a new server builder that can configure a [`Server`].
pub fn builder() -> Self {
Default::default()
}
}
/// Add a tls cert.
impl Server {
/// Set the [`Identity`] of this server using `openssl`.
///
/// ```no_run
/// # use tonic::transport::{Identity, Server};
/// # fn dothing() -> Result<(), Box<dyn std::error::Error>> {
/// # let mut builder = Server::builder();
/// let cert = std::fs::read_to_string("server.pem")?;
/// let key = std::fs::read_to_string("server.key")?;
///
/// let identity = Identity::from_pem(&cert, &key);
///
/// builder.openssl_tls(identity);
/// # Ok(())
/// # }
/// ```
#[cfg(feature = "openssl")]
pub fn openssl_tls(&mut self, identity: Identity) -> &mut Self {
let acceptor = TlsAcceptor::new_with_openssl(identity).unwrap();
@@ -66,6 +79,21 @@ impl Builder {
self
}
/// Set the [`Identity`] of this server using `rustls`.
///
/// ```no_run
/// # use tonic::transport::{Identity, Server};
/// # fn dothing() -> Result<(), Box<dyn std::error::Error>> {
/// # let mut builder = Server::builder();
/// let cert = std::fs::read_to_string("server.pem")?;
/// let key = std::fs::read_to_string("server.key")?;
///
/// let identity = Identity::from_pem(&cert, &key);
///
/// builder.rustls_tls(identity);
/// # Ok(())
/// # }
/// ```
#[cfg(feature = "rustls")]
pub fn rustls_tls(&mut self, identity: Identity) -> &mut Self {
let acceptor = TlsAcceptor::new_with_rustls(identity).unwrap();
@@ -73,13 +101,37 @@ impl Builder {
self
}
// FIXME: add server side layering ability
// pub fn concurrency_limit(&mut self, limit: usize) -> &mut Self {
// self.concurrency_limit = Some(limit);
/// Set the concurrency limit applied to on requests inbound per connection.
///
/// ```
/// # use tonic::transport::Server;
/// # use tower_service::Service;
/// # let mut builder = Server::builder();
/// builder.concurrency_limit_per_connection(32);
/// ```
pub fn concurrency_limit_per_connection(&mut self, limit: usize) -> &mut Self {
self.concurrency_limit = Some(limit);
self
}
// FIXME: tower-timeout currentlly uses `From` instead of `Into` for the error
// so our services do not align.
// pub fn timeout(&mut self, timeout: Duration) -> &mut Self {
// self.timeout = Some(timeout);
// self
// }
/// Intercept the execution of gRPC methods.
///
/// ```
/// # use tonic::transport::Server;
/// # use tower_service::Service;
/// # let mut builder = Server::builder();
/// builder.interceptor_fn(|svc, req| {
/// println!("request={:?}", req);
/// svc.call(req)
/// });
/// ```
pub fn interceptor_fn<F, Out>(&mut self, f: F) -> &mut Self
where
F: Fn(&mut BoxService, Request<Body>) -> Out + Send + Sync + 'static,
@@ -95,6 +147,8 @@ impl Builder {
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>
where
M: Service<(), Response = S>,
@@ -105,6 +159,8 @@ impl Builder {
S::Error: Into<crate::Error> + Send,
{
let interceptor = self.interceptor.clone();
let concurrency_limit = self.concurrency_limit.clone();
// let timeout = self.timeout.clone();
let incoming = hyper::server::accept::from_stream(async_stream::try_stream! {
let mut tcp = TcpIncoming::bind(addr)?;
@@ -126,6 +182,8 @@ impl Builder {
let svc = MakeSvc {
inner: svc,
interceptor,
concurrency_limit,
// timeout,
};
hyper::Server::builder(incoming)
@@ -139,10 +197,10 @@ impl Builder {
}
fn map_err(e: impl Into<crate::Error>) -> super::Error {
(super::ErrorKind::Server, e.into()).into()
super::Error::from_source(super::ErrorKind::Server, e.into())
}
impl fmt::Debug for Builder {
impl fmt::Debug for Server {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Builder").finish()
}
@@ -197,6 +255,8 @@ where
struct MakeSvc<M> {
interceptor: Option<Interceptor>,
concurrency_limit: Option<usize>,
// timeout: Option<Duration>,
inner: M,
}
@@ -209,7 +269,7 @@ where
S::Future: Send + 'static,
S::Error: Into<crate::Error> + Send,
{
type Response = Either<Svc<S>, BoxService>;
type Response = BoxService;
type Error = crate::Error;
type Future =
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
@@ -221,16 +281,25 @@ where
fn call(&mut self, _: T) -> Self::Future {
let interceptor = self.interceptor.clone();
let make = self.inner.make_service(());
let concurrency_limit = self.concurrency_limit.clone();
// let timeout = self.timeout.clone();
Box::pin(async move {
let svc = make.await.map_err(Into::into)?;
if let Some(interceptor) = interceptor {
let svc = ServiceBuilder::new()
.optional_layer(concurrency_limit.map(ConcurrencyLimitLayer::new))
// .optional_layer(timeout.map(TimeoutLayer::new))
.service(svc);
let svc = if let Some(interceptor) = interceptor {
let layered = interceptor.layer(BoxService::new(Svc(svc)));
Ok(Either::B(layered))
BoxService::new(Svc(layered))
} else {
Ok(Either::A(Svc(svc)))
}
BoxService::new(Svc(svc))
};
Ok(svc)
})
}
}
tonic/src/transport/service/boxed.rs
-38
View File
@@ -1,38 +0,0 @@
use std::{
future::Future,
pin::Pin,
task::{Context, Poll},
};
use tower_service::Service;
#[derive(Debug, Clone)]
pub(crate) struct BoxService<S> {
inner: S,
}
impl<S> BoxService<S> {
pub(crate) fn new(inner: S) -> Self {
Self { inner }
}
}
impl<S, Request> Service<Request> for BoxService<S>
where
S: Service<Request>,
S::Future: Send + 'static,
{
type Response = S::Response;
type Error = S::Error;
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>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, request: Request) -> Self::Future {
let fut = self.inner.call(request);
Box::pin(fut)
}
}
tonic/src/transport/service/connection.rs
+4 -4
View File
@@ -22,12 +22,12 @@ use tower_service::Service;
pub(crate) type Request = http::Request<BoxBody>;
pub(crate) type Response = http::Response<hyper::Body>;
pub struct Connection {
pub(crate) struct Connection {
inner: BoxService<Request, Response, crate::Error>,
}
impl Connection {
pub fn new(endpoint: Endpoint) -> Result<Self, crate::Error> {
pub(crate) fn new(endpoint: Endpoint) -> Self {
#[cfg(feature = "tls")]
let connector = connector(endpoint.tls.clone());
@@ -51,9 +51,9 @@ impl Connection {
let inner = stack.layer(conn);
Ok(Self {
Self {
inner: BoxService::new(inner),
})
}
}
}
tonic/src/transport/service/discover.rs
+6 -6
View File
@@ -6,13 +6,13 @@ use std::task::{Context, Poll};
use tower::discover::{Change, Discover};
#[derive(Debug)]
pub struct ServiceList {
pub(crate) struct ServiceList {
list: VecDeque<Endpoint>,
i: usize,
}
impl ServiceList {
pub fn new(list: Vec<Endpoint>) -> Self {
pub(crate) fn new(list: Vec<Endpoint>) -> Self {
Self {
list: list.into(),
i: 0,
@@ -34,10 +34,10 @@ impl Discover for ServiceList {
let i = self.i;
self.i += 1;
match Connection::new(endpoint) {
Ok(svc) => Poll::Ready(Ok(Change::Insert(i, svc))),
Err(e) => Poll::Ready(Err(e)),
}
let svc = Connection::new(endpoint);
let change = Ok(Change::Insert(i, svc));
Poll::Ready(change)
}
None => Poll::Pending,
}
tonic/src/transport/service/either.rs
+72
View File
@@ -0,0 +1,72 @@
use futures_util::try_future::{MapErr, TryFutureExt};
use std::{
future::Future,
pin::Pin,
task::{Context, Poll},
};
use tower::Service;
pub(crate) enum Either<A, B> {
A(A),
B(B),
}
impl<A, B, Request, Response> Service<Request> for Either<A, B>
where
A: Service<Request, Response = Response>,
B: Service<Request, Response = Response>,
A::Error: Into<crate::Error>,
B::Error: Into<crate::Error>,
{
type Response = 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>> {
match self {
Either::A(svc) => svc.poll_ready(cx).map_err(Into::into),
Either::B(svc) => svc.poll_ready(cx).map_err(Into::into),
}
}
fn call(&mut self, req: Request) -> Self::Future {
match self {
Either::A(svc) => {
let fut = svc
.call(req)
.map_err((|e| e.into()) as fn(A::Error) -> crate::Error);
Either::A(fut)
}
Either::B(svc) => {
let fut = svc
.call(req)
.map_err((|e| e.into()) as fn(B::Error) -> crate::Error);
Either::B(fut)
}
}
}
}
impl<A: Unpin, B: Unpin> Unpin for Either<A, B> {}
impl<A, B> Future for Either<A, B>
where
A: Future,
B: Future<Output = A::Output>,
{
type Output = A::Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// safe because we do not exposed an unchecked mut beyond this projection.
let mut me = unsafe { self.get_unchecked_mut() };
match &mut me {
Either::A(fut) => unsafe { Pin::new_unchecked(fut) }.poll(cx),
Either::B(fut) => unsafe { Pin::new_unchecked(fut) }.poll(cx),
}
}
}
tonic/src/transport/service/layer.rs
+1 -1
View File
@@ -1,6 +1,6 @@
use super::either::Either;
use tower::{
layer::{util::Stack, Layer},
util::Either,
ServiceBuilder,
};
pub(crate) trait ServiceBuilderExt<L> {
tonic/src/transport/service/mod.rs
+2 -3
View File
@@ -1,19 +1,18 @@
mod add_origin;
mod boxed;
mod connection;
mod connector;
mod discover;
mod either;
mod io;
mod layer;
#[cfg(feature = "tls")]
mod tls;
pub(crate) use self::add_origin::AddOrigin;
pub(crate) use self::boxed::BoxService;
pub(crate) use self::connection::Connection;
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;
pub(crate) use self::layer::{layer_fn, ServiceBuilderExt};
#[cfg(feature = "tls")]
pub(crate) use self::tls::{TlsAcceptor, TlsConnector};
tonic/src/transport/tls.rs
+12 -2
View File
@@ -1,8 +1,10 @@
/// Represents a X509 certificate.
#[derive(Debug, Clone)]
pub struct Certificate {
pub(crate) pem: Vec<u8>,
}
/// Represents a private key and X509 certificate.
#[derive(Debug, Clone)]
pub struct Identity {
pub(crate) cert: Certificate,
@@ -10,14 +12,22 @@ pub struct Identity {
}
impl Certificate {
pub fn from_pem(pem: Vec<u8>) -> Self {
/// Parse a PEM encoded X509 Certificate.
///
/// The provided PEM should include at least one PEM encoded certificate.
pub fn from_pem(pem: impl AsRef<[u8]>) -> Self {
let pem = pem.as_ref().into();
Self { pem }
}
}
impl Identity {
pub fn from_pem(cert: Vec<u8>, key: Vec<u8>) -> Self {
/// Parse a PEM encoded certificate and private key.
///
/// The provided cert must contain at least one PEM encoded certificate.
pub fn from_pem(cert: impl AsRef<[u8]>, key: impl AsRef<[u8]>) -> Self {
let cert = Certificate::from_pem(cert);
let key = key.as_ref().into();
Self { cert, key }
}
}