Migrate actix-server::ssl::nativetls to std futures (#61)

* Refactor 'nativetls' module

* Migrate 'actix-server-config' to std futures

- remove "uds" feature
- disable features by default

* Switch NativeTlsAcceptor to use 'tokio-tls' crate

* Bikeshed features names and remove unnecessary dependencies for 'actix-server-config' crate

actix-server-config/Cargo.toml

@@ -14,26 +14,14 @@ name = "actix_server_config"
 path = "src/lib.rs"
 
 [package.metadata.docs.rs]
-features = ["ssl", "rust-tls", "uds"]
+features = ["openssl", "rustls"]
 
 [features]
-default = ["ssl", "rustls" ]
-
-# openssl
-ssl = ["tokio-openssl"]
-
-# rustls
-rust-tls = ["rustls", "tokio-rustls"]
-
-# unix domain sockets
-# TODO: FIXME
-uds = [] # ["tokio-uds"]
+default = []
+openssl = ["tokio-openssl"]
+rustls = ["tokio-rustls"]
 
 [dependencies]
-futures = { package = "futures-preview", version = "0.3.0-alpha.18" }
-tokio = "0.2.0-alpha.4"
-
-tokio-openssl = { version="0.4.0-alpha.4", optional = true }
-rustls = { version = "0.16.0", optional = true }
-tokio-rustls = { version = "0.12.0-alpha.2", optional = true }
-#tokio-uds = { version="0.3.0-alpha.1", optional = true }
+tokio = "0.2.0-alpha.6"
+tokio-openssl = { version = "0.4.0-alpha.6", optional = true }
+tokio-rustls = { version = "0.12.0-alpha.8", optional = true }

actix-server-config/src/lib.rs

@@ -1,10 +1,12 @@
+//! Actix server config utils.
+
 use std::cell::Cell;
 use std::net::SocketAddr;
 use std::rc::Rc;
-use std::{fmt, io, net, time};
+use std::{fmt, io, net, ops, time};
 
 use tokio::io::{AsyncRead, AsyncWrite};
-use tokio::net::tcp::TcpStream;
+use tokio::net::TcpStream;
 
 #[derive(Debug, Clone)]
 pub struct ServerConfig {
@@ -13,6 +15,7 @@ pub struct ServerConfig {
 }
 
 impl ServerConfig {
+    #[inline]
     pub fn new(addr: SocketAddr) -> Self {
         ServerConfig {
             addr,
@@ -21,16 +24,19 @@ impl ServerConfig {
     }
 
     /// Returns the address of the local half of this TCP server socket
+    #[inline]
     pub fn local_addr(&self) -> SocketAddr {
         self.addr
     }
 
     /// Returns true if connection is secure (tls enabled)
+    #[inline]
     pub fn secure(&self) -> bool {
         self.secure.as_ref().get()
     }
 
     /// Set secure flag
+    #[inline]
     pub fn set_secure(&self) {
         self.secure.as_ref().set(true)
     }
@@ -114,7 +120,7 @@ impl<T, P> Io<T, P> {
     }
 }
 
-impl<T, P> std::ops::Deref for Io<T, P> {
+impl<T, P> ops::Deref for Io<T, P> {
     type Target = T;
 
     fn deref(&self) -> &T {
@@ -122,14 +128,14 @@ impl<T, P> std::ops::Deref for Io<T, P> {
     }
 }
 
-impl<T, P> std::ops::DerefMut for Io<T, P> {
+impl<T, P> ops::DerefMut for Io<T, P> {
     fn deref_mut(&mut self) -> &mut T {
         &mut self.io
     }
 }
 
 impl<T: fmt::Debug, P> fmt::Debug for Io<T, P> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "Io {{{:?}}}", self.io)
     }
 }
@@ -171,7 +177,7 @@ impl IoStream for TcpStream {
     }
 }
 
-#[cfg(any(feature = "ssl"))]
+#[cfg(feature = "openssl")]
 impl<T: IoStream + Unpin> IoStream for tokio_openssl::SslStream<T> {
     #[inline]
     fn peer_addr(&self) -> Option<net::SocketAddr> {
@@ -194,8 +200,8 @@ impl<T: IoStream + Unpin> IoStream for tokio_openssl::SslStream<T> {
     }
 }
 
-#[cfg(any(feature = "rust-tls"))]
-impl<T: IoStream> IoStream for tokio_rustls::server::TlsStream<T> {
+#[cfg(feature = "rustls")]
+impl<T: IoStream + Unpin> IoStream for tokio_rustls::server::TlsStream<T> {
     #[inline]
     fn peer_addr(&self) -> Option<net::SocketAddr> {
         self.get_ref().0.peer_addr()
@@ -217,8 +223,8 @@ impl<T: IoStream> IoStream for tokio_rustls::server::TlsStream<T> {
     }
 }
 
-#[cfg(all(unix, feature = "uds"))]
-impl IoStream for t::UnixStream {
+#[cfg(unix)]
+impl IoStream for tokio::net::UnixStream {
     #[inline]
     fn peer_addr(&self) -> Option<net::SocketAddr> {
         None

actix-server/Cargo.toml

@@ -14,7 +14,7 @@ edition = "2018"
 workspace = ".."
 
 [package.metadata.docs.rs]
-features = ["ssl", "tls", "rust-tls", "uds"]
+features = ["nativetls", "openssl", "rustls", "uds"]
 
 [lib]
 name = "actix_server"
@@ -22,15 +22,9 @@ path = "src/lib.rs"
 
 [features]
 default = []
-
-# tls
-tls = ["native-tls"]
-
-# openssl
-ssl = ["openssl", "tokio-openssl", "actix-server-config/ssl"]
-
-# rustls
-rust-tls = ["rustls", "tokio-rustls", "webpki", "webpki-roots", "actix-server-config/rust-tls"]
+nativetls = ["native-tls", "tokio-tls"]
+openssl = ["open-ssl", "tokio-openssl", "actix-server-config/openssl"]
+rustls = ["rust-tls", "tokio-rustls", "webpki", "webpki-roots", "actix-server-config/rustls"]
 
 # uds
 # uds = ["mio-uds", "tokio-uds", "actix-server-config/uds"]
@@ -54,18 +48,19 @@ tokio-net = { version = "0.2.0-alpha.6", features = ["signal"] }
 tokio-timer = "0.3.0-alpha.6"
 
 # unix domain sockets
-mio-uds = { version="0.6.7", optional = true }
+mio-uds = { version = "0.6.7", optional = true }
 #tokio-uds = { version="0.2.5", optional = true }
 
-# native-tls
-native-tls = { version="0.2", optional = true }
+# nativetls
+native-tls = { version = "0.2", optional = true }
+tokio-tls = { version = "0.3.0-alpha.6", optional = true }
 
 # openssl
-openssl = { version="0.10", optional = true }
-tokio-openssl = { version="0.4.0-alpha.6", optional = true }
+open-ssl = { version = "0.10", package = "openssl", optional = true }
+tokio-openssl = { version = "0.4.0-alpha.6", optional = true }
 
 # rustls
-rustls = { version = "0.16.0", optional = true }
+rust-tls = { version = "0.16.0", package = "rustls", optional = true }
 tokio-rustls = { version = "0.12.0-alpha.2", optional = true }
 webpki = { version = "0.21", optional = true }
 webpki-roots = { version = "0.17", optional = true }

actix-server/src/ssl/mod.rs

@@ -3,19 +3,19 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 
 use crate::counter::Counter;
 
-#[cfg(feature = "ssl")]
+#[cfg(feature = "openssl")]
 mod openssl;
-#[cfg(feature = "ssl")]
+#[cfg(feature = "openssl")]
 pub use self::openssl::OpensslAcceptor;
 
-#[cfg(feature = "tls")]
+#[cfg(feature = "nativetls")]
 mod nativetls;
-#[cfg(feature = "tls")]
-pub use self::nativetls::{NativeTlsAcceptor, TlsStream};
+#[cfg(feature = "nativetls")]
+pub use self::nativetls::NativeTlsAcceptor;
 
-#[cfg(feature = "rust-tls")]
+#[cfg(feature = "rustls")]
 mod rustls;
-#[cfg(feature = "rust-tls")]
+#[cfg(feature = "rustls")]
 pub use self::rustls::RustlsAcceptor;
 
 /// Sets the maximum per-worker concurrent ssl connection establish process.

actix-server/src/ssl/nativetls.rs

@@ -1,16 +1,19 @@
-use std::io;
+use std::convert::Infallible;
 use std::marker::PhantomData;
+use std::task::{Context, Poll};
 
-use actix_service::{NewService, Service};
-use futures::{future::ok, future::FutureResult, Async, Future, Poll};
-use native_tls::{self, Error, HandshakeError, TlsAcceptor};
-use tokio_io::{AsyncRead, AsyncWrite};
+use actix_service::{Service, ServiceFactory};
+use futures::{
+    future::{self, LocalBoxFuture},
+    FutureExt as _, TryFutureExt as _,
+};
+use native_tls::Error;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio_tls::{TlsAcceptor, TlsStream};
 
-use crate::counter::{Counter, CounterGuard};
+use crate::counter::Counter;
 use crate::ssl::MAX_CONN_COUNTER;
-use crate::{Io, Protocol, ServerConfig};
-use std::pin::Pin;
-use std::task::Context;
+use crate::{Io, ServerConfig};
 
 /// Support `SSL` connections via native-tls package
 ///
@@ -20,8 +23,12 @@ pub struct NativeTlsAcceptor<T, P = ()> {
     io: PhantomData<(T, P)>,
 }
 
-impl<T: AsyncRead + AsyncWrite, P> NativeTlsAcceptor<T, P> {
+impl<T, P> NativeTlsAcceptor<T, P>
+where
+    T: AsyncRead + AsyncWrite + Unpin,
+{
     /// Create `NativeTlsAcceptor` instance
+    #[inline]
     pub fn new(acceptor: TlsAcceptor) -> Self {
         NativeTlsAcceptor {
             acceptor,
@@ -30,7 +37,8 @@ impl<T: AsyncRead + AsyncWrite, P> NativeTlsAcceptor<T, P> {
     }
 }
 
-impl<T: AsyncRead + AsyncWrite, P> Clone for NativeTlsAcceptor<T, P> {
+impl<T, P> Clone for NativeTlsAcceptor<T, P> {
+    #[inline]
     fn clone(&self) -> Self {
         Self {
             acceptor: self.acceptor.clone(),
@@ -39,21 +47,25 @@ impl<T: AsyncRead + AsyncWrite, P> Clone for NativeTlsAcceptor<T, P> {
     }
 }
 
-impl<T: AsyncRead + AsyncWrite, P> NewService for NativeTlsAcceptor<T, P> {
+impl<T, P> ServiceFactory for NativeTlsAcceptor<T, P>
+where
+    T: AsyncRead + AsyncWrite + Unpin + 'static,
+    P: 'static,
+{
     type Request = Io<T, P>;
     type Response = Io<TlsStream<T>, P>;
     type Error = Error;
 
     type Config = ServerConfig;
     type Service = NativeTlsAcceptorService<T, P>;
-    type InitError = ();
-    type Future = FutureResult<Self::Service, Self::InitError>;
+    type InitError = Infallible;
+    type Future = future::Ready<Result<Self::Service, Self::InitError>>;
 
     fn new_service(&self, cfg: &ServerConfig) -> Self::Future {
         cfg.set_secure();
 
         MAX_CONN_COUNTER.with(|conns| {
-            ok(NativeTlsAcceptorService {
+            future::ok(NativeTlsAcceptorService {
                 acceptor: self.acceptor.clone(),
                 conns: conns.clone(),
                 io: PhantomData,
@@ -68,130 +80,46 @@ pub struct NativeTlsAcceptorService<T, P> {
     conns: Counter,
 }
 
-impl<T: AsyncRead + AsyncWrite, P> Service for NativeTlsAcceptorService<T, P> {
+impl<T, P> Clone for NativeTlsAcceptorService<T, P> {
+    fn clone(&self) -> Self {
+        Self {
+            acceptor: self.acceptor.clone(),
+            io: PhantomData,
+            conns: self.conns.clone(),
+        }
+    }
+}
+
+impl<T, P> Service for NativeTlsAcceptorService<T, P>
+where
+    T: AsyncRead + AsyncWrite + Unpin + 'static,
+    P: 'static,
+{
     type Request = Io<T, P>;
     type Response = Io<TlsStream<T>, P>;
     type Error = Error;
-    type Future = Accept<T, P>;
+    type Future = LocalBoxFuture<'static, Result<Io<TlsStream<T>, P>, Error>>;
 
-    fn poll_ready(
-        self: Pin<&mut Self>,
-        ctx: &mut Context<'_>,
-    ) -> Poll<Result<(), Self::Error>> {
-        if self.conns.available(ctx) {
-            Ok(Async::Ready(()))
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if self.conns.available(cx) {
+            Poll::Ready(Ok(()))
         } else {
-            Ok(Async::NotReady)
+            Poll::Pending
         }
     }
 
-    /*
-    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
-        if self.conns.available() {
-            Ok(Async::Ready(()))
-        } else {
-            Ok(Async::NotReady)
-        }
-    }
-    */
-
     fn call(&mut self, req: Self::Request) -> Self::Future {
-        let (io, params, _) = req.into_parts();
-        Accept {
-            _guard: self.conns.get(),
-            inner: Some(self.acceptor.accept(io)),
-            params: Some(params),
-        }
-    }
-}
-
-/// A wrapper around an underlying raw stream which implements the TLS or SSL
-/// protocol.
-///
-/// A `TlsStream<S>` represents a handshake that has been completed successfully
-/// and both the server and the client are ready for receiving and sending
-/// data. Bytes read from a `TlsStream` are decrypted from `S` and bytes written
-/// to a `TlsStream` are encrypted when passing through to `S`.
-#[derive(Debug)]
-pub struct TlsStream<S> {
-    inner: native_tls::TlsStream<S>,
-}
-
-/// Future returned from `NativeTlsAcceptor::accept` which will resolve
-/// once the accept handshake has finished.
-pub struct Accept<S, P> {
-    inner: Option<Result<native_tls::TlsStream<S>, HandshakeError<S>>>,
-    params: Option<P>,
-    _guard: CounterGuard,
-}
-
-impl<T: AsyncRead + AsyncWrite, P> Future for Accept<T, P> {
-    type Item = Io<TlsStream<T>, P>;
-    type Error = Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        match self.inner.take().expect("cannot poll MidHandshake twice") {
-            Ok(stream) => Ok(Async::Ready(Io::from_parts(
-                TlsStream { inner: stream },
-                self.params.take().unwrap(),
-                Protocol::Unknown,
-            ))),
-            Err(HandshakeError::Failure(e)) => Err(e),
-            Err(HandshakeError::WouldBlock(s)) => match s.handshake() {
-                Ok(stream) => Ok(Async::Ready(Io::from_parts(
-                    TlsStream { inner: stream },
-                    self.params.take().unwrap(),
-                    Protocol::Unknown,
-                ))),
-                Err(HandshakeError::Failure(e)) => Err(e),
-                Err(HandshakeError::WouldBlock(s)) => {
-                    self.inner = Some(Err(HandshakeError::WouldBlock(s)));
-                    Ok(Async::NotReady)
-                }
-            },
-        }
-    }
-}
-
-impl<S> TlsStream<S> {
-    /// Get access to the internal `native_tls::TlsStream` stream which also
-    /// transitively allows access to `S`.
-    pub fn get_ref(&self) -> &native_tls::TlsStream<S> {
-        &self.inner
-    }
-
-    /// Get mutable access to the internal `native_tls::TlsStream` stream which
-    /// also transitively allows mutable access to `S`.
-    pub fn get_mut(&mut self) -> &mut native_tls::TlsStream<S> {
-        &mut self.inner
-    }
-}
-
-impl<S: io::Read + io::Write> io::Read for TlsStream<S> {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        self.inner.read(buf)
-    }
-}
-
-impl<S: io::Read + io::Write> io::Write for TlsStream<S> {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.inner.write(buf)
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        self.inner.flush()
-    }
-}
-
-impl<S: AsyncRead + AsyncWrite> AsyncRead for TlsStream<S> {}
-
-impl<S: AsyncRead + AsyncWrite> AsyncWrite for TlsStream<S> {
-    fn shutdown(&mut self) -> Poll<(), io::Error> {
-        match self.inner.shutdown() {
-            Ok(_) => (),
-            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => (),
-            Err(e) => return Err(e),
-        }
-        self.inner.get_mut().shutdown()
+        let guard = self.conns.get();
+        let this = self.clone();
+        let (io, params, proto) = req.into_parts();
+        async move { this.acceptor.accept(io).await }
+            .map_ok(move |stream| Io::from_parts(stream, params, proto))
+            .map_ok(move |io| {
+                // Required to preserve `CounterGuard` until `Self::Future`
+                // is completely resolved.
+                let _ = guard;
+                io
+            })
+            .boxed_local()
     }
 }