rename crate

actix-connect/src/connect.rs

@@ -0,0 +1,161 @@
+use std::collections::VecDeque;
+use std::fmt;
+use std::net::SocketAddr;
+
+use either::Either;
+
+use crate::error::ConnectError;
+
+/// Connect request
+#[derive(Eq, PartialEq, Debug, Hash)]
+pub enum Connect {
+    /// Host name
+    Host { host: String, port: u16 },
+    /// Host name with address of this host
+    Addr {
+        host: String,
+        addr: Either<SocketAddr, VecDeque<SocketAddr>>,
+    },
+}
+
+impl Connect {
+    /// Create new `Connect` instance.
+    pub fn new<T: AsRef<str>>(host: T, port: u16) -> Connect {
+        Connect::Host {
+            host: host.as_ref().to_owned(),
+            port,
+        }
+    }
+
+    /// Create `Connect` instance by spliting the string by ':' and convert the second part to u16
+    pub fn with<T: AsRef<str>>(host: T) -> Result<Connect, ConnectError> {
+        let mut parts_iter = host.as_ref().splitn(2, ':');
+        let host = parts_iter.next().ok_or(ConnectError::InvalidInput)?;
+        let port_str = parts_iter.next().unwrap_or("");
+        let port = port_str
+            .parse::<u16>()
+            .map_err(|_| ConnectError::InvalidInput)?;
+        Ok(Connect::Host {
+            host: host.to_owned(),
+            port,
+        })
+    }
+
+    /// Create new `Connect` instance from host and address. Connector skips name resolution stage for such connect messages.
+    pub fn with_address<T: Into<String>>(host: T, addr: SocketAddr) -> Connect {
+        Connect::Addr {
+            addr: Either::Left(addr),
+            host: host.into(),
+        }
+    }
+
+    /// Host name
+    fn host(&self) -> &str {
+        match self {
+            Connect::Host { ref host, .. } => host,
+            Connect::Addr { ref host, .. } => host,
+        }
+    }
+}
+
+impl fmt::Display for Connect {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{}:{}", self.host(), 0)
+    }
+}
+
+pub struct Stream<T, P = ()> {
+    io: T,
+    host: String,
+    params: P,
+}
+
+impl<T> Stream<T, ()> {
+    pub fn new(io: T, host: String) -> Self {
+        Self {
+            io,
+            host,
+            params: (),
+        }
+    }
+}
+
+impl<T, P> Stream<T, P> {
+    /// Reconstruct from a parts.
+    pub fn from_parts(io: T, host: String, params: P) -> Self {
+        Self { io, params, host }
+    }
+
+    /// Deconstruct into a parts.
+    pub fn into_parts(self) -> (T, String, P) {
+        (self.io, self.host, self.params)
+    }
+
+    /// Replace inclosed object, return new Stream and old object
+    pub fn replace<U>(self, io: U) -> (T, Stream<U, P>) {
+        (
+            self.io,
+            Stream {
+                io,
+                host: self.host,
+                params: self.params,
+            },
+        )
+    }
+
+    /// Returns a shared reference to the underlying stream.
+    pub fn get_ref(&self) -> &T {
+        &self.io
+    }
+
+    /// Returns a mutable reference to the underlying stream.
+    pub fn get_mut(&mut self) -> &mut T {
+        &mut self.io
+    }
+
+    /// Get host name
+    pub fn host(&self) -> &str {
+        &self.host
+    }
+
+    /// Return new Io object with new parameter.
+    pub fn set<U>(self, params: U) -> Stream<T, U> {
+        Stream {
+            io: self.io,
+            host: self.host,
+            params: params,
+        }
+    }
+
+    /// Maps an Io<_, P> to Io<_, U> by applying a function to a contained value.
+    pub fn map<U, F>(self, op: F) -> Stream<T, U>
+    where
+        F: FnOnce(P) -> U,
+    {
+        Stream {
+            io: self.io,
+            host: self.host,
+            params: op(self.params),
+        }
+    }
+}
+
+impl<T, P> std::ops::Deref for Stream<T, P> {
+    type Target = T;
+
+    fn deref(&self) -> &T {
+        &self.io
+    }
+}
+
+impl<T, P> std::ops::DerefMut for Stream<T, P> {
+    fn deref_mut(&mut self) -> &mut T {
+        &mut self.io
+    }
+}
+
+impl<T: fmt::Debug, P> fmt::Debug for Stream<T, P> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "Stream {{{:?}}}", self.io)
+    }
+}

