move server to separate crate

actix-server/src/ssl/mod.rs

@@ -0,0 +1,35 @@
+//! SSL Services
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use crate::counter::Counter;
+
+#[cfg(feature = "ssl")]
+mod openssl;
+#[cfg(feature = "ssl")]
+pub use self::openssl::OpensslAcceptor;
+
+#[cfg(feature = "tls")]
+mod nativetls;
+#[cfg(feature = "tls")]
+pub use self::nativetls::{NativeTlsAcceptor, TlsStream};
+
+#[cfg(feature = "rust-tls")]
+mod rustls;
+#[cfg(feature = "rust-tls")]
+pub use self::rustls::RustlsAcceptor;
+
+/// Sets the maximum per-worker concurrent ssl connection establish process.
+///
+/// All listeners will stop accepting connections when this limit is
+/// reached. It can be used to limit the global SSL CPU usage.
+///
+/// By default max connections is set to a 256.
+pub fn max_concurrent_ssl_connect(num: usize) {
+    MAX_CONN.store(num, Ordering::Relaxed);
+}
+
+pub(crate) static MAX_CONN: AtomicUsize = AtomicUsize::new(256);
+
+thread_local! {
+    static MAX_CONN_COUNTER: Counter = Counter::new(MAX_CONN.load(Ordering::Relaxed));
+}

actix-server/src/ssl/nativetls.rs

