merge -connect and -tls and upgrade to rt v2 (#238)

actix-tls/src/connect/connect.rs

@@ -0,0 +1,282 @@
+use std::collections::{vec_deque, VecDeque};
+use std::fmt;
+use std::iter::{FromIterator, FusedIterator};
+use std::net::SocketAddr;
+
+use either::Either;
+
+/// Connect request
+pub trait Address: Unpin + 'static {
+    /// Host name of the request
+    fn host(&self) -> &str;
+
+    /// Port of the request
+    fn port(&self) -> Option<u16>;
+}
+
+impl Address for String {
+    fn host(&self) -> &str {
+        &self
+    }
+
+    fn port(&self) -> Option<u16> {
+        None
+    }
+}
+
+impl Address for &'static str {
+    fn host(&self) -> &str {
+        self
+    }
+
+    fn port(&self) -> Option<u16> {
+        None
+    }
+}
+
+/// Connect request
+#[derive(Eq, PartialEq, Debug, Hash)]
+pub struct Connect<T> {
+    pub(crate) req: T,
+    pub(crate) port: u16,
+    pub(crate) addr: Option<Either<SocketAddr, VecDeque<SocketAddr>>>,
+}
+
+impl<T: Address> Connect<T> {
+    /// Create `Connect` instance by splitting the string by ':' and convert the second part to u16
+    pub fn new(req: T) -> Connect<T> {
+        let (_, port) = parse(req.host());
+        Connect {
+            req,
+            port: port.unwrap_or(0),
+            addr: None,
+        }
+    }
+
+    /// Create new `Connect` instance from host and address. Connector skips name resolution stage
+    /// for such connect messages.
+    pub fn with(req: T, addr: SocketAddr) -> Connect<T> {
+        Connect {
+            req,
+            port: 0,
+            addr: Some(Either::Left(addr)),
+        }
+    }
+
+    /// Use port if address does not provide one.
+    ///
+    /// By default it set to 0
+    pub fn set_port(mut self, port: u16) -> Self {
+        self.port = port;
+        self
+    }
+
+    /// Use address.
+    pub fn set_addr(mut self, addr: Option<SocketAddr>) -> Self {
+        if let Some(addr) = addr {
+            self.addr = Some(Either::Left(addr));
+        }
+        self
+    }
+
+    /// Use addresses.
+    pub fn set_addrs<I>(mut self, addrs: I) -> Self
+    where
+        I: IntoIterator<Item = SocketAddr>,
+    {
+        let mut addrs = VecDeque::from_iter(addrs);
+        self.addr = if addrs.len() < 2 {
+            addrs.pop_front().map(Either::Left)
+        } else {
+            Some(Either::Right(addrs))
+        };
+        self
+    }
+
+    /// Host name
+    pub fn host(&self) -> &str {
+        self.req.host()
+    }
+
+    /// Port of the request
+    pub fn port(&self) -> u16 {
+        self.req.port().unwrap_or(self.port)
+    }
+
+    /// Pre-resolved addresses of the request.
+    pub fn addrs(&self) -> ConnectAddrsIter<'_> {
+        let inner = match self.addr {
+            None => Either::Left(None),
+            Some(Either::Left(addr)) => Either::Left(Some(addr)),
+            Some(Either::Right(ref addrs)) => Either::Right(addrs.iter()),
+        };
+
+        ConnectAddrsIter { inner }
+    }
+
+    /// Takes pre-resolved addresses of the request.
+    pub fn take_addrs(&mut self) -> ConnectTakeAddrsIter {
+        let inner = match self.addr.take() {
+            None => Either::Left(None),
+            Some(Either::Left(addr)) => Either::Left(Some(addr)),
+            Some(Either::Right(addrs)) => Either::Right(addrs.into_iter()),
+        };
+
+        ConnectTakeAddrsIter { inner }
+    }
+}
+
+impl<T: Address> From<T> for Connect<T> {
+    fn from(addr: T) -> Self {
+        Connect::new(addr)
+    }
+}
+
+impl<T: Address> fmt::Display for Connect<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}:{}", self.host(), self.port())
+    }
+}
+
+/// Iterator over addresses in a [`Connect`] request.
+#[derive(Clone)]
+pub struct ConnectAddrsIter<'a> {
+    inner: Either<Option<SocketAddr>, vec_deque::Iter<'a, SocketAddr>>,
+}
+
+impl Iterator for ConnectAddrsIter<'_> {
+    type Item = SocketAddr;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner {
+            Either::Left(ref mut opt) => opt.take(),
+            Either::Right(ref mut iter) => iter.next().copied(),
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        match self.inner {
+            Either::Left(Some(_)) => (1, Some(1)),
+            Either::Left(None) => (0, Some(0)),
+            Either::Right(ref iter) => iter.size_hint(),
+        }
+    }
+}
+
+impl fmt::Debug for ConnectAddrsIter<'_> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_list().entries(self.clone()).finish()
+    }
+}
+
+impl ExactSizeIterator for ConnectAddrsIter<'_> {}
+
+impl FusedIterator for ConnectAddrsIter<'_> {}
+
+/// Owned iterator over addresses in a [`Connect`] request.
+#[derive(Debug)]
+pub struct ConnectTakeAddrsIter {
+    inner: Either<Option<SocketAddr>, vec_deque::IntoIter<SocketAddr>>,
+}
+
+impl Iterator for ConnectTakeAddrsIter {
+    type Item = SocketAddr;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner {
+            Either::Left(ref mut opt) => opt.take(),
+            Either::Right(ref mut iter) => iter.next(),
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        match self.inner {
+            Either::Left(Some(_)) => (1, Some(1)),
+            Either::Left(None) => (0, Some(0)),
+            Either::Right(ref iter) => iter.size_hint(),
+        }
+    }
+}
+
+impl ExactSizeIterator for ConnectTakeAddrsIter {}
+
+impl FusedIterator for ConnectTakeAddrsIter {}
+
+fn parse(host: &str) -> (&str, Option<u16>) {
+    let mut parts_iter = host.splitn(2, ':');
+    if let Some(host) = parts_iter.next() {
+        let port_str = parts_iter.next().unwrap_or("");
+        if let Ok(port) = port_str.parse::<u16>() {
+            (host, Some(port))
+        } else {
+            (host, None)
+        }
+    } else {
+        (host, None)
+    }
+}
+
+pub struct Connection<T, U> {
+    io: U,
+    req: T,
+}
+
+impl<T, U> Connection<T, U> {
+    pub fn new(io: U, req: T) -> Self {
+        Self { io, req }
+    }
+}
+
+impl<T, U> Connection<T, U> {
+    /// Reconstruct from a parts.
+    pub fn from_parts(io: U, req: T) -> Self {
+        Self { io, req }
+    }
+
+    /// Deconstruct into a parts.
+    pub fn into_parts(self) -> (U, T) {
+        (self.io, self.req)
+    }
+
+    /// Replace inclosed object, return new Stream and old object
+    pub fn replace<Y>(self, io: Y) -> (U, Connection<T, Y>) {
+        (self.io, Connection { io, req: self.req })
+    }
+
+    /// Returns a shared reference to the underlying stream.
+    pub fn get_ref(&self) -> &U {
+        &self.io
+    }
+
+    /// Returns a mutable reference to the underlying stream.
+    pub fn get_mut(&mut self) -> &mut U {
+        &mut self.io
+    }
+}
+
+impl<T: Address, U> Connection<T, U> {
+    /// Get request
+    pub fn host(&self) -> &str {
+        &self.req.host()
+    }
+}
+
+impl<T, U> std::ops::Deref for Connection<T, U> {
+    type Target = U;
+
+    fn deref(&self) -> &U {
+        &self.io
+    }
+}
+
+impl<T, U> std::ops::DerefMut for Connection<T, U> {
+    fn deref_mut(&mut self) -> &mut U {
+        &mut self.io
+    }
+}
+
+impl<T, U: fmt::Debug> fmt::Debug for Connection<T, U> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Stream {{{:?}}}", self.io)
+    }
+}