actix-connect/src/connector.rs

@@ -0,0 +1,166 @@
+use std::collections::VecDeque;
+use std::net::SocketAddr;
+
+use actix_service::{NewService, Service};
+use futures::future::{err, ok, Either, FutureResult};
+use futures::{Async, Future, Poll};
+use tokio_tcp::{ConnectFuture, TcpStream};
+
+use super::connect::{Connect, Stream};
+use super::error::ConnectError;
+
+/// Tcp connector service factory
+#[derive(Copy, Clone, Debug)]
+pub struct ConnectorFactory;
+
+impl NewService for ConnectorFactory {
+    type Request = Connect;
+    type Response = Stream<TcpStream>;
+    type Error = ConnectError;
+    type Service = Connector;
+    type InitError = ();
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self, _: &()) -> Self::Future {
+        ok(Connector)
+    }
+}
+
+/// Tcp connector service
+#[derive(Copy, Clone, Debug)]
+pub struct Connector;
+
+impl Service for Connector {
+    type Request = Connect;
+    type Response = Stream<TcpStream>;
+    type Error = ConnectError;
+    type Future = Either<ConnectorResponse, FutureResult<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: Connect) -> Self::Future {
+        match req {
+            Connect::Host { .. } => {
+                error!("TCP connector: got unresolved address");
+                Either::B(err(ConnectError::Unresolverd))
+            }
+            Connect::Addr { host, addr } => Either::A(ConnectorResponse::new(host, addr)),
+        }
+    }
+}
+
+#[doc(hidden)]
+/// Tcp stream connector response future
+pub struct ConnectorResponse {
+    host: Option<String>,
+    addrs: Option<VecDeque<SocketAddr>>,
+    stream: Option<ConnectFuture>,
+}
+
+impl ConnectorResponse {
+    pub fn new(
+        host: String,
+        addr: either::Either<SocketAddr, VecDeque<SocketAddr>>,
+    ) -> ConnectorResponse {
+        trace!("TCP connector - connecting to {:?}", host);
+
+        match addr {
+            either::Either::Left(addr) => ConnectorResponse {
+                host: Some(host),
+                addrs: None,
+                stream: Some(TcpStream::connect(&addr)),
+            },
+            either::Either::Right(addrs) => ConnectorResponse {
+                host: Some(host),
+                addrs: Some(addrs),
+                stream: None,
+            },
+        }
+    }
+}
+
+impl Future for ConnectorResponse {
+    type Item = Stream<TcpStream>;
+    type Error = ConnectError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        // connect
+        loop {
+            if let Some(new) = self.stream.as_mut() {
+                match new.poll() {
+                    Ok(Async::Ready(sock)) => {
+                        let host = self.host.take().unwrap();
+                        trace!(
+                            "TCP connector - successfully connected to connecting to {:?} - {:?}",
+                            host, sock.peer_addr()
+                        );
+                        return Ok(Async::Ready(Stream::new(sock, self.host.take().unwrap())));
+                    }
+                    Ok(Async::NotReady) => return Ok(Async::NotReady),
+                    Err(err) => {
+                        trace!(
+                            "TCP connector - failed to connect to connecting to {:?}",
+                            self.host.as_ref().unwrap()
+                        );
+                        if self.addrs.as_ref().unwrap().is_empty() {
+                            return Err(err.into());
+                        }
+                    }
+                }
+            }
+
+            // try to connect
+            self.stream = Some(TcpStream::connect(
+                &self.addrs.as_mut().unwrap().pop_front().unwrap(),
+            ));
+        }
+    }
+}
+
+// #[derive(Clone)]
+// pub struct DefaultConnector(Connector);
+
+// impl Default for DefaultConnector {
+//     fn default() -> Self {
+//         DefaultConnector(Connector::default())
+//     }
+// }
+
+// impl DefaultConnector {
+//     pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
+//         DefaultConnector(Connector::new(cfg, opts))
+//     }
+// }
+
+// impl Service for DefaultConnector {
+//     type Request = Connect;
+//     type Response = TcpStream;
+//     type Error = ConnectorError;
+//     type Future = DefaultConnectorFuture;
+
+//     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+//         self.0.poll_ready()
+//     }
+
+//     fn call(&mut self, req: Connect) -> Self::Future {
+//         DefaultConnectorFuture {
+//             fut: self.0.call(req),
+//         }
+//     }
+// }
+
+// #[doc(hidden)]
+// pub struct DefaultConnectorFuture {
+//     fut: Either<ConnectorFuture, ConnectorTcpFuture>,
+// }
+
+// impl Future for DefaultConnectorFuture {
+//     type Item = TcpStream;
+//     type Error = ConnectorError;
+
+//     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+//         Ok(Async::Ready(try_ready!(self.fut.poll()).1))
+//     }
+// }