@@ -0,0 +1,164 @@
+use std::io;
+use std::marker::PhantomData;
+
+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 super::MAX_CONN_COUNTER;
+use crate::counter::{Counter, CounterGuard};
+
+/// Support `SSL` connections via native-tls package
+///
+/// `tls` feature enables `NativeTlsAcceptor` type
+pub struct NativeTlsAcceptor<T> {
+    acceptor: TlsAcceptor,
+    io: PhantomData<T>,
+}
+
+impl<T: AsyncRead + AsyncWrite> NativeTlsAcceptor<T> {
+    /// Create `NativeTlsAcceptor` instance
+    pub fn new(acceptor: TlsAcceptor) -> Self {
+        NativeTlsAcceptor {
+            acceptor,
+            io: PhantomData,
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> Clone for NativeTlsAcceptor<T> {
+    fn clone(&self) -> Self {
+        Self {
+            acceptor: self.acceptor.clone(),
+            io: PhantomData,
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> NewService<T> for NativeTlsAcceptor<T> {
+    type Response = TlsStream<T>;
+    type Error = Error;
+    type Service = NativeTlsAcceptorService<T>;
+    type InitError = ();
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        MAX_CONN_COUNTER.with(|conns| {
+            ok(NativeTlsAcceptorService {
+                acceptor: self.acceptor.clone(),
+                conns: conns.clone(),
+                io: PhantomData,
+            })
+        })
+    }
+}
+
+pub struct NativeTlsAcceptorService<T> {
+    acceptor: TlsAcceptor,
+    io: PhantomData<T>,
+    conns: Counter,
+}
+
+impl<T: AsyncRead + AsyncWrite> Service<T> for NativeTlsAcceptorService<T> {
+    type Response = TlsStream<T>;
+    type Error = Error;
+    type Future = Accept<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        if self.conns.available() {
+            Ok(Async::Ready(()))
+        } else {
+            Ok(Async::NotReady)
+        }
+    }
+
+    fn call(&mut self, req: T) -> Self::Future {
+        Accept {
+            _guard: self.conns.get(),
+            inner: Some(self.acceptor.accept(req)),
+        }
+    }
+}
+
+/// 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> {
+    inner: Option<Result<native_tls::TlsStream<S>, HandshakeError<S>>>,
+    _guard: CounterGuard,
+}
+
+impl<Io: AsyncRead + AsyncWrite> Future for Accept<Io> {
+    type Item = TlsStream<Io>;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self.inner.take().expect("cannot poll MidHandshake twice") {
+            Ok(stream) => Ok(TlsStream { inner: stream }.into()),
+            Err(HandshakeError::Failure(e)) => Err(e),
+            Err(HandshakeError::WouldBlock(s)) => match s.handshake() {
+                Ok(stream) => Ok(TlsStream { inner: stream }.into()),
+                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()
+    }
+}

actix-server/src/ssl/openssl.rs

@@ -0,0 +1,99 @@
+use std::marker::PhantomData;
+
+use actix_service::{NewService, Service};
+use futures::{future::ok, future::FutureResult, Async, Future, Poll};
+use openssl::ssl::{HandshakeError, SslAcceptor};
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_openssl::{AcceptAsync, SslAcceptorExt, SslStream};
+
+use super::MAX_CONN_COUNTER;
+use crate::counter::{Counter, CounterGuard};
+
+/// Support `SSL` connections via openssl package
+///
+/// `ssl` feature enables `OpensslAcceptor` type
+pub struct OpensslAcceptor<T> {
+    acceptor: SslAcceptor,
+    io: PhantomData<T>,
+}
+
+impl<T> OpensslAcceptor<T> {
+    /// Create default `OpensslAcceptor`
+    pub fn new(acceptor: SslAcceptor) -> Self {
+        OpensslAcceptor {
+            acceptor,
+            io: PhantomData,
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> Clone for OpensslAcceptor<T> {
+    fn clone(&self) -> Self {
+        Self {
+            acceptor: self.acceptor.clone(),
+            io: PhantomData,
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> NewService<T> for OpensslAcceptor<T> {
+    type Response = SslStream<T>;
+    type Error = HandshakeError<T>;
+    type Service = OpensslAcceptorService<T>;
+    type InitError = ();
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        MAX_CONN_COUNTER.with(|conns| {
+            ok(OpensslAcceptorService {
+                acceptor: self.acceptor.clone(),
+                conns: conns.clone(),
+                io: PhantomData,
+            })
+        })
+    }
+}
+
+pub struct OpensslAcceptorService<T> {
+    acceptor: SslAcceptor,
+    io: PhantomData<T>,
+    conns: Counter,
+}
+
+impl<T: AsyncRead + AsyncWrite> Service<T> for OpensslAcceptorService<T> {
+    type Response = SslStream<T>;
+    type Error = HandshakeError<T>;
+    type Future = OpensslAcceptorServiceFut<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        if self.conns.available() {
+            Ok(Async::Ready(()))
+        } else {
+            Ok(Async::NotReady)
+        }
+    }
+
+    fn call(&mut self, req: T) -> Self::Future {
+        OpensslAcceptorServiceFut {
+            _guard: self.conns.get(),
+            fut: SslAcceptorExt::accept_async(&self.acceptor, req),
+        }
+    }
+}
+
+pub struct OpensslAcceptorServiceFut<T>
+where
+    T: AsyncRead + AsyncWrite,
+{
+    fut: AcceptAsync<T>,
+    _guard: CounterGuard,
+}
+
+impl<T: AsyncRead + AsyncWrite> Future for OpensslAcceptorServiceFut<T> {
+    type Item = SslStream<T>;
+    type Error = HandshakeError<T>;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.fut.poll()
+    }
+}

actix-server/src/ssl/rustls.rs

@@ -0,0 +1,101 @@
+use std::io;
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+use actix_service::{NewService, Service};
+use futures::{future::ok, future::FutureResult, Async, Future, Poll};
+use rustls::{ServerConfig, ServerSession};
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_rustls::{Accept, TlsAcceptor, TlsStream};
+
+use super::MAX_CONN_COUNTER;
+use crate::counter::{Counter, CounterGuard};
+
+/// Support `SSL` connections via rustls package
+///
+/// `rust-tls` feature enables `RustlsAcceptor` type
+pub struct RustlsAcceptor<T> {
+    config: Arc<ServerConfig>,
+    io: PhantomData<T>,
+}
+
+impl<T: AsyncRead + AsyncWrite> RustlsAcceptor<T> {
+    /// Create `RustlsAcceptor` new service
+    pub fn new(config: ServerConfig) -> Self {
+        RustlsAcceptor {
+            config: Arc::new(config),
+            io: PhantomData,
+        }
+    }
+}
+
+impl<T> Clone for RustlsAcceptor<T> {
+    fn clone(&self) -> Self {
+        Self {
+            config: self.config.clone(),
+            io: PhantomData,
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> NewService<T> for RustlsAcceptor<T> {
+    type Response = TlsStream<T, ServerSession>;
+    type Error = io::Error;
+    type Service = RustlsAcceptorService<T>;
+    type InitError = ();
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        MAX_CONN_COUNTER.with(|conns| {
+            ok(RustlsAcceptorService {
+                acceptor: self.config.clone().into(),
+                conns: conns.clone(),
+                io: PhantomData,
+            })
+        })
+    }
+}
+
+pub struct RustlsAcceptorService<T> {
+    acceptor: TlsAcceptor,
+    io: PhantomData<T>,
+    conns: Counter,
+}
+
+impl<T: AsyncRead + AsyncWrite> Service<T> for RustlsAcceptorService<T> {
+    type Response = TlsStream<T, ServerSession>;
+    type Error = io::Error;
+    type Future = RustlsAcceptorServiceFut<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        if self.conns.available() {
+            Ok(Async::Ready(()))
+        } else {
+            Ok(Async::NotReady)
+        }
+    }
+
+    fn call(&mut self, req: T) -> Self::Future {
+        RustlsAcceptorServiceFut {
+            _guard: self.conns.get(),
+            fut: self.acceptor.accept(req),
+        }
+    }
+}
+
+pub struct RustlsAcceptorServiceFut<T>
+where
+    T: AsyncRead + AsyncWrite,
+{
+    fut: Accept<T>,
+    _guard: CounterGuard,
+}
+
+impl<T: AsyncRead + AsyncWrite> Future for RustlsAcceptorServiceFut<T> {
+    type Item = TlsStream<T, ServerSession>;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.fut.poll()
+    }
+}