From 1626c2eae0007a6743cfe8a84f60c24db9b6f055 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Wed, 11 Dec 2019 18:19:11 -0500 Subject: [PATCH] chore(transport): Clean up server and channel (#174) * chore(transport): Clean up server and channel * Fix tls feature compilation --- tonic/src/transport/{ => channel}/endpoint.rs | 125 ++-------- .../transport/{channel.rs => channel/mod.rs} | 13 +- tonic/src/transport/channel/tls.rs | 93 +++++++ tonic/src/transport/mod.rs | 6 +- .../transport/{server.rs => server/mod.rs} | 226 ++++-------------- tonic/src/transport/server/tls.rs | 70 ++++++ 6 files changed, 244 insertions(+), 289 deletions(-) rename tonic/src/transport/{ => channel}/endpoint.rs (63%) rename tonic/src/transport/{channel.rs => channel/mod.rs} (97%) create mode 100644 tonic/src/transport/channel/tls.rs rename tonic/src/transport/{server.rs => server/mod.rs} (71%) create mode 100644 tonic/src/transport/server/tls.rs diff --git a/tonic/src/transport/endpoint.rs b/tonic/src/transport/channel/endpoint.rs similarity index 63% rename from tonic/src/transport/endpoint.rs rename to tonic/src/transport/channel/endpoint.rs index e3ca9eb..cf518d2 100644 --- a/tonic/src/transport/endpoint.rs +++ b/tonic/src/transport/channel/endpoint.rs @@ -1,9 +1,9 @@ -use super::channel::Channel; +use super::Channel; #[cfg(feature = "tls")] -use super::{ - service::TlsConnector, - tls::{Certificate, Identity}, -}; +use super::ClientTlsConfig; +#[cfg(feature = "tls")] +use crate::transport::service::TlsConnector; +use crate::transport::{Error, ErrorKind}; use bytes::Bytes; use http::uri::{InvalidUri, Uri}; use std::{ @@ -18,33 +18,33 @@ use std::{ /// This struct is used to build and configure HTTP/2 channels. #[derive(Clone)] pub struct Endpoint { - pub(super) uri: Uri, - pub(super) timeout: Option, - pub(super) concurrency_limit: Option, - pub(super) rate_limit: Option<(u64, Duration)>, + pub(crate) uri: Uri, + pub(crate) timeout: Option, + pub(crate) concurrency_limit: Option, + pub(crate) rate_limit: Option<(u64, Duration)>, #[cfg(feature = "tls")] - pub(super) tls: Option, - pub(super) buffer_size: Option, - pub(super) interceptor_headers: + pub(crate) tls: Option, + pub(crate) buffer_size: Option, + pub(crate) interceptor_headers: Option>, - pub(super) init_stream_window_size: Option, - pub(super) init_connection_window_size: Option, - pub(super) tcp_keepalive: Option, - pub(super) tcp_nodelay: bool, + pub(crate) init_stream_window_size: Option, + pub(crate) init_connection_window_size: Option, + pub(crate) tcp_keepalive: Option, + pub(crate) tcp_nodelay: bool, } impl Endpoint { // FIXME: determine if we want to expose this or not. This is really // just used in codegen for a shortcut. #[doc(hidden)] - pub fn new(dst: D) -> Result + pub fn new(dst: D) -> Result where D: TryInto, D::Error: Into, { let me = dst .try_into() - .map_err(|e| super::Error::from_source(super::ErrorKind::Client, e.into()))?; + .map_err(|e| Error::from_source(ErrorKind::Client, e.into()))?; Ok(me) } @@ -181,7 +181,7 @@ impl Endpoint { } /// Create a channel from this config. - pub async fn connect(&self) -> Result { + pub async fn connect(&self) -> Result { Channel::connect(self.clone()).await } } @@ -245,90 +245,3 @@ impl fmt::Debug for Endpoint { f.debug_struct("Endpoint").finish() } } - -/// Configures TLS settings for endpoints. -#[cfg(feature = "tls")] -#[derive(Clone)] -pub struct ClientTlsConfig { - domain: Option, - cert: Option, - identity: Option, - rustls_raw: Option, -} - -#[cfg(feature = "tls")] -impl fmt::Debug for ClientTlsConfig { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ClientTlsConfig") - .field("domain", &self.domain) - .field("cert", &self.cert) - .field("identity", &self.identity) - .finish() - } -} - -#[cfg(feature = "tls")] -impl ClientTlsConfig { - /// Creates a new `ClientTlsConfig` using Rustls. - pub fn with_rustls() -> Self { - ClientTlsConfig { - domain: None, - cert: None, - identity: None, - rustls_raw: None, - } - } - - /// Sets the domain name against which to verify the server's TLS certificate. - /// - /// This has no effect if `rustls_client_config` is used to configure Rustls. - pub fn domain_name(self, domain_name: impl Into) -> Self { - ClientTlsConfig { - domain: Some(domain_name.into()), - ..self - } - } - - /// Sets the CA Certificate against which to verify the server's TLS certificate. - /// - /// This has no effect if `rustls_client_config` is used to configure Rustls. - pub fn ca_certificate(self, ca_certificate: Certificate) -> Self { - ClientTlsConfig { - cert: Some(ca_certificate), - ..self - } - } - - /// Sets the client identity to present to the server. - /// - /// This has no effect if `rustls_client_config` is used to configure Rustls. - pub fn identity(self, identity: Identity) -> Self { - ClientTlsConfig { - identity: Some(identity), - ..self - } - } - - /// Use options specified by the given `ClientConfig` to configure TLS. - /// - /// This overrides all other TLS options set via other means. - pub fn rustls_client_config(self, config: tokio_rustls::rustls::ClientConfig) -> Self { - ClientTlsConfig { - rustls_raw: Some(config), - ..self - } - } - - fn tls_connector(&self, uri: Uri) -> Result { - let domain = match &self.domain { - None => uri.to_string(), - Some(domain) => domain.clone(), - }; - match &self.rustls_raw { - None => { - TlsConnector::new_with_rustls_cert(self.cert.clone(), self.identity.clone(), domain) - } - Some(c) => TlsConnector::new_with_rustls_raw(c.clone(), domain), - } - } -} diff --git a/tonic/src/transport/channel.rs b/tonic/src/transport/channel/mod.rs similarity index 97% rename from tonic/src/transport/channel.rs rename to tonic/src/transport/channel/mod.rs index 0017add..586e177 100644 --- a/tonic/src/transport/channel.rs +++ b/tonic/src/transport/channel/mod.rs @@ -1,9 +1,14 @@ //! Client implementation and builder. -use super::{ - service::{Connection, ServiceList}, - Endpoint, -}; +mod endpoint; +#[cfg(feature = "tls")] +mod tls; + +pub use endpoint::Endpoint; +#[cfg(feature = "tls")] +pub use tls::ClientTlsConfig; + +use super::service::{Connection, ServiceList}; use crate::{body::BoxBody, client::GrpcService}; use bytes::Bytes; use http::{ diff --git a/tonic/src/transport/channel/tls.rs b/tonic/src/transport/channel/tls.rs new file mode 100644 index 0000000..f8e3f04 --- /dev/null +++ b/tonic/src/transport/channel/tls.rs @@ -0,0 +1,93 @@ +use crate::transport::{ + service::TlsConnector, + tls::{Certificate, Identity}, +}; +use http::Uri; +use std::fmt; + +/// Configures TLS settings for endpoints. +#[cfg(feature = "tls")] +#[derive(Clone)] +pub struct ClientTlsConfig { + domain: Option, + cert: Option, + identity: Option, + rustls_raw: Option, +} + +#[cfg(feature = "tls")] +impl fmt::Debug for ClientTlsConfig { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ClientTlsConfig") + .field("domain", &self.domain) + .field("cert", &self.cert) + .field("identity", &self.identity) + .finish() + } +} + +#[cfg(feature = "tls")] +impl ClientTlsConfig { + /// Creates a new `ClientTlsConfig` using Rustls. + pub fn with_rustls() -> Self { + ClientTlsConfig { + domain: None, + cert: None, + identity: None, + rustls_raw: None, + } + } + + /// Sets the domain name against which to verify the server's TLS certificate. + /// + /// This has no effect if `rustls_client_config` is used to configure Rustls. + pub fn domain_name(self, domain_name: impl Into) -> Self { + ClientTlsConfig { + domain: Some(domain_name.into()), + ..self + } + } + + /// Sets the CA Certificate against which to verify the server's TLS certificate. + /// + /// This has no effect if `rustls_client_config` is used to configure Rustls. + pub fn ca_certificate(self, ca_certificate: Certificate) -> Self { + ClientTlsConfig { + cert: Some(ca_certificate), + ..self + } + } + + /// Sets the client identity to present to the server. + /// + /// This has no effect if `rustls_client_config` is used to configure Rustls. + pub fn identity(self, identity: Identity) -> Self { + ClientTlsConfig { + identity: Some(identity), + ..self + } + } + + /// Use options specified by the given `ClientConfig` to configure TLS. + /// + /// This overrides all other TLS options set via other means. + pub fn rustls_client_config(self, config: tokio_rustls::rustls::ClientConfig) -> Self { + ClientTlsConfig { + rustls_raw: Some(config), + ..self + } + } + + pub(crate) fn tls_connector(&self, uri: Uri) -> Result { + let domain = match &self.domain { + None => uri.to_string(), + Some(domain) => domain.clone(), + }; + match &self.rustls_raw { + None => { + TlsConnector::new_with_rustls_cert(self.cert.clone(), self.identity.clone(), domain) + } + Some(c) => TlsConnector::new_with_rustls_raw(c.clone(), domain), + } + } +} diff --git a/tonic/src/transport/mod.rs b/tonic/src/transport/mod.rs index 64fa729..e4cabda 100644 --- a/tonic/src/transport/mod.rs +++ b/tonic/src/transport/mod.rs @@ -94,14 +94,12 @@ pub mod channel; pub mod server; -mod endpoint; mod error; mod service; mod tls; #[doc(inline)] -pub use self::channel::Channel; -pub use self::endpoint::Endpoint; +pub use self::channel::{Channel, Endpoint}; pub use self::error::Error; #[doc(inline)] pub use self::server::{Server, ServiceName}; @@ -109,7 +107,7 @@ pub use self::tls::{Certificate, Identity}; pub use hyper::Body; #[cfg(feature = "tls")] -pub use self::endpoint::ClientTlsConfig; +pub use self::channel::ClientTlsConfig; #[cfg(feature = "tls")] pub use self::server::ServerTlsConfig; diff --git a/tonic/src/transport/server.rs b/tonic/src/transport/server/mod.rs similarity index 71% rename from tonic/src/transport/server.rs rename to tonic/src/transport/server/mod.rs index a166574..1f4ec87 100644 --- a/tonic/src/transport/server.rs +++ b/tonic/src/transport/server/mod.rs @@ -1,13 +1,19 @@ //! Server implementation and builder. -use super::service::{layer_fn, BoxedIo, Or, Routes, ServiceBuilderExt}; #[cfg(feature = "tls")] -use super::{service::TlsAcceptor, tls::Identity, Certificate}; +mod tls; + +#[cfg(feature = "tls")] +pub use tls::ServerTlsConfig; + +#[cfg(feature = "tls")] +use super::service::TlsAcceptor; + +use super::service::{layer_fn, BoxedIo, Or, Routes, ServiceBuilderExt}; use crate::body::BoxBody; -use futures_core::Stream; use futures_util::{ - future::{self, MapErr}, - ready, TryFutureExt, TryStreamExt, + future::{self, poll_fn, MapErr}, + TryFutureExt, }; use http::{Request, Response}; use hyper::{ @@ -222,69 +228,11 @@ impl Server { Router::new(self.clone(), svc) } - pub(crate) async fn serve(self, addr: SocketAddr, svc: S) -> Result<(), super::Error> - where - S: Service, Response = Response> + Clone + Send + 'static, - S::Future: Send + 'static, - S::Error: Into + Send, - { - let interceptor = self.interceptor.clone(); - let concurrency_limit = self.concurrency_limit; - let init_connection_window_size = self.init_connection_window_size; - let init_stream_window_size = self.init_stream_window_size; - let max_concurrent_streams = self.max_concurrent_streams; - // let timeout = self.timeout.clone(); - - let incoming = hyper::server::accept::from_stream::<_, _, crate::Error>( - async_stream::try_stream! { - let mut tcp = TcpIncoming::bind(addr)? - .set_nodelay(self.tcp_nodelay) - .set_keepalive(self.tcp_keepalive); - - while let Some(stream) = tcp.try_next().await? { - #[cfg(feature = "tls")] - { - if let Some(tls) = &self.tls { - let io = match tls.connect(stream.into_inner()).await { - Ok(io) => io, - Err(error) => { - error!(message = "Unable to accept incoming connection.", %error); - continue - }, - }; - yield BoxedIo::new(io); - continue; - } - } - - yield BoxedIo::new(stream); - } - }, - ); - let svc = MakeSvc { - inner: svc, - interceptor, - concurrency_limit, - // timeout, - }; - - hyper::Server::builder(incoming) - .http2_only(true) - .http2_initial_connection_window_size(init_connection_window_size) - .http2_initial_stream_window_size(init_stream_window_size) - .http2_max_concurrent_streams(max_concurrent_streams) - .serve(svc) - .await - .map_err(map_err)?; - - Ok(()) - } - pub(crate) async fn serve_with_shutdown( self, addr: SocketAddr, svc: S, - signal: F, + signal: Option, ) -> Result<(), super::Error> where S: Service, Response = Response> + Clone + Send + 'static, @@ -301,11 +249,14 @@ impl Server { let incoming = hyper::server::accept::from_stream::<_, _, crate::Error>( async_stream::try_stream! { - let mut tcp = TcpIncoming::bind(addr)? - .set_nodelay(self.tcp_nodelay) - .set_keepalive(self.tcp_keepalive); + let mut incoming = conn::AddrIncoming::bind(&addr)?; - while let Some(stream) = tcp.try_next().await? { + incoming.set_nodelay(self.tcp_nodelay); + incoming.set_keepalive(self.tcp_keepalive); + + + + while let Some(stream) = next_accept(&mut incoming).await? { #[cfg(feature = "tls")] { if let Some(tls) = &self.tls { @@ -332,15 +283,22 @@ impl Server { concurrency_limit, // timeout, }; - hyper::Server::builder(incoming) + + let server = hyper::Server::builder(incoming) .http2_only(true) .http2_initial_connection_window_size(init_connection_window_size) .http2_initial_stream_window_size(init_stream_window_size) - .http2_max_concurrent_streams(max_concurrent_streams) - .serve(svc) - .with_graceful_shutdown(signal) - .await - .map_err(map_err)?; + .http2_max_concurrent_streams(max_concurrent_streams); + + if let Some(signal) = signal { + server + .serve(svc) + .with_graceful_shutdown(signal) + .await + .map_err(map_err)? + } else { + server.serve(svc).await.map_err(map_err)?; + } Ok(()) } @@ -410,7 +368,9 @@ where /// /// [`Server`]: struct.Server.html pub async fn serve(self, addr: SocketAddr) -> Result<(), super::Error> { - self.server.serve(addr, self.routes).await + self.server + .serve_with_shutdown::<_, future::Ready<()>>(addr, self.routes, None) + .await } /// Consume this [`Server`] creating a future that will execute the server @@ -423,7 +383,9 @@ where addr: SocketAddr, f: F, ) -> Result<(), super::Error> { - self.server.serve_with_shutdown(addr, self.routes, f).await + self.server + .serve_with_shutdown(addr, self.routes, Some(f)) + .await } } @@ -437,105 +399,6 @@ impl fmt::Debug for Server { } } -/// Configures TLS settings for servers. -#[cfg(feature = "tls")] -#[derive(Clone)] -pub struct ServerTlsConfig { - identity: Option, - client_ca_root: Option, - rustls_raw: Option, -} - -#[cfg(feature = "tls")] -impl fmt::Debug for ServerTlsConfig { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ServerTlsConfig").finish() - } -} - -#[cfg(feature = "tls")] -impl ServerTlsConfig { - /// Creates a new `ServerTlsConfig`. - pub fn with_rustls() -> Self { - ServerTlsConfig { - identity: None, - client_ca_root: None, - rustls_raw: None, - } - } - - /// Sets the [`Identity`] of the server. - pub fn identity(self, identity: Identity) -> Self { - ServerTlsConfig { - identity: Some(identity), - ..self - } - } - - /// Sets a certificate against which to validate client TLS certificates. - pub fn client_ca_root(self, cert: Certificate) -> Self { - ServerTlsConfig { - client_ca_root: Some(cert), - ..self - } - } - - /// Use options specified by the given `ServerConfig` to configure TLS. - /// - /// This overrides all other TLS options set via other means. - pub fn rustls_server_config( - &mut self, - config: tokio_rustls::rustls::ServerConfig, - ) -> &mut Self { - self.rustls_raw = Some(config); - self - } - - fn tls_acceptor(&self) -> Result { - match &self.rustls_raw { - None => TlsAcceptor::new_with_rustls_identity( - self.identity.clone().unwrap(), - self.client_ca_root.clone(), - ), - Some(config) => TlsAcceptor::new_with_rustls_raw(config.clone()), - } - } -} - -#[derive(Debug)] -struct TcpIncoming { - inner: conn::AddrIncoming, -} - -impl TcpIncoming { - fn bind(addr: SocketAddr) -> Result { - let inner = conn::AddrIncoming::bind(&addr).map_err(Box::new)?; - Ok(Self { inner }) - } - - fn set_nodelay(mut self, enabled: bool) -> Self { - self.inner.set_nodelay(enabled); - self - } - - fn set_keepalive(mut self, tcp_keepalive: Option) -> Self { - self.inner.set_keepalive(tcp_keepalive); - self - } -} - -impl Stream for TcpIncoming { - type Item = Result; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - match ready!(Accept::poll_accept(Pin::new(&mut self.inner), cx)) { - Some(Ok(s)) => Poll::Ready(Some(Ok(s))), - Some(Err(e)) => Poll::Ready(Some(Err(e.into()))), - None => Poll::Ready(None), - } - } -} - #[derive(Debug)] struct Svc(S); @@ -628,3 +491,16 @@ impl Service> for Unimplemented { ) } } + +// Implement try_next for `Accept::poll_accept`. +async fn next_accept( + incoming: &mut conn::AddrIncoming, +) -> Result, crate::Error> { + let res = poll_fn(|cx| Pin::new(&mut *incoming).poll_accept(cx)).await; + + if let Some(res) = res { + Ok(Some(res?)) + } else { + return Ok(None); + } +} diff --git a/tonic/src/transport/server/tls.rs b/tonic/src/transport/server/tls.rs new file mode 100644 index 0000000..c2977ff --- /dev/null +++ b/tonic/src/transport/server/tls.rs @@ -0,0 +1,70 @@ +use crate::transport::{ + service::TlsAcceptor, + tls::{Certificate, Identity}, +}; +use std::fmt; + +/// Configures TLS settings for servers. +#[cfg(feature = "tls")] +#[derive(Clone)] +pub struct ServerTlsConfig { + identity: Option, + client_ca_root: Option, + rustls_raw: Option, +} + +#[cfg(feature = "tls")] +impl fmt::Debug for ServerTlsConfig { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ServerTlsConfig").finish() + } +} + +#[cfg(feature = "tls")] +impl ServerTlsConfig { + /// Creates a new `ServerTlsConfig`. + pub fn with_rustls() -> Self { + ServerTlsConfig { + identity: None, + client_ca_root: None, + rustls_raw: None, + } + } + + /// Sets the [`Identity`] of the server. + pub fn identity(self, identity: Identity) -> Self { + ServerTlsConfig { + identity: Some(identity), + ..self + } + } + + /// Sets a certificate against which to validate client TLS certificates. + pub fn client_ca_root(self, cert: Certificate) -> Self { + ServerTlsConfig { + client_ca_root: Some(cert), + ..self + } + } + + /// Use options specified by the given `ServerConfig` to configure TLS. + /// + /// This overrides all other TLS options set via other means. + pub fn rustls_server_config( + &mut self, + config: tokio_rustls::rustls::ServerConfig, + ) -> &mut Self { + self.rustls_raw = Some(config); + self + } + + pub(crate) fn tls_acceptor(&self) -> Result { + match &self.rustls_raw { + None => TlsAcceptor::new_with_rustls_identity( + self.identity.clone().unwrap(), + self.client_ca_root.clone(), + ), + Some(config) => TlsAcceptor::new_with_rustls_raw(config.clone()), + } + } +}