actix-tls/src/connect/connector.rs

@@ -0,0 +1,169 @@
+use std::collections::VecDeque;
+use std::future::Future;
+use std::io;
+use std::marker::PhantomData;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use actix_rt::net::TcpStream;
+use actix_service::{Service, ServiceFactory};
+use futures_util::future::{err, ok, BoxFuture, Either, FutureExt, Ready};
+use log::{error, trace};
+
+use super::connect::{Address, Connect, Connection};
+use super::error::ConnectError;
+
+/// TCP connector service factory
+#[derive(Debug)]
+pub struct TcpConnectorFactory<T>(PhantomData<T>);
+
+impl<T> TcpConnectorFactory<T> {
+    pub fn new() -> Self {
+        TcpConnectorFactory(PhantomData)
+    }
+
+    /// Create TCP connector service
+    pub fn service(&self) -> TcpConnector<T> {
+        TcpConnector(PhantomData)
+    }
+}
+
+impl<T> Default for TcpConnectorFactory<T> {
+    fn default() -> Self {
+        TcpConnectorFactory(PhantomData)
+    }
+}
+
+impl<T> Clone for TcpConnectorFactory<T> {
+    fn clone(&self) -> Self {
+        TcpConnectorFactory(PhantomData)
+    }
+}
+
+impl<T: Address> ServiceFactory<Connect<T>> for TcpConnectorFactory<T> {
+    type Response = Connection<T, TcpStream>;
+    type Error = ConnectError;
+    type Config = ();
+    type Service = TcpConnector<T>;
+    type InitError = ();
+    type Future = Ready<Result<Self::Service, Self::InitError>>;
+
+    fn new_service(&self, _: ()) -> Self::Future {
+        ok(self.service())
+    }
+}
+
+/// TCP connector service
+#[derive(Default, Debug)]
+pub struct TcpConnector<T>(PhantomData<T>);
+
+impl<T> TcpConnector<T> {
+    pub fn new() -> Self {
+        TcpConnector(PhantomData)
+    }
+}
+
+impl<T> Clone for TcpConnector<T> {
+    fn clone(&self) -> Self {
+        TcpConnector(PhantomData)
+    }
+}
+
+impl<T: Address> Service<Connect<T>> for TcpConnector<T> {
+    type Response = Connection<T, TcpStream>;
+    type Error = ConnectError;
+    #[allow(clippy::type_complexity)]
+    type Future = Either<TcpConnectorResponse<T>, Ready<Result<Self::Response, Self::Error>>>;
+
+    actix_service::always_ready!();
+
+    fn call(&mut self, req: Connect<T>) -> Self::Future {
+        let port = req.port();
+        let Connect { req, addr, .. } = req;
+
+        if let Some(addr) = addr {
+            Either::Left(TcpConnectorResponse::new(req, port, addr))
+        } else {
+            error!("TCP connector: got unresolved address");
+            Either::Right(err(ConnectError::Unresolved))
+        }
+    }
+}
+
+#[doc(hidden)]
+/// TCP stream connector response future
+pub struct TcpConnectorResponse<T> {
+    req: Option<T>,
+    port: u16,
+    addrs: Option<VecDeque<SocketAddr>>,
+    stream: Option<BoxFuture<'static, Result<TcpStream, io::Error>>>,
+}
+
+impl<T: Address> TcpConnectorResponse<T> {
+    pub fn new(
+        req: T,
+        port: u16,
+        addr: either::Either<SocketAddr, VecDeque<SocketAddr>>,
+    ) -> TcpConnectorResponse<T> {
+        trace!(
+            "TCP connector - connecting to {:?} port:{}",
+            req.host(),
+            port
+        );
+
+        match addr {
+            either::Either::Left(addr) => TcpConnectorResponse {
+                req: Some(req),
+                port,
+                addrs: None,
+                stream: Some(TcpStream::connect(addr).boxed()),
+            },
+            either::Either::Right(addrs) => TcpConnectorResponse {
+                req: Some(req),
+                port,
+                addrs: Some(addrs),
+                stream: None,
+            },
+        }
+    }
+}
+
+impl<T: Address> Future for TcpConnectorResponse<T> {
+    type Output = Result<Connection<T, TcpStream>, ConnectError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+
+        // connect
+        loop {
+            if let Some(new) = this.stream.as_mut() {
+                match new.as_mut().poll(cx) {
+                    Poll::Ready(Ok(sock)) => {
+                        let req = this.req.take().unwrap();
+                        trace!(
+                            "TCP connector - successfully connected to connecting to {:?} - {:?}",
+                            req.host(), sock.peer_addr()
+                        );
+                        return Poll::Ready(Ok(Connection::new(sock, req)));
+                    }
+                    Poll::Pending => return Poll::Pending,
+                    Poll::Ready(Err(err)) => {
+                        trace!(
+                            "TCP connector - failed to connect to connecting to {:?} port: {}",
+                            this.req.as_ref().unwrap().host(),
+                            this.port,
+                        );
+                        if this.addrs.is_none() || this.addrs.as_ref().unwrap().is_empty() {
+                            return Poll::Ready(Err(err.into()));
+                        }
+                    }
+                }
+            }
+
+            // try to connect
+            let addr = this.addrs.as_mut().unwrap().pop_front().unwrap();
+            this.stream = Some(TcpStream::connect(addr).boxed());
+        }
+    }
+}