actix-connect/src/error.rs

@@ -0,0 +1,30 @@
+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,
+
+    /// Connecting took too long
+    #[display(fmt = "Timeout out while establishing connection")]
+    Timeout,
+
+    /// Invalid input
+    InvalidInput,
+
+    /// Unresolved host name
+    #[display(fmt = "Connector received `Connect` method with unresolved host")]
+    Unresolverd,
+
+    /// Connection io error
+    #[display(fmt = "{}", _0)]
+    IoError(io::Error),
+}

actix-connect/src/lib.rs

@@ -0,0 +1,65 @@
+//! Actix connect - tcp connector service
+//!
+//! ## Package feature
+//!
+//! * `ssl` - enables ssl support via `openssl` crate
+//! * `rust-tls` - enables ssl support via `rustls` crate
+
+#[macro_use]
+extern crate log;
+
+mod connect;
+mod connector;
+mod error;
+mod resolver;
+pub mod ssl;
+
+pub use trust_dns_resolver::error::ResolveError;
+
+pub use self::connect::{Connect, Stream};
+pub use self::connector::{Connector, ConnectorFactory};
+pub use self::error::ConnectError;
+pub use self::resolver::{Resolver, ResolverFactory};
+
+use actix_service::{NewService, Service, ServiceExt};
+use tokio_tcp::TcpStream;
+use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
+
+/// Create tcp connector service
+pub fn new_connector(
+    cfg: ResolverConfig,
+    opts: ResolverOpts,
+) -> impl Service<Request = Connect, Response = Stream<TcpStream>, Error = ConnectError> + Clone
+{
+    Resolver::new(cfg, opts).and_then(Connector)
+}
+
+/// Create tcp connector service
+pub fn new_connector_factory(
+    cfg: ResolverConfig,
+    opts: ResolverOpts,
+) -> impl NewService<
+    Request = Connect,
+    Response = Stream<TcpStream>,
+    Error = ConnectError,
+    InitError = (),
+> + Clone {
+    ResolverFactory::new(cfg, opts).and_then(ConnectorFactory)
+}
+
+/// Create connector service with default parameters
+pub fn default_connector(
+) -> impl Service<Request = Connect, Response = Stream<TcpStream>, Error = ConnectError> + Clone
+{
+    Resolver::default().and_then(Connector)
+}
+
+/// Create connector service factory with default parameters
+pub fn default_connector_factory() -> impl NewService<
+    Request = Connect,
+    Response = Stream<TcpStream>,
+    Error = ConnectError,
+    InitError = (),
+> + Clone {
+    ResolverFactory::default().and_then(ConnectorFactory)
+}

actix-connect/src/resolver.rs

