Switch to using endpoint

tonic-examples/src/helloworld/client.rs

@@ -8,9 +8,8 @@ pub mod hello_world {
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let origin = vec![
-        http::Uri::from_static("http://[::1]:50051"),
+        http::Uri::from_static("http://[::1]:50051").into(),
-        http::Uri::from_static("http://[::1]:50051"),
+
-        http::Uri::from_static("http://[::1]:50051"),
     ];
 
     let svc = Channel::builder().balance_list(origin)?;

tonic/Cargo.toml

@@ -43,7 +43,7 @@ transport = [
     "hyper",
     "tower",
     "tokio",
-    "openssl-1",
+    "native-tls",
 ]
 tower = [
     "tower-reconnect",
@@ -52,5 +52,5 @@ tower = [
     "tower-load",
     "tower-discover"
 ]
-openssl-1 = ["openssl", "tokio-openssl"]
+native-tls = ["openssl", "tokio-openssl"]
 rustls = ["tokio-rustls"]

tonic/src/transport/channel.rs

@@ -1,15 +1,17 @@
-use super::service::{AddOrigin, BoxService, ServiceList};
+use super::{
+    service::{BoxService, Connection, ServiceList},
+    Endpoint,
+};
 use crate::{BoxBody, GrpcService};
 use futures_util::try_future::{MapErr, TryFutureExt};
 use http::Uri;
-use hyper::client::conn;
-use hyper::client::connect::HttpConnector;
-use hyper::client::service::Connect;
 use hyper::{Request, Response};
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use tower_balance::p2c::Balance;
 use tower_buffer::{future::ResponseFuture, Buffer};
+use tower_discover::Discover;
 use tower_service::Service;
 
 type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
@@ -89,20 +91,26 @@ impl Builder {
         self
     }
 
-    pub fn balance_list(&mut self, list: Vec<Uri>) -> Result<Channel, super::Error> {
+    pub fn balance_list(&mut self, list: Vec<Endpoint>) -> Result<Channel, super::Error> {
         let discover = ServiceList::new(list);
-        let svc = tower_balance::p2c::Balance::from_entropy(discover);
+        self.balance(discover)
+    }
+
+    fn balance<D>(&mut self, discover: D) -> Result<Channel, super::Error>
+    where
+        D: Discover<Service = Connection> + Send + 'static,
+        D::Error: Into<crate::Error>,
+        D::Key: Send + Clone,
+    {
+        let svc = Balance::from_entropy(discover);
+
         let svc = BoxService::new(svc);
         let svc = Buffer::new(Box::new(svc) as Inner, 100);
+
         Ok(Channel { svc })
     }
 
-    // pub fn balance<D: Discover>(&mut self, discover: D) -> &mut Self<D> {
+    pub fn build<T>(&mut self, uri: T) -> Result<Channel, super::Error>
-    //     self.balance = Some(discover);
-    //     self
-    // }
-
-    pub fn build<T>(&self, uri: T) -> Result<Channel, super::Error>
     where
         Uri: http::HttpTryFrom<T>,
     {
@@ -111,41 +119,6 @@ impl Builder {
             Err(e) => panic!("Invalid uri: {}", e.into()),
         };
 
-        let settings = conn::Builder::new().http2_only(true).clone();
+        self.balance_list(vec![uri.into()])
-
-        let svc = if let Some(ca) = &self.ca {
-            let domain = self
-                .override_domain
-                .clone()
-                .unwrap_or_else(|| uri.to_string());
-
-            #[cfg(not(any(feature = "openssl-1", feature = "rustls")))]
-            unreachable!("tls configured when no tls implementation feature was selected!");
-
-            #[cfg(feature = "openssl-1")]
-            let connector = super::openssl::TlsConnector::new(ca.clone(), domain)?;
-
-            #[cfg(feature = "rustls")]
-            #[cfg(not(feature = "openssl-1"))]
-            let connector = super::openssl::TlsConnector::new(ca.clone(), domain)?;
-
-            let maker = Connect::new(connector, settings);
-            let svc = tower_reconnect::Reconnect::new(maker, uri.clone());
-
-            let svc = AddOrigin::new(svc, uri);
-            let svc = BoxService::new(svc);
-            Buffer::new(Box::new(svc) as Inner, 100)
-        } else {
-            let connector = HttpConnector::new();
-            let maker = Connect::new(connector, settings);
-            let svc = tower_reconnect::Reconnect::new(maker, uri.clone());
-
-            let svc = AddOrigin::new(svc, uri);
-
-            let svc = BoxService::new(svc);
-            Buffer::new(Box::new(svc) as Inner, 100)
-        };
-
-        Ok(Channel { svc })
     }
 }

tonic/src/transport/endpoint.rs

@@ -0,0 +1,33 @@
+use super::tls::Cert;
+use http::uri::Uri;
+
+#[derive(Debug, Clone)]
+pub struct Endpoint {
+    uri: Uri,
+    cert: Option<Cert>,
+}
+
+impl Endpoint {
+    pub fn with_pem(uri: Uri, ca: Vec<u8>, domain: Option<String>) -> Self {
+        let domain = domain.unwrap_or_else(|| uri.clone().to_string());
+
+        Self {
+            uri,
+            cert: Some(Cert { ca, domain }),
+        }
+    }
+
+    pub(crate) fn uri(&self) -> &Uri {
+        &self.uri
+    }
+
+    pub(crate) fn take_cert(&mut self) -> Option<Cert> {
+        self.cert.take()
+    }
+}
+
+impl From<Uri> for Endpoint {
+    fn from(uri: Uri) -> Self {
+        Self { uri, cert: None }
+    }
+}

tonic/src/transport/mod.rs

@@ -1,11 +1,10 @@
 mod channel;
-#[cfg(feature = "openssl-1")]
+mod endpoint;
-mod openssl;
-#[cfg(feature = "rustls")]
-mod rustls;
 mod service;
+mod tls;
 
 pub use self::channel::Channel;
+pub use self::endpoint::Endpoint;
 
 use std::{error, fmt};
 

tonic/src/transport/openssl.rs

@@ -1,72 +0,0 @@
-use http::Uri;
-use hyper::client::connect::HttpConnector;
-use openssl::ssl::{SslConnector, SslMethod};
-use openssl::x509::X509;
-use std::{
-    future::Future,
-    pin::Pin,
-    task::{Context, Poll},
-};
-use tokio::net::TcpStream;
-use tokio_openssl::{connect, SslStream};
-use tower_make::MakeConnection;
-use tower_service::Service;
-
-const ALPN_H2: &[u8] = b"\x02h2";
-
-#[derive(Clone)]
-pub struct TlsConnector {
-    http: HttpConnector,
-    config: SslConnector,
-    domain: String,
-}
-
-impl TlsConnector {
-    pub fn new(ca: Vec<u8>, domain: String) -> Result<Self, super::Error> {
-        let mut config = SslConnector::builder(SslMethod::tls()).unwrap();
-
-        config.set_alpn_protos(ALPN_H2).unwrap();
-
-        let ca = X509::from_pem(&ca[..]).unwrap();
-
-        config.cert_store_mut().add_cert(ca).unwrap();
-
-        let config = config.build();
-
-        let mut http = HttpConnector::new();
-        http.enforce_http(false);
-
-        Ok(Self {
-            http,
-            config,
-            domain,
-        })
-    }
-}
-
-impl Service<Uri> for TlsConnector {
-    type Response = SslStream<TcpStream>;
-    type Error = super::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>> {
-        MakeConnection::poll_ready(&mut self.http, cx)
-            .map_err(|e| super::Error::from((super::ErrorKind::Client, e.into())))
-    }
-
-    fn call(&mut self, uri: Uri) -> Self::Future {
-        let config = self.config.configure().unwrap();
-        let tcp = self.http.make_connection(uri.clone());
-        let domain = self.domain.clone();
-
-        let fut = async move {
-            let io = tcp.await.unwrap();
-            let tls = connect(config, &domain, io).await.unwrap();
-            Ok(tls)
-        };
-
-        Box::pin(fut)
-    }
-}

tonic/src/transport/service/connect.rs

@@ -1,5 +1,5 @@
 use super::{add_origin::AddOrigin, connector::Connector};
-use crate::body::BoxBody;
+use crate::{transport::Endpoint, BoxBody};
 use http::{Request, Response, Uri};
 use hyper::client::conn::Builder;
 use hyper::client::service::Connect as HyperConnect;
@@ -15,14 +15,15 @@ pub struct Connection {
 }
 
 impl Connection {
-    pub fn new(uri: Uri) -> Self {
+    pub fn new(mut endpoint: Endpoint) -> Result<Self, crate::Error> {
-        let connector = Connector::new();
+        let connector = Connector::new(endpoint.take_cert())?;
+
         let settings = Builder::new().http2_only(true).clone();
         let connect = HyperConnect::new(connector, settings);
-        let reconnect = Reconnect::new(connect, uri.clone());
+        let reconnect = Reconnect::new(connect, endpoint.uri().clone());
-        let inner = AddOrigin::new(reconnect, uri);
+        let inner = AddOrigin::new(reconnect, endpoint.uri().clone());
 
-        Self { inner }
+        Ok(Self { inner })
     }
 }
 

tonic/src/transport/service/connector.rs

@@ -1,4 +1,5 @@
 use super::io::BoxedIo;
+use crate::transport::tls::{Cert, TlsAcceptor};
 use http::Uri;
 use hyper::client::connect::HttpConnector;
 use std::future::Future;
@@ -11,13 +12,21 @@ type ConnectFuture = <HttpConnector as MakeConnection<Uri>>::Future;
 
 pub struct Connector {
     http: HttpConnector,
+    tls: Option<TlsAcceptor>,
 }
 
 impl Connector {
-    pub fn new() -> Self {
+    pub fn new(cert: Option<Cert>) -> Result<Self, crate::Error> {
-        Self {
+        let mut http = HttpConnector::new();
-            http: HttpConnector::new(),
+        http.enforce_http(false);
-        }
+
+        let tls = if let Some(cert) = cert {
+            Some(TlsAcceptor::new(cert)?)
+        } else {
+            None
+        };
+
+        Ok(Self { http, tls })
     }
 }
 
@@ -33,16 +42,23 @@ impl Service<Uri> for Connector {
     }
 
     fn call(&mut self, uri: Uri) -> Self::Future {
-        let connect_fut = MakeConnection::make_connection(&mut self.http, uri);
+        let io = MakeConnection::make_connection(&mut self.http, uri);
+        let tls = self.tls.clone();
 
-        Box::pin(connect(connect_fut))
+        Box::pin(connect(io, tls))
     }
 }
 
-async fn connect(connect: ConnectFuture) -> Result<BoxedIo, crate::Error> {
+async fn connect(
+    connect: ConnectFuture,
+    tls: Option<TlsAcceptor>,
+) -> Result<BoxedIo, crate::Error> {
     let io = connect.await?;
 
-    // TODO: build tls based on creds and features
+    if let Some(tls) = tls {
-
+        let conn = tls.connect(io).await?;
-    Ok(BoxedIo::new(io))
+        Ok(BoxedIo::new(conn))
+    } else {
+        Ok(BoxedIo::new(io))
+    }
 }

tonic/src/transport/service/discover.rs

@@ -1,17 +1,17 @@
 use super::connect::Connection;
-use http::Uri;
+use crate::transport::Endpoint;
 use std::collections::VecDeque;
 use std::task::{Context, Poll};
 use tower_discover::{Change, Discover};
 
 #[derive(Debug)]
 pub struct ServiceList {
-    list: VecDeque<Uri>,
+    list: VecDeque<Endpoint>,
     i: usize,
 }
 
 impl ServiceList {
-    pub fn new(list: Vec<Uri>) -> Self {
+    pub fn new(list: Vec<Endpoint>) -> Self {
         Self {
             list: list.into(),
             i: 0,
@@ -22,18 +22,21 @@ impl ServiceList {
 impl Discover for ServiceList {
     type Key = usize;
     type Service = Connection;
-    type Error = hyper::Error;
+    type Error = crate::Error;
 
     fn poll(
         &mut self,
         _cx: &mut Context<'_>,
     ) -> Poll<Result<Change<Self::Key, Self::Service>, Self::Error>> {
         match self.list.pop_front() {
-            Some(uri) => {
+            Some(endpoint) => {
                 let i = self.i;
                 self.i += 1;
-                let service = Connection::new(uri);
+
-                Poll::Ready(Ok(Change::Insert(i, service)))
+                match Connection::new(endpoint) {
+                    Ok(svc) => Poll::Ready(Ok(Change::Insert(i, svc))),
+                    Err(e) => Poll::Ready(Err(e)),
+                }
             }
             None => Poll::Pending,
         }

tonic/src/transport/service/mod.rs

@@ -4,8 +4,8 @@ mod connect;
 mod connector;
 mod discover;
 mod io;
-mod tls;
 
 pub use self::add_origin::AddOrigin;
 pub use self::boxed::BoxService;
+pub use self::connect::Connection;
 pub use self::discover::ServiceList;

tonic/src/transport/service/tls/mod.rs

@@ -1 +0,0 @@
-

tonic/src/transport/tls/mod.rs

@@ -0,0 +1,33 @@
+// #[cfg(feature = "openssl-1")]
+// #[cfg(not(feature = "rustls"))]
+// #[path = "rustls.rs"]
+// mod imp;
+
+#[cfg(feature = "native-tls")]
+#[cfg(not(feature = "rustls"))]
+#[path = "openssl.rs"]
+mod imp;
+
+use tokio::net::TcpStream;
+
+#[derive(Debug, Clone)]
+pub struct Cert {
+    pub(crate) ca: Vec<u8>,
+    pub(crate) domain: String,
+}
+
+#[derive(Clone)]
+pub struct TlsAcceptor {
+    inner: imp::TlsAcceptor,
+}
+
+impl TlsAcceptor {
+    pub fn new(cert: Cert) -> Result<Self, crate::Error> {
+        let inner = imp::TlsAcceptor::new(cert)?;
+        Ok(Self { inner })
+    }
+
+    pub async fn connect(&self, io: TcpStream) -> Result<imp::TlsStream, crate::Error> {
+        self.inner.connect(io).await
+    }
+}

tonic/src/transport/tls/openssl.rs

@@ -0,0 +1,42 @@
+use super::Cert;
+use openssl::ssl::{SslConnector, SslMethod};
+use openssl::x509::X509;
+use std::sync::Arc;
+use tokio::net::TcpStream;
+use tokio_openssl::SslStream;
+
+const ALPN_H2: &[u8] = b"\x02h2";
+
+pub type TlsStream = SslStream<TcpStream>;
+
+#[derive(Clone)]
+pub struct TlsAcceptor {
+    config: SslConnector,
+    domain: Arc<String>,
+}
+
+impl TlsAcceptor {
+    pub fn new(cert: Cert) -> Result<Self, crate::Error> {
+        let Cert { ca, domain } = cert;
+        let mut config = SslConnector::builder(SslMethod::tls()).unwrap();
+
+        config.set_alpn_protos(ALPN_H2)?;
+
+        let ca = X509::from_pem(&ca[..])?;
+
+        config.cert_store_mut().add_cert(ca)?;
+
+        let config = config.build();
+
+        Ok(Self {
+            config,
+            domain: Arc::new(domain),
+        })
+    }
+
+    pub async fn connect(&self, io: TcpStream) -> Result<TlsStream, crate::Error> {
+        let config = self.config.configure()?;
+        let tls = tokio_openssl::connect(config, &self.domain, io).await?;
+        Ok(tls)
+    }
+}