actix-tls/src/connect/error.rs

@@ -0,0 +1,26 @@
+use std::io;
+
+use derive_more::{Display, From};
+use trust_dns_resolver::error::ResolveError;
+
+#[derive(Debug, From, Display)]
+pub enum ConnectError {
+    /// Failed to resolve the hostname
+    #[display(fmt = "Failed resolving hostname: {}", _0)]
+    Resolver(ResolveError),
+
+    /// No dns records
+    #[display(fmt = "No dns records found for the input")]
+    NoRecords,
+
+    /// Invalid input
+    InvalidInput,
+
+    /// Unresolved host name
+    #[display(fmt = "Connector received `Connect` method with unresolved host")]
+    Unresolved,
+
+    /// Connection IO error
+    #[display(fmt = "{}", _0)]
+    Io(io::Error),
+}

actix-tls/src/connect/mod.rs

@@ -0,0 +1,103 @@
+//! TCP connector service for Actix ecosystem.
+//!
+//! ## Package feature
+//!
+//! * `openssl` - enables TLS support via `openssl` crate
+//! * `rustls` - enables TLS support via `rustls` crate
+
+mod connect;
+mod connector;
+mod error;
+mod resolve;
+mod service;
+pub mod ssl;
+mod uri;
+
+use actix_rt::{net::TcpStream, Arbiter};
+use actix_service::{pipeline, pipeline_factory, Service, ServiceFactory};
+use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
+use trust_dns_resolver::system_conf::read_system_conf;
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
+
+pub mod resolver {
+    pub use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
+    pub use trust_dns_resolver::system_conf::read_system_conf;
+    pub use trust_dns_resolver::{error::ResolveError, AsyncResolver};
+}
+
+pub use self::connect::{Address, Connect, Connection};
+pub use self::connector::{TcpConnector, TcpConnectorFactory};
+pub use self::error::ConnectError;
+pub use self::resolve::{Resolver, ResolverFactory};
+pub use self::service::{ConnectService, ConnectServiceFactory, TcpConnectService};
+
+pub async fn start_resolver(
+    cfg: ResolverConfig,
+    opts: ResolverOpts,
+) -> Result<AsyncResolver, ConnectError> {
+    Ok(AsyncResolver::tokio(cfg, opts).await?)
+}
+
+struct DefaultResolver(AsyncResolver);
+
+pub(crate) async fn get_default_resolver() -> Result<AsyncResolver, ConnectError> {
+    if Arbiter::contains_item::<DefaultResolver>() {
+        Ok(Arbiter::get_item(|item: &DefaultResolver| item.0.clone()))
+    } else {
+        let (cfg, opts) = match read_system_conf() {
+            Ok((cfg, opts)) => (cfg, opts),
+            Err(e) => {
+                log::error!("TRust-DNS can not load system config: {}", e);
+                (ResolverConfig::default(), ResolverOpts::default())
+            }
+        };
+
+        let resolver = AsyncResolver::tokio(cfg, opts).await?;
+
+        Arbiter::set_item(DefaultResolver(resolver.clone()));
+        Ok(resolver)
+    }
+}
+
+pub async fn start_default_resolver() -> Result<AsyncResolver, ConnectError> {
+    get_default_resolver().await
+}
+
+/// Create TCP connector service.
+pub fn new_connector<T: Address + 'static>(
+    resolver: AsyncResolver,
+) -> impl Service<Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError> + Clone
+{
+    pipeline(Resolver::new(resolver)).and_then(TcpConnector::new())
+}
+
+/// Create TCP connector service factory.
+pub fn new_connector_factory<T: Address + 'static>(
+    resolver: AsyncResolver,
+) -> impl ServiceFactory<
+    Connect<T>,
+    Config = (),
+    Response = Connection<T, TcpStream>,
+    Error = ConnectError,
+    InitError = (),
+> + Clone {
+    pipeline_factory(ResolverFactory::new(resolver)).and_then(TcpConnectorFactory::new())
+}
+
+/// Create connector service with default parameters.
+pub fn default_connector<T: Address + 'static>(
+) -> impl Service<Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError> + Clone
+{
+    pipeline(Resolver::default()).and_then(TcpConnector::new())
+}
+
+/// Create connector service factory with default parameters.
+pub fn default_connector_factory<T: Address + 'static>() -> impl ServiceFactory<
+    Connect<T>,
+    Config = (),
+    Response = Connection<T, TcpStream>,
+    Error = ConnectError,
+    InitError = (),
+> + Clone {
+    pipeline_factory(ResolverFactory::default()).and_then(TcpConnectorFactory::new())
+}