@@ -0,0 +1,179 @@
+use std::collections::VecDeque;
+use std::net::SocketAddr;
+
+use actix_service::{NewService, Service};
+use futures::future::{ok, Either, FutureResult};
+use futures::{Async, Future, Poll};
+use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
+use trust_dns_resolver::lookup_ip::LookupIpFuture;
+use trust_dns_resolver::system_conf::read_system_conf;
+use trust_dns_resolver::{AsyncResolver, Background};
+
+use crate::connect::Connect;
+use crate::error::ConnectError;
+
+/// DNS Resolver Service factory
+pub struct ResolverFactory {
+    resolver: AsyncResolver,
+}
+
+impl Default for ResolverFactory {
+    fn default() -> Self {
+        let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
+            (cfg, opts)
+        } else {
+            (ResolverConfig::default(), ResolverOpts::default())
+        };
+
+        ResolverFactory::new(cfg, opts)
+    }
+}
+
+impl ResolverFactory {
+    /// Create new resolver instance with custom configuration and options.
+    pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
+        let (resolver, bg) = AsyncResolver::new(cfg, opts);
+        tokio_current_thread::spawn(bg);
+        ResolverFactory { resolver }
+    }
+}
+
+impl Clone for ResolverFactory {
+    fn clone(&self) -> Self {
+        ResolverFactory {
+            resolver: self.resolver.clone(),
+        }
+    }
+}
+
+impl NewService for ResolverFactory {
+    type Request = Connect;
+    type Response = Connect;
+    type Error = ConnectError;
+    type Service = Resolver;
+    type InitError = ();
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self, _: &()) -> Self::Future {
+        ok(Resolver {
+            resolver: self.resolver.clone(),
+        })
+    }
+}
+
+/// DNS Resolver Service
+pub struct Resolver {
+    resolver: AsyncResolver,
+}
+
+impl Default for Resolver {
+    fn default() -> Self {
+        let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
+            (cfg, opts)
+        } else {
+            (ResolverConfig::default(), ResolverOpts::default())
+        };
+
+        Resolver::new(cfg, opts)
+    }
+}
+
+impl Resolver {
+    /// Create new resolver instance with custom configuration and options.
+    pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
+        let (resolver, bg) = AsyncResolver::new(cfg, opts);
+        tokio_current_thread::spawn(bg);
+        Resolver { resolver }
+    }
+}
+
+impl Clone for Resolver {
+    fn clone(&self) -> Self {
+        Resolver {
+            resolver: self.resolver.clone(),
+        }
+    }
+}
+
+impl Service for Resolver {
+    type Request = Connect;
+    type Response = Connect;
+    type Error = ConnectError;
+    type Future = Either<ResolverFuture, FutureResult<Connect, Self::Error>>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: Connect) -> Self::Future {
+        match req {
+            Connect::Host { host, port } => {
+                if let Ok(ip) = host.parse() {
+                    Either::B(ok(Connect::Addr {
+                        host: host,
+                        addr: either::Either::Left(SocketAddr::new(ip, port)),
+                    }))
+                } else {
+                    trace!("DNS resolver: resolving host {:?}", host);
+                    Either::A(ResolverFuture::new(host, port, &self.resolver))
+                }
+            }
+            other => Either::B(ok(other)),
+        }
+    }
+}
+
+#[doc(hidden)]
+/// Resolver future
+pub struct ResolverFuture {
+    host: Option<String>,
+    port: u16,
+    lookup: Option<Background<LookupIpFuture>>,
+}
+
+impl ResolverFuture {
+    pub fn new(host: String, port: u16, resolver: &AsyncResolver) -> Self {
+        ResolverFuture {
+            lookup: Some(resolver.lookup_ip(host.as_str())),
+            host: Some(host),
+            port,
+        }
+    }
+}
+
+impl Future for ResolverFuture {
+    type Item = Connect;
+    type Error = ConnectError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self.lookup.as_mut().unwrap().poll().map_err(|e| {
+            trace!(
+                "DNS resolver: failed to resolve host {:?} err: {}",
+                self.host.as_ref().unwrap(),
+                e
+            );
+            e
+        })? {
+            Async::NotReady => Ok(Async::NotReady),
+            Async::Ready(ips) => {
+                let host = self.host.take().unwrap();
+                let mut addrs: VecDeque<_> = ips
+                    .iter()
+                    .map(|ip| SocketAddr::new(ip, self.port))
+                    .collect();
+                trace!("DNS resolver: host {:?} resolved to {:?}", host, addrs);
+                if addrs.len() == 1 {
+                    Ok(Async::Ready(Connect::Addr {
+                        addr: either::Either::Left(addrs.pop_front().unwrap()),
+                        host,
+                    }))
+                } else {
+                    Ok(Async::Ready(Connect::Addr {
+                        addr: either::Either::Right(addrs),
+                        host,
+                    }))
+                }
+            }
+        }
+    }
+}

actix-connect/src/ssl/mod.rs

@@ -0,0 +1,6 @@
+//! SSL Services
+
+#[cfg(feature = "ssl")]
+mod openssl;
+#[cfg(feature = "ssl")]
+pub use self::openssl::OpensslConnector;

actix-connect/src/ssl/openssl.rs

@@ -0,0 +1,123 @@
+use std::fmt;
+use std::marker::PhantomData;
+
+use actix_codec::{AsyncRead, AsyncWrite};
+use actix_service::{NewService, Service};
+use futures::{future::ok, future::FutureResult, Async, Future, Poll};
+use openssl::ssl::{HandshakeError, SslConnector};
+use tokio_openssl::{ConnectAsync, SslConnectorExt, SslStream};
+
+use crate::Stream;
+
+/// Openssl connector factory
+pub struct OpensslConnector<T, P, E> {
+    connector: SslConnector,
+    _t: PhantomData<(T, P, E)>,
+}
+
+impl<T, P, E> OpensslConnector<T, P, E> {
+    pub fn new(connector: SslConnector) -> Self {
+        OpensslConnector {
+            connector,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + fmt::Debug, P> OpensslConnector<T, P, ()> {
+    pub fn service(
+        connector: SslConnector,
+    ) -> impl Service<
+        Request = Stream<T, P>,
+        Response = Stream<SslStream<T>, P>,
+        Error = HandshakeError<T>,
+    > {
+        OpensslConnectorService {
+            connector: connector,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, P, E> Clone for OpensslConnector<T, P, E> {
+    fn clone(&self) -> Self {
+        Self {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, P, E> NewService<()> for OpensslConnector<T, P, E>
+where
+    T: AsyncRead + AsyncWrite + fmt::Debug,
+{
+    type Request = Stream<T, P>;
+    type Response = Stream<SslStream<T>, P>;
+    type Error = HandshakeError<T>;
+    type Service = OpensslConnectorService<T, P>;
+    type InitError = E;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self, _: &()) -> Self::Future {
+        ok(OpensslConnectorService {
+            connector: self.connector.clone(),
+            _t: PhantomData,
+        })
+    }
+}
+
+pub struct OpensslConnectorService<T, P> {
+    connector: SslConnector,
+    _t: PhantomData<(T, P)>,
+}
+
+impl<T, P> Service for OpensslConnectorService<T, P>
+where
+    T: AsyncRead + AsyncWrite + fmt::Debug,
+{
+    type Request = Stream<T, P>;
+    type Response = Stream<SslStream<T>, P>;
+    type Error = HandshakeError<T>;
+    type Future = ConnectAsyncExt<T, P>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, stream: Stream<T, P>) -> Self::Future {
+        trace!("SSL Handshake start for: {:?}", stream.host());
+        let (io, stream) = stream.replace(());
+        ConnectAsyncExt {
+            fut: SslConnectorExt::connect_async(&self.connector, stream.host(), io),
+            stream: Some(stream),
+        }
+    }
+}
+
+pub struct ConnectAsyncExt<T, P> {
+    fut: ConnectAsync<T>,
+    stream: Option<Stream<(), P>>,
+}
+
+impl<T, P> Future for ConnectAsyncExt<T, P>
+where
+    T: AsyncRead + AsyncWrite + fmt::Debug,
+{
+    type Item = Stream<SslStream<T>, P>;
+    type Error = HandshakeError<T>;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self.fut.poll().map_err(|e| {
+            trace!("SSL Handshake error: {:?}", e);
+            e
+        })? {
+            Async::Ready(stream) => {
+                let s = self.stream.take().unwrap();
+                trace!("SSL Handshake success: {:?}", s.host());
+                Ok(Async::Ready(s.replace(stream).1))
+            }
+            Async::NotReady => Ok(Async::NotReady),
+        }
+    }
+}