actix-tls/src/connect/resolve.rs

@@ -0,0 +1,205 @@
+use std::future::Future;
+use std::marker::PhantomData;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use actix_service::{Service, ServiceFactory};
+use futures_util::future::{ok, Either, Ready};
+use log::trace;
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
+use trust_dns_resolver::{error::ResolveError, lookup_ip::LookupIp};
+
+use super::connect::{Address, Connect};
+use super::error::ConnectError;
+use super::get_default_resolver;
+
+/// DNS Resolver Service factory
+pub struct ResolverFactory<T> {
+    resolver: Option<AsyncResolver>,
+    _t: PhantomData<T>,
+}
+
+impl<T> ResolverFactory<T> {
+    /// Create new resolver instance with custom configuration and options.
+    pub fn new(resolver: AsyncResolver) -> Self {
+        ResolverFactory {
+            resolver: Some(resolver),
+            _t: PhantomData,
+        }
+    }
+
+    pub fn service(&self) -> Resolver<T> {
+        Resolver {
+            resolver: self.resolver.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T> Default for ResolverFactory<T> {
+    fn default() -> Self {
+        ResolverFactory {
+            resolver: None,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T> Clone for ResolverFactory<T> {
+    fn clone(&self) -> Self {
+        ResolverFactory {
+            resolver: self.resolver.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T: Address> ServiceFactory<Connect<T>> for ResolverFactory<T> {
+    type Response = Connect<T>;
+    type Error = ConnectError;
+    type Config = ();
+    type Service = Resolver<T>;
+    type InitError = ();
+    type Future = Ready<Result<Self::Service, Self::InitError>>;
+
+    fn new_service(&self, _: ()) -> Self::Future {
+        ok(self.service())
+    }
+}
+
+/// DNS Resolver Service
+pub struct Resolver<T> {
+    resolver: Option<AsyncResolver>,
+    _t: PhantomData<T>,
+}
+
+impl<T> Resolver<T> {
+    /// Create new resolver instance with custom configuration and options.
+    pub fn new(resolver: AsyncResolver) -> Self {
+        Resolver {
+            resolver: Some(resolver),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T> Default for Resolver<T> {
+    fn default() -> Self {
+        Resolver {
+            resolver: None,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T> Clone for Resolver<T> {
+    fn clone(&self) -> Self {
+        Resolver {
+            resolver: self.resolver.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T: Address> Service<Connect<T>> for Resolver<T> {
+    type Response = Connect<T>;
+    type Error = ConnectError;
+    #[allow(clippy::type_complexity)]
+    type Future = Either<
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>,
+        Ready<Result<Connect<T>, Self::Error>>,
+    >;
+
+    actix_service::always_ready!();
+
+    fn call(&mut self, mut req: Connect<T>) -> Self::Future {
+        if req.addr.is_some() {
+            Either::Right(ok(req))
+        } else if let Ok(ip) = req.host().parse() {
+            req.addr = Some(either::Either::Left(SocketAddr::new(ip, req.port())));
+            Either::Right(ok(req))
+        } else {
+            let resolver = self.resolver.as_ref().map(AsyncResolver::clone);
+            Either::Left(Box::pin(async move {
+                trace!("DNS resolver: resolving host {:?}", req.host());
+                let resolver = if let Some(resolver) = resolver {
+                    resolver
+                } else {
+                    get_default_resolver()
+                        .await
+                        .expect("Failed to get default resolver")
+                };
+                ResolverFuture::new(req, &resolver).await
+            }))
+        }
+    }
+}
+
+type LookupIpFuture = Pin<Box<dyn Future<Output = Result<LookupIp, ResolveError>>>>;
+
+#[doc(hidden)]
+/// Resolver future
+pub struct ResolverFuture<T: Address> {
+    req: Option<Connect<T>>,
+    lookup: LookupIpFuture,
+}
+
+impl<T: Address> ResolverFuture<T> {
+    pub fn new(req: Connect<T>, resolver: &AsyncResolver) -> Self {
+        let host = if let Some(host) = req.host().splitn(2, ':').next() {
+            host
+        } else {
+            req.host()
+        };
+
+        // Clone data to be moved to the lookup future
+        let host_clone = host.to_owned();
+        let resolver_clone = resolver.clone();
+
+        ResolverFuture {
+            lookup: Box::pin(async move {
+                let resolver = resolver_clone;
+                resolver.lookup_ip(host_clone).await
+            }),
+            req: Some(req),
+        }
+    }
+}
+
+impl<T: Address> Future for ResolverFuture<T> {
+    type Output = Result<Connect<T>, ConnectError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+
+        match Pin::new(&mut this.lookup).poll(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Ok(ips)) => {
+                let req = this.req.take().unwrap();
+                let port = req.port();
+                let req = req.set_addrs(ips.iter().map(|ip| SocketAddr::new(ip, port)));
+
+                trace!(
+                    "DNS resolver: host {:?} resolved to {:?}",
+                    req.host(),
+                    req.addrs()
+                );
+
+                if req.addr.is_none() {
+                    Poll::Ready(Err(ConnectError::NoRecords))
+                } else {
+                    Poll::Ready(Ok(req))
+                }
+            }
+            Poll::Ready(Err(e)) => {
+                trace!(
+                    "DNS resolver: failed to resolve host {:?} err: {}",
+                    this.req.as_ref().unwrap().host(),
+                    e
+                );
+                Poll::Ready(Err(e.into()))
+            }
+        }
+    }
+}

actix-tls/src/connect/service.rs

@@ -0,0 +1,225 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use actix_rt::net::TcpStream;
+use actix_service::{Service, ServiceFactory};
+use either::Either;
+use futures_util::future::{ok, Ready};
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
+
+use super::connect::{Address, Connect, Connection};
+use super::connector::{TcpConnector, TcpConnectorFactory};
+use super::error::ConnectError;
+use super::resolve::{Resolver, ResolverFactory};
+
+pub struct ConnectServiceFactory<T> {
+    tcp: TcpConnectorFactory<T>,
+    resolver: ResolverFactory<T>,
+}
+
+impl<T> ConnectServiceFactory<T> {
+    /// Construct new ConnectService factory
+    pub fn new() -> Self {
+        ConnectServiceFactory {
+            tcp: TcpConnectorFactory::default(),
+            resolver: ResolverFactory::default(),
+        }
+    }
+
+    /// Construct new connect service with custom dns resolver
+    pub fn with_resolver(resolver: AsyncResolver) -> Self {
+        ConnectServiceFactory {
+            tcp: TcpConnectorFactory::default(),
+            resolver: ResolverFactory::new(resolver),
+        }
+    }
+
+    /// Construct new service
+    pub fn service(&self) -> ConnectService<T> {
+        ConnectService {
+            tcp: self.tcp.service(),
+            resolver: self.resolver.service(),
+        }
+    }
+
+    /// Construct new tcp stream service
+    pub fn tcp_service(&self) -> TcpConnectService<T> {
+        TcpConnectService {
+            tcp: self.tcp.service(),
+            resolver: self.resolver.service(),
+        }
+    }
+}
+
+impl<T> Default for ConnectServiceFactory<T> {
+    fn default() -> Self {
+        ConnectServiceFactory {
+            tcp: TcpConnectorFactory::default(),
+            resolver: ResolverFactory::default(),
+        }
+    }
+}
+
+impl<T> Clone for ConnectServiceFactory<T> {
+    fn clone(&self) -> Self {
+        ConnectServiceFactory {
+            tcp: self.tcp.clone(),
+            resolver: self.resolver.clone(),
+        }
+    }
+}
+
+impl<T: Address> ServiceFactory<Connect<T>> for ConnectServiceFactory<T> {
+    type Response = Connection<T, TcpStream>;
+    type Error = ConnectError;
+    type Config = ();
+    type Service = ConnectService<T>;
+    type InitError = ();
+    type Future = Ready<Result<Self::Service, Self::InitError>>;
+
+    fn new_service(&self, _: ()) -> Self::Future {
+        ok(self.service())
+    }
+}
+
+#[derive(Clone)]
+pub struct ConnectService<T> {
+    tcp: TcpConnector<T>,
+    resolver: Resolver<T>,
+}
+
+impl<T: Address> Service<Connect<T>> for ConnectService<T> {
+    type Response = Connection<T, TcpStream>;
+    type Error = ConnectError;
+    type Future = ConnectServiceResponse<T>;
+
+    actix_service::always_ready!();
+
+    fn call(&mut self, req: Connect<T>) -> Self::Future {
+        ConnectServiceResponse {
+            state: ConnectState::Resolve(self.resolver.call(req)),
+            tcp: self.tcp.clone(),
+        }
+    }
+}
+
+enum ConnectState<T: Address> {
+    Resolve(<Resolver<T> as Service<Connect<T>>>::Future),
+    Connect(<TcpConnector<T> as Service<Connect<T>>>::Future),
+}
+
+impl<T: Address> ConnectState<T> {
+    #[allow(clippy::type_complexity)]
+    fn poll(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Either<Poll<Result<Connection<T, TcpStream>, ConnectError>>, Connect<T>> {
+        match self {
+            ConnectState::Resolve(ref mut fut) => match Pin::new(fut).poll(cx) {
+                Poll::Pending => Either::Left(Poll::Pending),
+                Poll::Ready(Ok(res)) => Either::Right(res),
+                Poll::Ready(Err(err)) => Either::Left(Poll::Ready(Err(err))),
+            },
+            ConnectState::Connect(ref mut fut) => Either::Left(Pin::new(fut).poll(cx)),
+        }
+    }
+}
+
+pub struct ConnectServiceResponse<T: Address> {
+    state: ConnectState<T>,
+    tcp: TcpConnector<T>,
+}
+
+impl<T: Address> Future for ConnectServiceResponse<T> {
+    type Output = Result<Connection<T, TcpStream>, ConnectError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let res = match self.state.poll(cx) {
+            Either::Right(res) => {
+                self.state = ConnectState::Connect(self.tcp.call(res));
+                self.state.poll(cx)
+            }
+            Either::Left(res) => return res,
+        };
+
+        match res {
+            Either::Left(res) => res,
+            Either::Right(_) => panic!(),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct TcpConnectService<T> {
+    tcp: TcpConnector<T>,
+    resolver: Resolver<T>,
+}
+
+impl<T: Address + 'static> Service<Connect<T>> for TcpConnectService<T> {
+    type Response = TcpStream;
+    type Error = ConnectError;
+    type Future = TcpConnectServiceResponse<T>;
+
+    actix_service::always_ready!();
+
+    fn call(&mut self, req: Connect<T>) -> Self::Future {
+        TcpConnectServiceResponse {
+            state: TcpConnectState::Resolve(self.resolver.call(req)),
+            tcp: self.tcp.clone(),
+        }
+    }
+}
+
+enum TcpConnectState<T: Address> {
+    Resolve(<Resolver<T> as Service<Connect<T>>>::Future),
+    Connect(<TcpConnector<T> as Service<Connect<T>>>::Future),
+}
+
+impl<T: Address> TcpConnectState<T> {
+    fn poll(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Either<Poll<Result<TcpStream, ConnectError>>, Connect<T>> {
+        match self {
+            TcpConnectState::Resolve(ref mut fut) => match Pin::new(fut).poll(cx) {
+                Poll::Pending => (),
+                Poll::Ready(Ok(res)) => return Either::Right(res),
+                Poll::Ready(Err(err)) => return Either::Left(Poll::Ready(Err(err))),
+            },
+            TcpConnectState::Connect(ref mut fut) => {
+                if let Poll::Ready(res) = Pin::new(fut).poll(cx) {
+                    return match res {
+                        Ok(conn) => Either::Left(Poll::Ready(Ok(conn.into_parts().0))),
+                        Err(err) => Either::Left(Poll::Ready(Err(err))),
+                    };
+                }
+            }
+        }
+        Either::Left(Poll::Pending)
+    }
+}
+
+pub struct TcpConnectServiceResponse<T: Address> {
+    state: TcpConnectState<T>,
+    tcp: TcpConnector<T>,
+}
+
+impl<T: Address> Future for TcpConnectServiceResponse<T> {
+    type Output = Result<TcpStream, ConnectError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let res = match self.state.poll(cx) {
+            Either::Right(res) => {
+                self.state = TcpConnectState::Connect(self.tcp.call(res));
+                self.state.poll(cx)
+            }
+            Either::Left(res) => return res,
+        };
+
+        match res {
+            Either::Left(res) => res,
+            Either::Right(_) => panic!(),
+        }
+    }
+}

actix-tls/src/connect/ssl/mod.rs

@@ -0,0 +1,7 @@
+//! SSL Services
+
+#[cfg(feature = "openssl")]
+pub mod openssl;
+
+#[cfg(feature = "rustls")]
+pub mod rustls;

actix-tls/src/connect/ssl/openssl.rs

@@ -0,0 +1,269 @@
+use std::future::Future;
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::{fmt, io};
+
+use actix_codec::{AsyncRead, AsyncWrite};
+use actix_rt::net::TcpStream;
+use actix_service::{Service, ServiceFactory};
+use futures_util::{
+    future::{err, ok, Either, Ready},
+    ready,
+};
+use log::trace;
+pub use openssl::ssl::{Error as SslError, HandshakeError, SslConnector, SslMethod};
+pub use tokio_openssl::SslStream;
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
+
+use crate::connect::{
+    Address, Connect, ConnectError, ConnectService, ConnectServiceFactory, Connection,
+};
+
+/// OpenSSL connector factory
+pub struct OpensslConnector<T, U> {
+    connector: SslConnector,
+    _t: PhantomData<(T, U)>,
+}
+
+impl<T, U> OpensslConnector<T, U> {
+    pub fn new(connector: SslConnector) -> Self {
+        OpensslConnector {
+            connector,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, U> OpensslConnector<T, U>
+where
+    T: Address + 'static,
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
+{
+    pub fn service(connector: SslConnector) -> OpensslConnectorService<T, U> {
+        OpensslConnectorService {
+            connector,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, U> Clone for OpensslConnector<T, U> {
+    fn clone(&self) -> Self {
+        Self {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, U> ServiceFactory<Connection<T, U>> for OpensslConnector<T, U>
+where
+    T: Address + 'static,
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
+{
+    type Response = Connection<T, SslStream<U>>;
+    type Error = io::Error;
+    type Config = ();
+    type Service = OpensslConnectorService<T, U>;
+    type InitError = ();
+    type Future = Ready<Result<Self::Service, Self::InitError>>;
+
+    fn new_service(&self, _: ()) -> Self::Future {
+        ok(OpensslConnectorService {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        })
+    }
+}
+
+pub struct OpensslConnectorService<T, U> {
+    connector: SslConnector,
+    _t: PhantomData<(T, U)>,
+}
+
+impl<T, U> Clone for OpensslConnectorService<T, U> {
+    fn clone(&self) -> Self {
+        Self {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, U> Service<Connection<T, U>> for OpensslConnectorService<T, U>
+where
+    T: Address + 'static,
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
+{
+    type Response = Connection<T, SslStream<U>>;
+    type Error = io::Error;
+    #[allow(clippy::type_complexity)]
+    type Future = Either<ConnectAsyncExt<T, U>, Ready<Result<Self::Response, Self::Error>>>;
+
+    actix_service::always_ready!();
+
+    fn call(&mut self, stream: Connection<T, U>) -> Self::Future {
+        trace!("SSL Handshake start for: {:?}", stream.host());
+        let (io, stream) = stream.replace(());
+        let host = stream.host().to_string();
+
+        match self.connector.configure() {
+            Err(e) => Either::Right(err(io::Error::new(io::ErrorKind::Other, e))),
+            Ok(config) => {
+                let ssl = config
+                    .into_ssl(&host)
+                    .expect("SSL connect configuration was invalid.");
+
+                Either::Left(ConnectAsyncExt {
+                    io: Some(SslStream::new(ssl, io).unwrap()),
+                    stream: Some(stream),
+                    _t: PhantomData,
+                })
+            }
+        }
+    }
+}
+
+pub struct ConnectAsyncExt<T, U> {
+    io: Option<SslStream<U>>,
+    stream: Option<Connection<T, ()>>,
+    _t: PhantomData<U>,
+}
+
+impl<T: Address, U> Future for ConnectAsyncExt<T, U>
+where
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
+{
+    type Output = Result<Connection<T, SslStream<U>>, io::Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+
+        match ready!(Pin::new(this.io.as_mut().unwrap()).poll_connect(cx)) {
+            Ok(_) => {
+                let stream = this.stream.take().unwrap();
+                trace!("SSL Handshake success: {:?}", stream.host());
+                Poll::Ready(Ok(stream.replace(this.io.take().unwrap()).1))
+            }
+            Err(e) => {
+                trace!("SSL Handshake error: {:?}", e);
+                Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, format!("{}", e))))
+            }
+        }
+    }
+}
+
+pub struct OpensslConnectServiceFactory<T> {
+    tcp: ConnectServiceFactory<T>,
+    openssl: OpensslConnector<T, TcpStream>,
+}
+
+impl<T> OpensslConnectServiceFactory<T> {
+    /// Construct new OpensslConnectService factory
+    pub fn new(connector: SslConnector) -> Self {
+        OpensslConnectServiceFactory {
+            tcp: ConnectServiceFactory::default(),
+            openssl: OpensslConnector::new(connector),
+        }
+    }
+
+    /// Construct new connect service with custom DNS resolver
+    pub fn with_resolver(connector: SslConnector, resolver: AsyncResolver) -> Self {
+        OpensslConnectServiceFactory {
+            tcp: ConnectServiceFactory::with_resolver(resolver),
+            openssl: OpensslConnector::new(connector),
+        }
+    }
+
+    /// Construct OpenSSL connect service
+    pub fn service(&self) -> OpensslConnectService<T> {
+        OpensslConnectService {
+            tcp: self.tcp.service(),
+            openssl: OpensslConnectorService {
+                connector: self.openssl.connector.clone(),
+                _t: PhantomData,
+            },
+        }
+    }
+}
+
+impl<T> Clone for OpensslConnectServiceFactory<T> {
+    fn clone(&self) -> Self {
+        OpensslConnectServiceFactory {
+            tcp: self.tcp.clone(),
+            openssl: self.openssl.clone(),
+        }
+    }
+}
+
+impl<T: Address + 'static> ServiceFactory<Connect<T>> for OpensslConnectServiceFactory<T> {
+    type Response = SslStream<TcpStream>;
+    type Error = ConnectError;
+    type Config = ();
+    type Service = OpensslConnectService<T>;
+    type InitError = ();
+    type Future = Ready<Result<Self::Service, Self::InitError>>;
+
+    fn new_service(&self, _: ()) -> Self::Future {
+        ok(self.service())
+    }
+}
+
+#[derive(Clone)]
+pub struct OpensslConnectService<T> {
+    tcp: ConnectService<T>,
+    openssl: OpensslConnectorService<T, TcpStream>,
+}
+
+impl<T: Address + 'static> Service<Connect<T>> for OpensslConnectService<T> {
+    type Response = SslStream<TcpStream>;
+    type Error = ConnectError;
+    type Future = OpensslConnectServiceResponse<T>;
+
+    actix_service::always_ready!();
+
+    fn call(&mut self, req: Connect<T>) -> Self::Future {
+        OpensslConnectServiceResponse {
+            fut1: Some(self.tcp.call(req)),
+            fut2: None,
+            openssl: self.openssl.clone(),
+        }
+    }
+}
+
+pub struct OpensslConnectServiceResponse<T: Address + 'static> {
+    fut1: Option<<ConnectService<T> as Service<Connect<T>>>::Future>,
+    fut2: Option<
+        <OpensslConnectorService<T, TcpStream> as Service<Connection<T, TcpStream>>>::Future,
+    >,
+    openssl: OpensslConnectorService<T, TcpStream>,
+}
+
+impl<T: Address> Future for OpensslConnectServiceResponse<T> {
+    type Output = Result<SslStream<TcpStream>, ConnectError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if let Some(ref mut fut) = self.fut1 {
+            match ready!(Pin::new(fut).poll(cx)) {
+                Ok(res) => {
+                    let _ = self.fut1.take();
+                    self.fut2 = Some(self.openssl.call(res));
+                }
+                Err(e) => return Poll::Ready(Err(e)),
+            }
+        }
+
+        if let Some(ref mut fut) = self.fut2 {
+            match ready!(Pin::new(fut).poll(cx)) {
+                Ok(connect) => Poll::Ready(Ok(connect.into_parts().0)),
+                Err(e) => Poll::Ready(Err(ConnectError::Io(io::Error::new(
+                    io::ErrorKind::Other,
+                    e,
+                )))),
+            }
+        } else {
+            Poll::Pending
+        }
+    }
+}

actix-tls/src/connect/ssl/rustls.rs

@@ -0,0 +1,133 @@
+use std::fmt;
+use std::future::Future;
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+pub use rustls::Session;
+pub use tokio_rustls::{client::TlsStream, rustls::ClientConfig};
+
+use actix_codec::{AsyncRead, AsyncWrite};
+use actix_service::{Service, ServiceFactory};
+use futures_util::future::{ok, Ready};
+use log::trace;
+use tokio_rustls::{Connect, TlsConnector};
+use webpki::DNSNameRef;
+
+use crate::connect::{Address, Connection};
+
+/// Rustls connector factory
+pub struct RustlsConnector<T, U> {
+    connector: Arc<ClientConfig>,
+    _t: PhantomData<(T, U)>,
+}
+
+impl<T, U> RustlsConnector<T, U> {
+    pub fn new(connector: Arc<ClientConfig>) -> Self {
+        RustlsConnector {
+            connector,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, U> RustlsConnector<T, U>
+where
+    T: Address,
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
+{
+    pub fn service(connector: Arc<ClientConfig>) -> RustlsConnectorService<T, U> {
+        RustlsConnectorService {
+            connector,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, U> Clone for RustlsConnector<T, U> {
+    fn clone(&self) -> Self {
+        Self {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T: Address, U> ServiceFactory<Connection<T, U>> for RustlsConnector<T, U>
+where
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
+{
+    type Response = Connection<T, TlsStream<U>>;
+    type Error = std::io::Error;
+    type Config = ();
+    type Service = RustlsConnectorService<T, U>;
+    type InitError = ();
+    type Future = Ready<Result<Self::Service, Self::InitError>>;
+
+    fn new_service(&self, _: ()) -> Self::Future {
+        ok(RustlsConnectorService {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        })
+    }
+}
+
+pub struct RustlsConnectorService<T, U> {
+    connector: Arc<ClientConfig>,
+    _t: PhantomData<(T, U)>,
+}
+
+impl<T, U> Clone for RustlsConnectorService<T, U> {
+    fn clone(&self) -> Self {
+        Self {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T: Address, U> Service<Connection<T, U>> for RustlsConnectorService<T, U>
+where
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
+{
+    type Response = Connection<T, TlsStream<U>>;
+    type Error = std::io::Error;
+    type Future = ConnectAsyncExt<T, U>;
+
+    actix_service::always_ready!();
+
+    fn call(&mut self, stream: Connection<T, U>) -> Self::Future {
+        trace!("SSL Handshake start for: {:?}", stream.host());
+        let (io, stream) = stream.replace(());
+        let host = DNSNameRef::try_from_ascii_str(stream.host())
+            .expect("rustls currently only handles hostname-based connections. See https://github.com/briansmith/webpki/issues/54");
+        ConnectAsyncExt {
+            fut: TlsConnector::from(self.connector.clone()).connect(host, io),
+            stream: Some(stream),
+        }
+    }
+}
+
+pub struct ConnectAsyncExt<T, U> {
+    fut: Connect<U>,
+    stream: Option<Connection<T, ()>>,
+}
+
+impl<T: Address, U> Future for ConnectAsyncExt<T, U>
+where
+    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
+{
+    type Output = Result<Connection<T, TlsStream<U>>, std::io::Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+        Poll::Ready(
+            futures_util::ready!(Pin::new(&mut this.fut).poll(cx)).map(|stream| {
+                let s = this.stream.take().unwrap();
+                trace!("SSL Handshake success: {:?}", s.host());
+                s.replace(stream).1
+            }),
+        )
+    }
+}

actix-tls/src/connect/uri.rs

@@ -0,0 +1,37 @@
+use http::Uri;
+
+use super::Address;
+
+impl Address for Uri {
+    fn host(&self) -> &str {
+        self.host().unwrap_or("")
+    }
+
+    fn port(&self) -> Option<u16> {
+        if let Some(port) = self.port_u16() {
+            Some(port)
+        } else {
+            port(self.scheme_str())
+        }
+    }
+}
+
+// TODO: load data from file
+fn port(scheme: Option<&str>) -> Option<u16> {
+    if let Some(scheme) = scheme {
+        match scheme {
+            "http" => Some(80),
+            "https" => Some(443),
+            "ws" => Some(80),
+            "wss" => Some(443),
+            "amqp" => Some(5672),
+            "amqps" => Some(5671),
+            "sb" => Some(5671),
+            "mqtt" => Some(1883),
+            "mqtts" => Some(8883),
+            _ => None,
+        }
+    } else {
+        None
+    }
+}