no readme

.travis.yml

@@ -33,7 +33,12 @@ script:
     if [[ "$TRAVIS_RUST_VERSION" != "nightly" ]]; then
     cargo clean
     cargo test --features="ssl,tls,rust-tls" -- --nocapture
-    cd actix-service && cargo test
+    cd actix-codec && cargo test && cd ..
+    cd actix-service && cargo test && cd ..
+    cd actix-server && cargo test --features="ssl,tls,rust-tls" -- --nocapture && cd ..
+    cd actix-rt && cargo test && cd ..
+    cd actix-connector && cargo test && cd ..
+    cd actix-utils && cargo test && cd ..
     fi
   - |
     if [[ "$TRAVIS_RUST_VERSION" == "nightly" ]]; then
@@ -41,16 +46,10 @@ script:
     cargo tarpaulin --features="ssl,tls,rust-tls" --out Xml
     bash <(curl -s https://codecov.io/bash)
     echo "Uploaded code coverage"
-    cd actix-service && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash)
+    cd actix-service && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    fi
+    cd actix-rt && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-
+    cd actix-connector && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-# Upload docs
+    cd actix-codec && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-after_success:
+    cd actix-server && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-  - |
+    cd actix-utils && cargo tarpaulin --out Xml && bash <(curl -s https://codecov.io/bash) && cd ..
-    if [[ "$TRAVIS_OS_NAME" == "linux" && "$TRAVIS_PULL_REQUEST" = "false" && "$TRAVIS_BRANCH" == "master" && "$TRAVIS_RUST_VERSION" == "beta" ]]; then
-      cargo doc --features "ssl,tls,rust-tls" --no-deps &&
-      echo "<meta http-equiv=refresh content=0;url=os_balloon/index.html>" > target/doc/index.html &&
-      git clone https://github.com/davisp/ghp-import.git &&
-      ./ghp-import/ghp_import.py -n -p -f -m "Documentation upload" -r https://"$GH_TOKEN"@github.com/"$TRAVIS_REPO_SLUG.git" target/doc &&
-      echo "Uploaded documentation"
     fi

Cargo.toml

@@ -15,76 +15,20 @@ edition = "2018"
 
 [workspace]
 members = [
-  "./",
+  "actix-codec",
+  "actix-connector",
   "actix-service",
+  "actix-server",
+  "actix-rt",
+  "actix-utils",
 ]
 
-[package.metadata.docs.rs]
-features = ["ssl", "tls", "rust-tls"]
-
-[badges]
-travis-ci = { repository = "actix/actix-net", branch = "master" }
-# appveyor = { repository = "fafhrd91/actix-web-hdy9d" }
-codecov = { repository = "actix/actix-net", branch = "master", service = "github" }
-
-[lib]
-name = "actix_net"
-path = "src/lib.rs"
-
-[features]
-default = []
-
-# tls
-tls = ["native-tls"]
-
-# openssl
-ssl = ["openssl", "tokio-openssl"]
-
-# rustls
-rust-tls = ["rustls", "tokio-rustls", "webpki", "webpki-roots"]
-
-cell = []
-
-[dependencies]
-actix = "0.7.6"
-actix-service = "0.1.1"
-
-log = "0.4"
-num_cpus = "1.0"
-
-# io
-mio = "^0.6.13"
-net2 = "0.2"
-bytes = "0.4"
-futures = "0.1"
-slab = "0.4"
-tokio = "0.1"
-tokio-codec = "0.1"
-tokio-io = "0.1"
-tokio-tcp = "0.1"
-tokio-timer = "0.2"
-tokio-reactor = "0.1"
-tokio-current-thread = "0.1"
-trust-dns-proto = "^0.5.0"
-trust-dns-resolver = "^0.10.0"
-
-# native-tls
-native-tls = { version="0.2", optional = true }
-
-# openssl
-openssl = { version="0.10", optional = true }
-tokio-openssl = { version="0.2", optional = true }
-
-#rustls
-rustls = { version = "^0.14", optional = true }
-tokio-rustls = { version = "^0.8", optional = true }
-webpki = { version = "0.18", optional = true }
-webpki-roots = { version = "0.15", optional = true }
-
 [dev-dependencies]
+actix-service = "0.1.1"
+actix-codec = "0.1.0"
+actix-rt = { path="actix-rt" }
+actix-server = { path="actix-server", features=["ssl"] }
 env_logger = "0.5"
-
+futures = "0.1.24"
-[profile.release]
+openssl = { version="0.10" }
-lto = true
+tokio-openssl = { version="0.3" }
-opt-level = 3
-codegen-units = 1

README.md

@@ -13,7 +13,7 @@ Actix net - framework for composable network services (experimental)
 
 ```rust
 fn main() {
-    let sys = actix::System::new("test");
+    let sys = actix_rt::System::new("test");
 
     // load ssl keys
     let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
@@ -26,7 +26,7 @@ fn main() {
     // bind socket address and start workers. By default server uses number of
     // available logical cpu as threads count. actix net start separate
     // instances of service pipeline in each worker.
-    Server::default()
+    actix_server::build()
         .bind(
             // configure service pipeline
             "basic", "0.0.0.0:8443",

actix-codec/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2018-12-09
+
+* Move codec to separate crate

actix-codec/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "actix-codec"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Utilities for encoding and decoding frames"
+keywords = ["network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-codec/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = "../"
+
+[lib]
+name = "actix_codec"
+path = "src/lib.rs"
+
+[dependencies]
+bytes = "0.4"
+futures = "0.1.24"
+tokio-io = "0.1"
+tokio-codec = "0.1"
+log = "0.4"

actix-codec/src/lib.rs

@@ -10,16 +10,15 @@
 //! [`Stream`]: #
 //! [transports]: #
 
-#![deny(missing_docs, missing_debug_implementations, warnings)]
-
 mod bcodec;
 mod framed;
-// mod framed2;
 mod framed_read;
 mod framed_write;
 
 pub use self::bcodec::BytesCodec;
 pub use self::framed::{Framed, FramedParts};
-// pub use self::framed2::{Framed2, FramedParts2};
 pub use self::framed_read::FramedRead;
 pub use self::framed_write::FramedWrite;
+
+pub use tokio_codec::{Decoder, Encoder};
+pub use tokio_io::{AsyncRead, AsyncWrite};

actix-connector/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2018-12-09
+
+* Move server to separate crate

actix-connector/Cargo.toml

@@ -0,0 +1,40 @@
+[package]
+name = "actix-connector"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Actix Connector - tcp connector service"
+keywords = ["network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-net/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = "../"
+
+[package.metadata.docs.rs]
+features = ["ssl"]
+
+[lib]
+name = "actix_connector"
+path = "src/lib.rs"
+
+[features]
+default = []
+
+# openssl
+ssl = ["openssl", "tokio-openssl"]
+
+[dependencies]
+actix-service = "0.1.1"
+actix-codec = "0.1.0"
+actix-rt = "0.1.0"
+futures = "0.1"
+tokio-tcp = "0.1"
+trust-dns-proto = "^0.5.0"
+trust-dns-resolver = "^0.10.0"
+
+# openssl
+openssl = { version="0.10", optional = true }
+tokio-openssl = { version="0.3", optional = true }

actix-connector/src/lib.rs

@@ -0,0 +1,16 @@
+//! Actix Connector - tcp connector service
+//!
+//! ## Package feature
+//!
+//! * `tls` - enables ssl support via `native-tls` crate
+//! * `ssl` - enables ssl support via `openssl` crate
+//! * `rust-tls` - enables ssl support via `rustls` crate
+
+mod connector;
+mod resolver;
+pub mod ssl;
+
+pub use self::connector::{
+    Connect, Connector, ConnectorError, DefaultConnector, RequestPort, TcpConnector,
+};
+pub use self::resolver::{RequestHost, Resolver};

actix-connector/src/resolver.rs

@@ -2,10 +2,8 @@ use std::collections::VecDeque;
 use std::marker::PhantomData;
 use std::net::IpAddr;
 
-use futures::{Async, Future, Poll};
-
 use actix_service::Service;
-use tokio_current_thread::spawn;
+use futures::{Async, Future, Poll};
 use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
 pub use trust_dns_resolver::error::ResolveError;
 use trust_dns_resolver::lookup_ip::LookupIpFuture;
@@ -44,7 +42,7 @@ impl<T: RequestHost> Resolver<T> {
     /// Create new resolver instance with custom configuration and options.
     pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
         let (resolver, bg) = AsyncResolver::new(cfg, opts);
-        spawn(bg);
+        actix_rt::Arbiter::spawn(bg);
         Resolver {
             resolver,
             req: PhantomData,
@@ -79,7 +77,13 @@ impl<T: RequestHost> Service<T> for Resolver<T> {
     }
 
     fn call(&mut self, req: T) -> Self::Future {
-        ResolverFuture::new(req, &self.resolver)
+        if let Ok(ip) = req.host().parse() {
+            let mut addrs = VecDeque::new();
+            addrs.push_back(ip);
+            ResolverFuture::new(req, &self.resolver, Some(addrs))
+        } else {
+            ResolverFuture::new(req, &self.resolver, None)
+        }
     }
 }
 
@@ -92,13 +96,13 @@ pub struct ResolverFuture<T> {
 }
 
 impl<T: RequestHost> ResolverFuture<T> {
-    pub fn new(addr: T, resolver: &AsyncResolver) -> Self {
+    pub fn new(addr: T, resolver: &AsyncResolver, addrs: Option<VecDeque<IpAddr>>) -> Self {
         // we need to do dns resolution
         let lookup = Some(resolver.lookup_ip(addr.host()));
         ResolverFuture {
             lookup,
+            addrs,
             req: Some(addr),
-            addrs: None,
         }
     }
 }

actix-connector/src/ssl/mod.rs

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

actix-connector/src/ssl/openssl.rs

@@ -1,104 +1,13 @@
 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::{Error, SslAcceptor, SslConnector};
+use openssl::ssl::{HandshakeError, SslConnector};
-use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_openssl::{ConnectAsync, SslConnectorExt, SslStream};
-use tokio_openssl::{AcceptAsync, ConnectAsync, SslAcceptorExt, SslConnectorExt, SslStream};
 
-use super::MAX_CONN_COUNTER;
-use crate::counter::{Counter, CounterGuard};
 use crate::resolver::RequestHost;
 
-/// 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 = Error;
-    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 = Error;
-    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 = Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        self.fut.poll()
-    }
-}
-
 /// Openssl connector factory
 pub struct OpensslConnector<R, T, E> {
     connector: SslConnector,
@@ -117,7 +26,7 @@ impl<R, T, E> OpensslConnector<R, T, E> {
 impl<R: RequestHost, T: AsyncRead + AsyncWrite> OpensslConnector<R, T, ()> {
     pub fn service(
         connector: SslConnector,
-    ) -> impl Service<(R, T), Response = (R, SslStream<T>), Error = Error> {
+    ) -> impl Service<(R, T), Response = (R, SslStream<T>), Error = HandshakeError<T>> {
         OpensslConnectorService {
             connector: connector,
             _t: PhantomData,
@@ -138,7 +47,7 @@ impl<R: RequestHost, T: AsyncRead + AsyncWrite, E> NewService<(R, T)>
     for OpensslConnector<R, T, E>
 {
     type Response = (R, SslStream<T>);
-    type Error = Error;
+    type Error = HandshakeError<T>;
     type Service = OpensslConnectorService<R, T>;
     type InitError = E;
     type Future = FutureResult<Self::Service, Self::InitError>;
@@ -160,7 +69,7 @@ impl<R: RequestHost, T: AsyncRead + AsyncWrite> Service<(R, T)>
     for OpensslConnectorService<R, T>
 {
     type Response = (R, SslStream<T>);
-    type Error = Error;
+    type Error = HandshakeError<T>;
     type Future = ConnectAsyncExt<R, T>;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
@@ -186,7 +95,7 @@ where
     T: AsyncRead + AsyncWrite,
 {
     type Item = (R, SslStream<T>);
-    type Error = Error;
+    type Error = HandshakeError<T>;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         match self.fut.poll()? {

actix-rt/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2018-12-09
+
+* Initial release

actix-rt/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "actix-rt"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Actix runtime"
+keywords = ["network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-rt/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = "../"
+
+[lib]
+name = "actix_rt"
+path = "src/lib.rs"
+
+[dependencies]
+log = "0.4"
+bytes = "0.4"
+futures = "0.1.24"
+tokio-current-thread = "0.1"
+tokio-executor = "0.1.5"
+tokio-reactor = "0.1.7"
+tokio-timer = "0.2.8"

actix-rt/src/arbiter.rs

@@ -0,0 +1,267 @@
+use std::cell::{Cell, RefCell};
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::{fmt, thread};
+
+use futures::sync::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
+use futures::sync::oneshot::{channel, Sender};
+use futures::{future, Async, Future, IntoFuture, Poll, Stream};
+use tokio_current_thread::spawn;
+
+use crate::builder::Builder;
+use crate::system::System;
+
+thread_local!(
+    static ADDR: RefCell<Option<Arbiter>> = RefCell::new(None);
+    static RUNNING: Cell<bool> = Cell::new(false);
+    static Q: RefCell<Vec<Box<Future<Item = (), Error = ()>>>> = RefCell::new(Vec::new());
+);
+
+pub(crate) static COUNT: AtomicUsize = AtomicUsize::new(0);
+
+pub(crate) enum ArbiterCommand {
+    Stop,
+    Execute(Box<Future<Item = (), Error = ()> + Send>),
+}
+
+impl fmt::Debug for ArbiterCommand {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            ArbiterCommand::Stop => write!(f, "ArbiterCommand::Stop"),
+            ArbiterCommand::Execute(_) => write!(f, "ArbiterCommand::Execute"),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct Arbiter(UnboundedSender<ArbiterCommand>);
+
+impl Default for Arbiter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Arbiter {
+    pub(crate) fn new_system() -> Self {
+        let (tx, rx) = unbounded();
+
+        let arb = Arbiter(tx);
+        ADDR.with(|cell| *cell.borrow_mut() = Some(arb.clone()));
+        RUNNING.with(|cell| cell.set(false));
+        Arbiter::spawn(ArbiterController { stop: None, rx });
+
+        arb
+    }
+
+    /// Returns current arbiter's address
+    pub fn current() -> Arbiter {
+        ADDR.with(|cell| match *cell.borrow() {
+            Some(ref addr) => addr.clone(),
+            None => panic!("Arbiter is not running"),
+        })
+    }
+
+    /// Stop arbiter
+    pub fn stop(&self) {
+        let _ = self.0.unbounded_send(ArbiterCommand::Stop);
+    }
+
+    /// Spawn new thread and run event loop in spawned thread.
+    /// Returns address of newly created arbiter.
+    pub fn new() -> Arbiter {
+        let id = COUNT.fetch_add(1, Ordering::Relaxed);
+        let name = format!("actix-rt:worker:{}", id);
+        let sys = System::current();
+        let (arb_tx, arb_rx) = unbounded();
+        let arb_tx2 = arb_tx.clone();
+
+        let _ = thread::Builder::new().name(name.clone()).spawn(move || {
+            let mut rt = Builder::new().build_rt().expect("Can not create Runtime");
+            let arb = Arbiter(arb_tx);
+
+            let (stop, stop_rx) = channel();
+            RUNNING.with(|cell| cell.set(true));
+
+            System::set_current(sys);
+
+            // start arbiter controller
+            rt.spawn(ArbiterController {
+                stop: Some(stop),
+                rx: arb_rx,
+            });
+            ADDR.with(|cell| *cell.borrow_mut() = Some(arb.clone()));
+
+            // register arbiter
+            let _ = System::current()
+                .sys()
+                .unbounded_send(SystemCommand::RegisterArbiter(id, arb.clone()));
+
+            // run loop
+            let _ = match rt.block_on(stop_rx) {
+                Ok(code) => code,
+                Err(_) => 1,
+            };
+
+            // unregister arbiter
+            let _ = System::current()
+                .sys()
+                .unbounded_send(SystemCommand::UnregisterArbiter(id));
+        });
+
+        Arbiter(arb_tx2)
+    }
+
+    pub(crate) fn run_system() {
+        RUNNING.with(|cell| cell.set(true));
+        Q.with(|cell| {
+            let mut v = cell.borrow_mut();
+            for fut in v.drain(..) {
+                spawn(fut);
+            }
+        });
+    }
+
+    pub(crate) fn stop_system() {
+        RUNNING.with(|cell| cell.set(false));
+    }
+
+    /// Spawn a future on the current thread.
+    pub fn spawn<F>(future: F)
+    where
+        F: Future<Item = (), Error = ()> + 'static,
+    {
+        RUNNING.with(move |cell| {
+            if cell.get() {
+                spawn(Box::new(future));
+            } else {
+                Q.with(move |cell| cell.borrow_mut().push(Box::new(future)));
+            }
+        });
+    }
+
+    /// Executes a future on the current thread.
+    pub fn spawn_fn<F, R>(f: F)
+    where
+        F: FnOnce() -> R + 'static,
+        R: IntoFuture<Item = (), Error = ()> + 'static,
+    {
+        Arbiter::spawn(future::lazy(f))
+    }
+
+    /// Send a future on the arbiter's thread and spawn.
+    pub fn send<F>(&self, future: F)
+    where
+        F: Future<Item = (), Error = ()> + Send + 'static,
+    {
+        let _ = self
+            .0
+            .unbounded_send(ArbiterCommand::Execute(Box::new(future)));
+    }
+}
+
+struct ArbiterController {
+    stop: Option<Sender<i32>>,
+    rx: UnboundedReceiver<ArbiterCommand>,
+}
+
+impl Drop for ArbiterController {
+    fn drop(&mut self) {
+        if thread::panicking() {
+            eprintln!("Panic in Arbiter thread, shutting down system.");
+            if System::current().stop_on_panic() {
+                System::current().stop_with_code(1)
+            }
+        }
+    }
+}
+
+impl Future for ArbiterController {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            match self.rx.poll() {
+                Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
+                Ok(Async::Ready(Some(item))) => match item {
+                    ArbiterCommand::Stop => {
+                        if let Some(stop) = self.stop.take() {
+                            let _ = stop.send(0);
+                        };
+                        return Ok(Async::Ready(()));
+                    }
+                    ArbiterCommand::Execute(fut) => {
+                        spawn(fut);
+                    }
+                },
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+            }
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum SystemCommand {
+    Exit(i32),
+    RegisterArbiter(usize, Arbiter),
+    UnregisterArbiter(usize),
+}
+
+#[derive(Debug)]
+pub(crate) struct SystemArbiter {
+    stop: Option<Sender<i32>>,
+    commands: UnboundedReceiver<SystemCommand>,
+    arbiters: HashMap<usize, Arbiter>,
+}
+
+impl SystemArbiter {
+    pub(crate) fn new(stop: Sender<i32>, commands: UnboundedReceiver<SystemCommand>) -> Self {
+        SystemArbiter {
+            commands,
+            stop: Some(stop),
+            arbiters: HashMap::new(),
+        }
+    }
+}
+
+impl Future for SystemArbiter {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            match self.commands.poll() {
+                Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
+                Ok(Async::Ready(Some(cmd))) => match cmd {
+                    SystemCommand::Exit(code) => {
+                        // stop arbiters
+                        for arb in self.arbiters.values() {
+                            arb.stop();
+                        }
+                        // stop event loop
+                        if let Some(stop) = self.stop.take() {
+                            let _ = stop.send(code);
+                        }
+                    }
+                    SystemCommand::RegisterArbiter(name, hnd) => {
+                        self.arbiters.insert(name, hnd);
+                    }
+                    SystemCommand::UnregisterArbiter(name) => {
+                        self.arbiters.remove(&name);
+                    }
+                },
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+            }
+        }
+    }
+}
+
+// /// Execute function in arbiter's thread
+// impl<I: Send, E: Send> Handler<Execute<I, E>> for SystemArbiter {
+//     type Result = Result<I, E>;
+
+//     fn handle(&mut self, msg: Execute<I, E>, _: &mut Context<Self>) -> Result<I, E> {
+//         msg.exec()
+//     }
+// }

actix-rt/src/builder.rs

@@ -0,0 +1,175 @@
+use std::borrow::Cow;
+use std::io;
+
+use futures::future::{lazy, Future};
+use futures::sync::mpsc::unbounded;
+use futures::sync::oneshot::{channel, Receiver};
+
+use tokio_current_thread::CurrentThread;
+use tokio_reactor::Reactor;
+use tokio_timer::clock::Clock;
+use tokio_timer::timer::Timer;
+
+use crate::arbiter::{Arbiter, SystemArbiter};
+use crate::runtime::Runtime;
+use crate::system::System;
+
+/// Builder struct for a actix runtime.
+///
+/// Either use `Builder::build` to create a system and start actors.
+/// Alternatively, use `Builder::run` to start the tokio runtime and
+/// run a function in its context.
+pub struct Builder {
+    /// Name of the System. Defaults to "actix" if unset.
+    name: Cow<'static, str>,
+
+    /// The clock to use
+    clock: Clock,
+
+    /// Whether the Arbiter will stop the whole System on uncaught panic. Defaults to false.
+    stop_on_panic: bool,
+}
+
+impl Builder {
+    pub(crate) fn new() -> Self {
+        Builder {
+            name: Cow::Borrowed("actix"),
+            clock: Clock::new(),
+            stop_on_panic: false,
+        }
+    }
+
+    /// Sets the name of the System.
+    pub fn name<T: Into<String>>(mut self, name: T) -> Self {
+        self.name = Cow::Owned(name.into());
+        self
+    }
+
+    /// Set the Clock instance that will be used by this System.
+    ///
+    /// Defaults to the system clock.
+    pub fn clock(mut self, clock: Clock) -> Self {
+        self.clock = clock;
+        self
+    }
+
+    /// Sets the option 'stop_on_panic' which controls whether the System is stopped when an
+    /// uncaught panic is thrown from a worker thread.
+    ///
+    /// Defaults to false.
+    pub fn stop_on_panic(mut self, stop_on_panic: bool) -> Self {
+        self.stop_on_panic = stop_on_panic;
+        self
+    }
+
+    /// Create new System.
+    ///
+    /// This method panics if it can not create tokio runtime
+    pub fn build(self) -> SystemRunner {
+        self.create_runtime(|| {})
+    }
+
+    /// This function will start tokio runtime and will finish once the
+    /// `System::stop()` message get called.
+    /// Function `f` get called within tokio runtime context.
+    pub fn run<F>(self, f: F) -> i32
+    where
+        F: FnOnce() + 'static,
+    {
+        self.create_runtime(f).run()
+    }
+
+    fn create_runtime<F>(self, f: F) -> SystemRunner
+    where
+        F: FnOnce() + 'static,
+    {
+        let (stop_tx, stop) = channel();
+        let (sys_sender, sys_receiver) = unbounded();
+
+        let arbiter = Arbiter::new_system();
+        let system = System::construct(sys_sender, arbiter.clone(), self.stop_on_panic);
+
+        // system arbiter
+        let arb = SystemArbiter::new(stop_tx, sys_receiver);
+
+        let mut rt = self.build_rt().unwrap();
+        rt.spawn(arb);
+
+        // init system arbiter and run configuration method
+        let _ = rt.block_on(lazy(move || {
+            f();
+            Ok::<_, ()>(())
+        }));
+
+        SystemRunner { rt, stop, system }
+    }
+
+    pub(crate) fn build_rt(&self) -> io::Result<Runtime> {
+        // We need a reactor to receive events about IO objects from kernel
+        let reactor = Reactor::new()?;
+        let reactor_handle = reactor.handle();
+
+        // Place a timer wheel on top of the reactor. If there are no timeouts to fire, it'll let the
+        // reactor pick up some new external events.
+        let timer = Timer::new_with_now(reactor, self.clock.clone());
+        let timer_handle = timer.handle();
+
+        // And now put a single-threaded executor on top of the timer. When there are no futures ready
+        // to do something, it'll let the timer or the reactor to generate some new stimuli for the
+        // futures to continue in their life.
+        let executor = CurrentThread::new_with_park(timer);
+
+        Ok(Runtime::new2(
+            reactor_handle,
+            timer_handle,
+            self.clock.clone(),
+            executor,
+        ))
+    }
+}
+
+/// Helper object that runs System's event loop
+#[must_use = "SystemRunner must be run"]
+#[derive(Debug)]
+pub struct SystemRunner {
+    rt: Runtime,
+    stop: Receiver<i32>,
+    system: System,
+}
+
+impl SystemRunner {
+    /// This function will start event loop and will finish once the
+    /// `System::stop()` function is called.
+    pub fn run(self) -> i32 {
+        let SystemRunner { mut rt, stop, .. } = self;
+
+        // run loop
+        let _ = rt.block_on(lazy(move || {
+            Arbiter::run_system();
+            Ok::<_, ()>(())
+        }));
+        let code = match rt.block_on(stop) {
+            Ok(code) => code,
+            Err(_) => 1,
+        };
+        Arbiter::stop_system();
+        code
+    }
+
+    /// Execute a future and wait for result.
+    pub fn block_on<F, I, E>(&mut self, fut: F) -> Result<I, E>
+    where
+        F: Future<Item = I, Error = E>,
+    {
+        let _ = self.rt.block_on(lazy(move || {
+            Arbiter::run_system();
+            Ok::<_, ()>(())
+        }));
+        let res = self.rt.block_on(fut);
+        let _ = self.rt.block_on(lazy(move || {
+            Arbiter::stop_system();
+            Ok::<_, ()>(())
+        }));
+        res
+    }
+}

actix-rt/src/lib.rs

@@ -0,0 +1,27 @@
+//! A runtime implementation that runs everything on the current thread.
+
+mod arbiter;
+mod builder;
+mod runtime;
+mod system;
+
+pub use self::arbiter::Arbiter;
+pub use self::builder::{Builder, SystemRunner};
+pub use self::runtime::{Handle, Runtime};
+pub use self::system::System;
+
+/// Spawns a future on the current arbiter.
+///
+/// # Panics
+///
+/// This function panics if actix system is not running.
+pub fn spawn<F>(f: F)
+where
+    F: futures::Future<Item = (), Error = ()> + 'static,
+{
+    if !System::is_set() {
+        panic!("System is not running");
+    }
+
+    Arbiter::spawn(f);
+}

actix-rt/src/mod.rs

@@ -0,0 +1,92 @@
+//! A runtime implementation that runs everything on the current thread.
+//!
+//! [`current_thread::Runtime`][rt] is similar to the primary
+//! [`Runtime`][concurrent-rt] except that it runs all components on the current
+//! thread instead of using a thread pool. This means that it is able to spawn
+//! futures that do not implement `Send`.
+//!
+//! Same as the default [`Runtime`][concurrent-rt], the
+//! [`current_thread::Runtime`][rt] includes:
+//!
+//! * A [reactor] to drive I/O resources.
+//! * An [executor] to execute tasks that use these I/O resources.
+//! * A [timer] for scheduling work to run after a set period of time.
+//!
+//! Note that [`current_thread::Runtime`][rt] does not implement `Send` itself
+//! and cannot be safely moved to other threads.
+//!
+//! # Spawning from other threads
+//!
+//! While [`current_thread::Runtime`][rt] does not implement `Send` and cannot
+//! safely be moved to other threads, it provides a `Handle` that can be sent
+//! to other threads and allows to spawn new tasks from there.
+//!
+//! For example:
+//!
+//! ```
+//! # extern crate tokio;
+//! # extern crate futures;
+//! use tokio::runtime::current_thread::Runtime;
+//! use tokio::prelude::*;
+//! use std::thread;
+//!
+//! # fn main() {
+//! let mut runtime = Runtime::new().unwrap();
+//! let handle = runtime.handle();
+//!
+//! thread::spawn(move || {
+//!     handle.spawn(future::ok(()));
+//! }).join().unwrap();
+//!
+//! # /*
+//! runtime.run().unwrap();
+//! # */
+//! # }
+//! ```
+//!
+//! # Examples
+//!
+//! Creating a new `Runtime` and running a future `f` until its completion and
+//! returning its result.
+//!
+//! ```
+//! use tokio::runtime::current_thread::Runtime;
+//! use tokio::prelude::*;
+//!
+//! let mut runtime = Runtime::new().unwrap();
+//!
+//! // Use the runtime...
+//! // runtime.block_on(f); // where f is a future
+//! ```
+//!
+//! [rt]: struct.Runtime.html
+//! [concurrent-rt]: ../struct.Runtime.html
+//! [chan]: https://docs.rs/futures/0.1/futures/sync/mpsc/fn.channel.html
+//! [reactor]: ../../reactor/struct.Reactor.html
+//! [executor]: https://tokio.rs/docs/getting-started/runtime-model/#executors
+//! [timer]: ../../timer/index.html
+
+mod builder;
+mod runtime;
+
+pub use self::builder::Builder;
+pub use self::runtime::{Runtime, Handle};
+pub use tokio_current_thread::spawn;
+pub use tokio_current_thread::TaskExecutor;
+
+use futures::Future;
+
+/// Run the provided future to completion using a runtime running on the current thread.
+///
+/// This first creates a new [`Runtime`], and calls [`Runtime::block_on`] with the provided future,
+/// which blocks the current thread until the provided future completes. It then calls
+/// [`Runtime::run`] to wait for any other spawned futures to resolve.
+pub fn block_on_all<F>(future: F) -> Result<F::Item, F::Error>
+where
+    F: Future,
+{
+    let mut r = Runtime::new().expect("failed to start runtime on current thread");
+    let v = r.block_on(future)?;
+    r.run().expect("failed to resolve remaining futures");
+    Ok(v)
+}

actix-rt/src/runtime.rs

@@ -0,0 +1,236 @@
+use std::error::Error;
+use std::fmt;
+use std::io;
+
+use futures::{future, Future};
+use tokio_current_thread::Handle as ExecutorHandle;
+use tokio_current_thread::{self as current_thread, CurrentThread};
+use tokio_executor;
+use tokio_reactor::{self, Reactor};
+use tokio_timer::clock::{self, Clock};
+use tokio_timer::timer::{self, Timer};
+
+use crate::builder::Builder;
+
+/// Single-threaded runtime provides a way to start reactor
+/// and executor on the current thread.
+///
+/// See [module level][mod] documentation for more details.
+///
+/// [mod]: index.html
+#[derive(Debug)]
+pub struct Runtime {
+    reactor_handle: tokio_reactor::Handle,
+    timer_handle: timer::Handle,
+    clock: Clock,
+    executor: CurrentThread<Timer<Reactor>>,
+}
+
+/// Handle to spawn a future on the corresponding `CurrentThread` runtime instance
+#[derive(Debug, Clone)]
+pub struct Handle(ExecutorHandle);
+
+impl Handle {
+    /// Spawn a future onto the `CurrentThread` runtime instance corresponding to this handle
+    ///
+    /// # Panics
+    ///
+    /// This function panics if the spawn fails. Failure occurs if the `CurrentThread`
+    /// instance of the `Handle` does not exist anymore.
+    pub fn spawn<F>(&self, future: F) -> Result<(), tokio_executor::SpawnError>
+    where
+        F: Future<Item = (), Error = ()> + Send + 'static,
+    {
+        self.0.spawn(future)
+    }
+
+    /// Provides a best effort **hint** to whether or not `spawn` will succeed.
+    ///
+    /// This function may return both false positives **and** false negatives.
+    /// If `status` returns `Ok`, then a call to `spawn` will *probably*
+    /// succeed, but may fail. If `status` returns `Err`, a call to `spawn` will
+    /// *probably* fail, but may succeed.
+    ///
+    /// This allows a caller to avoid creating the task if the call to `spawn`
+    /// has a high likelihood of failing.
+    pub fn status(&self) -> Result<(), tokio_executor::SpawnError> {
+        self.0.status()
+    }
+}
+
+impl<T> future::Executor<T> for Handle
+where
+    T: Future<Item = (), Error = ()> + Send + 'static,
+{
+    fn execute(&self, future: T) -> Result<(), future::ExecuteError<T>> {
+        if let Err(e) = self.status() {
+            let kind = if e.is_at_capacity() {
+                future::ExecuteErrorKind::NoCapacity
+            } else {
+                future::ExecuteErrorKind::Shutdown
+            };
+
+            return Err(future::ExecuteError::new(kind, future));
+        }
+
+        let _ = self.spawn(future);
+        Ok(())
+    }
+}
+
+/// Error returned by the `run` function.
+#[derive(Debug)]
+pub struct RunError {
+    inner: current_thread::RunError,
+}
+
+impl fmt::Display for RunError {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        write!(fmt, "{}", self.inner)
+    }
+}
+
+impl Error for RunError {
+    fn description(&self) -> &str {
+        self.inner.description()
+    }
+    fn cause(&self) -> Option<&Error> {
+        self.inner.cause()
+    }
+}
+
+impl Runtime {
+    #[allow(clippy::new_ret_no_self)]
+    /// Returns a new runtime initialized with default configuration values.
+    pub fn new() -> io::Result<Runtime> {
+        Builder::new().build_rt()
+    }
+
+    pub(super) fn new2(
+        reactor_handle: tokio_reactor::Handle,
+        timer_handle: timer::Handle,
+        clock: Clock,
+        executor: CurrentThread<Timer<Reactor>>,
+    ) -> Runtime {
+        Runtime {
+            reactor_handle,
+            timer_handle,
+            clock,
+            executor,
+        }
+    }
+
+    /// Get a new handle to spawn futures on the single-threaded Tokio runtime
+    ///
+    /// Different to the runtime itself, the handle can be sent to different
+    /// threads.
+    pub fn handle(&self) -> Handle {
+        Handle(self.executor.handle().clone())
+    }
+
+    /// Spawn a future onto the single-threaded Tokio runtime.
+    ///
+    /// See [module level][mod] documentation for more details.
+    ///
+    /// [mod]: index.html
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// # use futures::{future, Future, Stream};
+    /// use actix_rt::Runtime;
+    ///
+    /// # fn dox() {
+    /// // Create the runtime
+    /// let mut rt = Runtime::new().unwrap();
+    ///
+    /// // Spawn a future onto the runtime
+    /// rt.spawn(future::lazy(|| {
+    ///     println!("running on the runtime");
+    ///     Ok(())
+    /// }));
+    /// # }
+    /// # pub fn main() {}
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if the spawn fails. Failure occurs if the executor
+    /// is currently at capacity and is unable to spawn a new future.
+    pub fn spawn<F>(&mut self, future: F) -> &mut Self
+    where
+        F: Future<Item = (), Error = ()> + 'static,
+    {
+        self.executor.spawn(future);
+        self
+    }
+
+    /// Runs the provided future, blocking the current thread until the future
+    /// completes.
+    ///
+    /// This function can be used to synchronously block the current thread
+    /// until the provided `future` has resolved either successfully or with an
+    /// error. The result of the future is then returned from this function
+    /// call.
+    ///
+    /// Note that this function will **also** execute any spawned futures on the
+    /// current thread, but will **not** block until these other spawned futures
+    /// have completed. Once the function returns, any uncompleted futures
+    /// remain pending in the `Runtime` instance. These futures will not run
+    /// until `block_on` or `run` is called again.
+    ///
+    /// The caller is responsible for ensuring that other spawned futures
+    /// complete execution by calling `block_on` or `run`.
+    pub fn block_on<F>(&mut self, f: F) -> Result<F::Item, F::Error>
+    where
+        F: Future,
+    {
+        self.enter(|executor| {
+            // Run the provided future
+            let ret = executor.block_on(f);
+            ret.map_err(|e| e.into_inner().expect("unexpected execution error"))
+        })
+    }
+
+    /// Run the executor to completion, blocking the thread until **all**
+    /// spawned futures have completed.
+    pub fn run(&mut self) -> Result<(), RunError> {
+        self.enter(|executor| executor.run())
+            .map_err(|e| RunError { inner: e })
+    }
+
+    fn enter<F, R>(&mut self, f: F) -> R
+    where
+        F: FnOnce(&mut current_thread::Entered<Timer<Reactor>>) -> R,
+    {
+        let Runtime {
+            ref reactor_handle,
+            ref timer_handle,
+            ref clock,
+            ref mut executor,
+            ..
+        } = *self;
+
+        // Binds an executor to this thread
+        let mut enter = tokio_executor::enter().expect("Multiple executors at once");
+
+        // This will set the default handle and timer to use inside the closure
+        // and run the future.
+        tokio_reactor::with_default(&reactor_handle, &mut enter, |enter| {
+            clock::with_default(clock, enter, |enter| {
+                timer::with_default(&timer_handle, enter, |enter| {
+                    // The TaskExecutor is a fake executor that looks into the
+                    // current single-threaded executor when used. This is a trick,
+                    // because we need two mutable references to the executor (one
+                    // to run the provided future, another to install as the default
+                    // one). We use the fake one here as the default one.
+                    let mut default_executor = current_thread::TaskExecutor::current();
+                    tokio_executor::with_default(&mut default_executor, enter, |enter| {
+                        let mut executor = executor.enter(enter);
+                        f(&mut executor)
+                    })
+                })
+            })
+        })
+    }
+}

actix-rt/src/system.rs

@@ -0,0 +1,118 @@
+use std::cell::RefCell;
+
+use futures::sync::mpsc::UnboundedSender;
+
+use crate::arbiter::{Arbiter, SystemCommand};
+use crate::builder::{Builder, SystemRunner};
+
+/// System is a runtime manager.
+#[derive(Clone, Debug)]
+pub struct System {
+    sys: UnboundedSender<SystemCommand>,
+    arbiter: Arbiter,
+    stop_on_panic: bool,
+}
+
+thread_local!(
+    static CURRENT: RefCell<Option<System>> = RefCell::new(None);
+);
+
+impl System {
+    /// Constructs new system and sets it as current
+    pub(crate) fn construct(
+        sys: UnboundedSender<SystemCommand>,
+        arbiter: Arbiter,
+        stop_on_panic: bool,
+    ) -> Self {
+        let sys = System {
+            sys,
+            arbiter,
+            stop_on_panic,
+        };
+        System::set_current(sys.clone());
+        sys
+    }
+
+    /// Build a new system with a customized tokio runtime.
+    ///
+    /// This allows to customize the runtime. See struct level docs on
+    /// `Builder` for more information.
+    pub fn builder() -> Builder {
+        Builder::new()
+    }
+
+    #[allow(clippy::new_ret_no_self)]
+    /// Create new system.
+    ///
+    /// This method panics if it can not create tokio runtime
+    pub fn new<T: Into<String>>(name: T) -> SystemRunner {
+        Self::builder().name(name).build()
+    }
+
+    /// Get current running system.
+    pub fn current() -> System {
+        CURRENT.with(|cell| match *cell.borrow() {
+            Some(ref sys) => sys.clone(),
+            None => panic!("System is not running"),
+        })
+    }
+
+    /// Set current running system.
+    pub(crate) fn is_set() -> bool {
+        CURRENT.with(|cell| cell.borrow().is_some())
+    }
+
+    /// Set current running system.
+    #[doc(hidden)]
+    pub fn set_current(sys: System) {
+        CURRENT.with(|s| {
+            *s.borrow_mut() = Some(sys);
+        })
+    }
+
+    /// Execute function with system reference.
+    pub fn with_current<F, R>(f: F) -> R
+    where
+        F: FnOnce(&System) -> R,
+    {
+        CURRENT.with(|cell| match *cell.borrow() {
+            Some(ref sys) => f(sys),
+            None => panic!("System is not running"),
+        })
+    }
+
+    /// Stop the system
+    pub fn stop(&self) {
+        self.stop_with_code(0)
+    }
+
+    /// Stop the system with a particular exit code.
+    pub fn stop_with_code(&self, code: i32) {
+        let _ = self.sys.unbounded_send(SystemCommand::Exit(code));
+    }
+
+    pub(crate) fn sys(&self) -> &UnboundedSender<SystemCommand> {
+        &self.sys
+    }
+
+    /// Return status of 'stop_on_panic' option which controls whether the System is stopped when an
+    /// uncaught panic is thrown from a worker thread.
+    pub fn stop_on_panic(&self) -> bool {
+        self.stop_on_panic
+    }
+
+    /// System arbiter
+    pub fn arbiter(&self) -> &Arbiter {
+        &self.arbiter
+    }
+
+    /// This function will start tokio runtime and will finish once the
+    /// `System::stop()` message get called.
+    /// Function `f` get called within tokio runtime context.
+    pub fn run<F>(f: F) -> i32
+    where
+        F: FnOnce() + 'static,
+    {
+        Self::builder().run(f)
+    }
+}

actix-server/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2018-12-09
+
+* Move server to separate crate

actix-server/Cargo.toml

@@ -0,0 +1,68 @@
+[package]
+name = "actix-server"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Actix server - General purpose tcp server"
+keywords = ["network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-server/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = "../"
+
+[package.metadata.docs.rs]
+features = ["ssl", "tls", "rust-tls"]
+
+[lib]
+name = "actix_server"
+path = "src/lib.rs"
+
+[features]
+default = []
+
+# tls
+tls = ["native-tls"]
+
+# openssl
+ssl = ["openssl", "tokio-openssl"]
+
+# rustls
+rust-tls = ["rustls", "tokio-rustls", "webpki", "webpki-roots"]
+
+[dependencies]
+actix-service = "0.1.1"
+actix-rt = "0.1.0"
+
+log = "0.4"
+num_cpus = "1.0"
+
+# io
+mio = "^0.6.13"
+net2 = "0.2"
+bytes = "0.4"
+futures = "0.1"
+slab = "0.4"
+tokio-io = "0.1"
+tokio-tcp = "0.1"
+tokio-timer = "0.2"
+tokio-reactor = "0.1"
+tokio-signal = "0.2"
+
+# native-tls
+native-tls = { version="0.2", optional = true }
+
+# openssl
+openssl = { version="0.10", optional = true }
+tokio-openssl = { version="0.3", optional = true }
+
+#rustls
+rustls = { version = "^0.14", optional = true }
+tokio-rustls = { version = "^0.8", optional = true }
+webpki = { version = "0.18", optional = true }
+webpki-roots = { version = "0.15", optional = true }
+
+[dev-dependencies]
+env_logger = "0.5"

actix-server/src/accept.rs

@@ -2,15 +2,14 @@ use std::sync::mpsc as sync_mpsc;
 use std::time::{Duration, Instant};
 use std::{io, net, thread};
 
-use futures::{sync::mpsc, Future};
+use actix_rt::System;
+use futures::future::{lazy, Future};
 use log::{error, info};
 use mio;
 use slab::Slab;
 use tokio_timer::Delay;
 
-use actix::{msgs::Execute, Arbiter, System};
+use super::server::Server;
-
-use super::server::ServerCommand;
 use super::worker::{Conn, WorkerClient};
 use super::Token;
 
@@ -54,14 +53,11 @@ pub(crate) struct AcceptLoop {
     notify_ready: mio::SetReadiness,
     tx: sync_mpsc::Sender<Command>,
     rx: Option<sync_mpsc::Receiver<Command>>,
-    srv: Option<(
+    srv: Option<Server>,
-        mpsc::UnboundedSender<ServerCommand>,
-        mpsc::UnboundedReceiver<ServerCommand>,
-    )>,
 }
 
 impl AcceptLoop {
-    pub fn new() -> AcceptLoop {
+    pub fn new(srv: Server) -> AcceptLoop {
         let (tx, rx) = sync_mpsc::channel();
         let (cmd_reg, cmd_ready) = mio::Registration::new2();
         let (notify_reg, notify_ready) = mio::Registration::new2();
@@ -73,7 +69,7 @@ impl AcceptLoop {
             notify_ready,
             notify_reg: Some(notify_reg),
             rx: Some(rx),
-            srv: Some(mpsc::unbounded()),
+            srv: Some(srv),
         }
     }
 
@@ -90,18 +86,17 @@ impl AcceptLoop {
         &mut self,
         socks: Vec<(Token, net::TcpListener)>,
         workers: Vec<WorkerClient>,
-    ) -> mpsc::UnboundedReceiver<ServerCommand> {
+    ) {
-        let (tx, rx) = self.srv.take().expect("Can not re-use AcceptInfo");
+        let srv = self.srv.take().expect("Can not re-use AcceptInfo");
 
         Accept::start(
             self.rx.take().expect("Can not re-use AcceptInfo"),
             self.cmd_reg.take().expect("Can not re-use AcceptInfo"),
             self.notify_reg.take().expect("Can not re-use AcceptInfo"),
             socks,
-            tx,
+            srv,
             workers,
         );
-        rx
     }
 }
 
@@ -110,7 +105,7 @@ struct Accept {
     rx: sync_mpsc::Receiver<Command>,
     sockets: Slab<ServerSocketInfo>,
     workers: Vec<WorkerClient>,
-    srv: mpsc::UnboundedSender<ServerCommand>,
+    srv: Server,
     timer: (mio::Registration, mio::SetReadiness),
     next: usize,
     backpressure: bool,
@@ -141,14 +136,14 @@ impl Accept {
         cmd_reg: mio::Registration,
         notify_reg: mio::Registration,
         socks: Vec<(Token, net::TcpListener)>,
-        srv: mpsc::UnboundedSender<ServerCommand>,
+        srv: Server,
         workers: Vec<WorkerClient>,
     ) {
         let sys = System::current();
 
         // start accept thread
         let _ = thread::Builder::new()
-            .name("actix-web accept loop".to_owned())
+            .name("actix-server accept loop".to_owned())
             .spawn(move || {
                 System::set_current(sys);
                 let mut accept = Accept::new(rx, socks, workers, srv);
@@ -181,7 +176,7 @@ impl Accept {
         rx: sync_mpsc::Receiver<Command>,
         socks: Vec<(Token, net::TcpListener)>,
         workers: Vec<WorkerClient>,
-        srv: mpsc::UnboundedSender<ServerCommand>,
+        srv: Server,
     ) -> Accept {
         // Create a poll instance
         let poll = match mio::Poll::new() {
@@ -376,9 +371,7 @@ impl Accept {
                 match self.workers[self.next].send(msg) {
                     Ok(_) => (),
                     Err(tmp) => {
-                        let _ = self.srv.unbounded_send(ServerCommand::WorkerDied(
+                        self.srv.worker_died(self.workers[self.next].idx);
-                            self.workers[self.next].idx,
-                        ));
                         msg = tmp;
                         self.workers.swap_remove(self.next);
                         if self.workers.is_empty() {
@@ -404,9 +397,7 @@ impl Accept {
                             return;
                         }
                         Err(tmp) => {
-                            let _ = self.srv.unbounded_send(ServerCommand::WorkerDied(
+                            self.srv.worker_died(self.workers[self.next].idx);
-                                self.workers[self.next].idx,
-                            ));
                             msg = tmp;
                             self.workers.swap_remove(self.next);
                             if self.workers.is_empty() {
@@ -449,19 +440,14 @@ impl Accept {
                         info.timeout = Some(Instant::now() + Duration::from_millis(500));
 
                         let r = self.timer.1.clone();
-                        System::current().arbiter().do_send(Execute::new(
+                        System::current().arbiter().send(lazy(move || {
-                            move || -> Result<(), ()> {
-                                Arbiter::spawn(
                             Delay::new(Instant::now() + Duration::from_millis(510))
                                 .map_err(|_| ())
                                 .and_then(move |_| {
                                     let _ = r.set_readiness(mio::Ready::readable());
                                     Ok(())
-                                        }),
+                                })
-                                );
+                        }));
-                                Ok(())
-                            },
-                        ));
                         return;
                     }
                 }

actix-server/src/builder.rs

@@ -1,30 +1,27 @@
 use std::time::Duration;
 use std::{io, mem, net};
 
-use futures::sync::{mpsc, mpsc::unbounded};
+use actix_rt::{spawn, Arbiter, System};
-use futures::{Future, Sink, Stream};
+use futures::future::{lazy, ok};
+use futures::stream::futures_unordered;
+use futures::sync::mpsc::{unbounded, UnboundedReceiver};
+use futures::{Async, Future, Poll, Stream};
 use log::{error, info};
 use net2::TcpBuilder;
 use num_cpus;
+use tokio_timer::sleep;
 
-use actix::{
+use crate::accept::{AcceptLoop, AcceptNotify, Command};
-    actors::signal, fut, msgs::Execute, Actor, ActorFuture, Addr, Arbiter, AsyncContext,
+use crate::config::{ConfiguredService, ServiceConfig};
-    Context, Handler, Response, StreamHandler, System, WrapFuture,
+use crate::server::{Server, ServerCommand};
-};
+use crate::services::{InternalServiceFactory, StreamNewService, StreamServiceFactory};
+use crate::services::{ServiceFactory, ServiceNewService};
+use crate::signals::{Signal, Signals};
+use crate::worker::{self, Worker, WorkerAvailability, WorkerClient};
+use crate::Token;
 
-use super::accept::{AcceptLoop, AcceptNotify, Command};
+/// Server builder
-use super::config::{ConfiguredService, ServiceConfig};
+pub struct ServerBuilder {
-use super::services::{InternalServiceFactory, StreamNewService, StreamServiceFactory};
-use super::services::{ServiceFactory, ServiceNewService};
-use super::worker::{self, Worker, WorkerAvailability, WorkerClient};
-use super::{PauseServer, ResumeServer, StopServer, Token};
-
-pub(crate) enum ServerCommand {
-    WorkerDied(usize),
-}
-
-/// Server
-pub struct Server {
     threads: usize,
     token: Token,
     workers: Vec<(usize, WorkerClient)>,
@@ -33,30 +30,35 @@ pub struct Server {
     accept: AcceptLoop,
     exit: bool,
     shutdown_timeout: Duration,
-    signals: Option<Addr<signal::ProcessSignals>>,
     no_signals: bool,
+    cmd: UnboundedReceiver<ServerCommand>,
+    server: Server,
 }
 
-impl Default for Server {
+impl Default for ServerBuilder {
     fn default() -> Self {
         Self::new()
     }
 }
 
-impl Server {
+impl ServerBuilder {
-    /// Create new Server instance
+    /// Create new Server builder instance
-    pub fn new() -> Server {
+    pub fn new() -> ServerBuilder {
-        Server {
+        let (tx, rx) = unbounded();
+        let server = Server::new(tx);
+
+        ServerBuilder {
             threads: num_cpus::get(),
             token: Token(0),
             workers: Vec::new(),
             services: Vec::new(),
             sockets: Vec::new(),
-            accept: AcceptLoop::new(),
+            accept: AcceptLoop::new(server.clone()),
             exit: false,
             shutdown_timeout: Duration::from_secs(30),
-            signals: None,
             no_signals: false,
+            cmd: rx,
+            server,
         }
     }
 
@@ -88,13 +90,6 @@ impl Server {
         self
     }
 
-    #[doc(hidden)]
-    /// Set alternative address for `ProcessSignals` actor.
-    pub fn signals(mut self, addr: Addr<signal::ProcessSignals>) -> Self {
-        self.signals = Some(addr);
-        self
-    }
-
     /// Disable signal handling
     pub fn disable_signals(mut self) -> Self {
         self.no_signals = true;
@@ -118,7 +113,7 @@ impl Server {
     ///
     /// This function is useful for moving parts of configuration to a
     /// different module or even library.
-    pub fn configure<F>(mut self, f: F) -> io::Result<Server>
+    pub fn configure<F>(mut self, f: F) -> io::Result<ServerBuilder>
     where
         F: Fn(&mut ServiceConfig) -> io::Result<()>,
     {
@@ -137,7 +132,7 @@ impl Server {
         Ok(self)
     }
 
-    /// Add new service to server
+    /// Add new service to the server.
     pub fn bind<F, U, N: AsRef<str>>(mut self, name: N, addr: U, factory: F) -> io::Result<Self>
     where
         F: StreamServiceFactory,
@@ -158,7 +153,7 @@ impl Server {
         Ok(self)
     }
 
-    /// Add new service to server
+    /// Add new service to the server.
     pub fn listen<F, N: AsRef<str>>(
         mut self,
         name: N,
@@ -178,7 +173,7 @@ impl Server {
         self
     }
 
-    /// Add new service to server
+    /// Add new service to the server.
     pub fn listen2<F, N: AsRef<str>>(
         mut self,
         name: N,
@@ -226,10 +221,10 @@ impl Server {
         sys.run();
     }
 
-    /// Starts Server Actor and returns its address
+    /// Starts processing incoming connections and return server controller.
-    pub fn start(mut self) -> Addr<Server> {
+    pub fn start(mut self) -> Server {
         if self.sockets.is_empty() {
-            panic!("Service should have at least one bound socket");
+            panic!("Server should have at least one bound socket");
         } else {
             info!("Starting {} workers", self.threads);
 
@@ -245,33 +240,18 @@ impl Server {
             for sock in &self.sockets {
                 info!("Starting server on {}", sock.1.local_addr().ok().unwrap());
             }
-            let rx = self
+            self.accept
-                .accept
                 .start(mem::replace(&mut self.sockets, Vec::new()), workers);
 
-            // start http server actor
+            // handle signals
-            let signals = self.subscribe_to_signals();
+            if !self.no_signals {
-            let addr = Actor::create(move |ctx| {
+                Signals::start(self.server.clone());
-                ctx.add_stream(rx);
-                self
-            });
-            if let Some(signals) = signals {
-                signals.do_send(signal::Subscribe(addr.clone().recipient()))
-            }
-            addr
-        }
             }
 
-    // subscribe to os signals
+            // start http server actor
-    fn subscribe_to_signals(&self) -> Option<Addr<signal::ProcessSignals>> {
+            let server = self.server.clone();
-        if !self.no_signals {
+            spawn(self);
-            if let Some(ref signals) = self.signals {
+            server
-                Some(signals.clone())
-            } else {
-                Some(System::current().registry().get::<signal::ProcessSignals>())
-            }
-        } else {
-            None
         }
     }
 
@@ -284,118 +264,99 @@ impl Server {
         let services: Vec<Box<InternalServiceFactory>> =
             self.services.iter().map(|v| v.clone_factory()).collect();
 
-        Arbiter::new(format!("actix-net-worker-{}", idx)).do_send(Execute::new(move || {
+        Arbiter::new().send(lazy(move || {
             Worker::start(rx1, rx2, services, avail, timeout);
             Ok::<_, ()>(())
         }));
 
         worker
     }
-}
 
-impl Actor for Server {
+    fn handle_cmd(&mut self, item: ServerCommand) {
-    type Context = Context<Self>;
+        match item {
-}
+            ServerCommand::Pause(tx) => {
-
+                self.accept.send(Command::Pause);
-/// Signals support
+                let _ = tx.send(());
-/// Handle `SIGINT`, `SIGTERM`, `SIGQUIT` signals and stop actix system
+            }
-/// message to `System` actor.
+            ServerCommand::Resume(tx) => {
-impl Handler<signal::Signal> for Server {
+                self.accept.send(Command::Resume);
-    type Result = ();
+                let _ = tx.send(());
-
+            }
-    fn handle(&mut self, msg: signal::Signal, ctx: &mut Context<Self>) {
+            ServerCommand::Signal(sig) => {
-        match msg.0 {
+                // Signals support
-            signal::SignalType::Int => {
+                // Handle `SIGINT`, `SIGTERM`, `SIGQUIT` signals and stop actix system
+                match sig {
+                    Signal::Int => {
                         info!("SIGINT received, exiting");
                         self.exit = true;
-                Handler::<StopServer>::handle(self, StopServer { graceful: false }, ctx);
+                        self.handle_cmd(ServerCommand::Stop {
+                            graceful: false,
+                            completion: None,
+                        })
                     }
-            signal::SignalType::Term => {
+                    Signal::Term => {
                         info!("SIGTERM received, stopping");
                         self.exit = true;
-                Handler::<StopServer>::handle(self, StopServer { graceful: true }, ctx);
+                        self.handle_cmd(ServerCommand::Stop {
+                            graceful: true,
+                            completion: None,
+                        })
                     }
-            signal::SignalType::Quit => {
+                    Signal::Quit => {
                         info!("SIGQUIT received, exiting");
                         self.exit = true;
-                Handler::<StopServer>::handle(self, StopServer { graceful: false }, ctx);
+                        self.handle_cmd(ServerCommand::Stop {
+                            graceful: false,
+                            completion: None,
+                        })
                     }
                     _ => (),
                 }
             }
-}
+            ServerCommand::Stop {
+                graceful,
+                completion,
+            } => {
+                let exit = self.exit;
 
-impl Handler<PauseServer> for Server {
-    type Result = ();
-
-    fn handle(&mut self, _: PauseServer, _: &mut Context<Self>) {
-        self.accept.send(Command::Pause);
-    }
-}
-
-impl Handler<ResumeServer> for Server {
-    type Result = ();
-
-    fn handle(&mut self, _: ResumeServer, _: &mut Context<Self>) {
-        self.accept.send(Command::Resume);
-    }
-}
-
-impl Handler<StopServer> for Server {
-    type Result = Response<(), ()>;
-
-    fn handle(&mut self, msg: StopServer, ctx: &mut Context<Self>) -> Self::Result {
                 // stop accept thread
                 self.accept.send(Command::Stop);
 
                 // stop workers
-        let (tx, rx) = mpsc::channel(1);
-
-        for worker in &self.workers {
-            let tx2 = tx.clone();
-            ctx.spawn(
-                worker
-                    .1
-                    .stop(msg.graceful)
-                    .into_actor(self)
-                    .then(move |_, slf, ctx| {
-                        slf.workers.pop();
-                        if slf.workers.is_empty() {
-                            let _ = tx2.send(());
-
-                            // we need to stop system if server was spawned
-                            if slf.exit {
-                                ctx.run_later(Duration::from_millis(300), |_, _| {
-                                    System::current().stop();
-                                });
-                            }
-                        }
-
-                        fut::ok(())
-                    }),
-            );
-        }
-
                 if !self.workers.is_empty() {
-            Response::r#async(rx.into_future().map(|_| ()).map_err(|_| ()))
+                    spawn(
+                        futures_unordered(
+                            self.workers
+                                .iter()
+                                .map(move |worker| worker.1.stop(graceful)),
+                        )
+                        .collect()
+                        .then(move |_| {
+                            if let Some(tx) = completion {
+                                let _ = tx.send(());
+                            }
+                            if exit {
+                                spawn(sleep(Duration::from_millis(300)).then(|_| {
+                                    System::current().stop();
+                                    ok(())
+                                }));
+                            }
+                            ok(())
+                        }),
+                    )
                 } else {
                     // we need to stop system if server was spawned
                     if self.exit {
-                ctx.run_later(Duration::from_millis(300), |_, _| {
+                        spawn(sleep(Duration::from_millis(300)).then(|_| {
                             System::current().stop();
-                });
+                            ok(())
+                        }));
+                    }
+                    if let Some(tx) = completion {
+                        let _ = tx.send(());
                     }
-            Response::reply(Ok(()))
                 }
             }
-}
-
-/// Commands from accept threads
-impl StreamHandler<ServerCommand, ()> for Server {
-    fn finished(&mut self, _: &mut Context<Self>) {}
-
-    fn handle(&mut self, msg: ServerCommand, _: &mut Context<Self>) {
-        match msg {
             ServerCommand::WorkerDied(idx) => {
                 let mut found = false;
                 for i in 0..self.workers.len() {
@@ -429,6 +390,21 @@ impl StreamHandler<ServerCommand, ()> for Server {
     }
 }
 
+impl Future for ServerBuilder {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            match self.cmd.poll() {
+                Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Ready(Some(item))) => self.handle_cmd(item),
+            }
+        }
+    }
+}
+
 pub(super) fn bind_addr<S: net::ToSocketAddrs>(addr: S) -> io::Result<Vec<net::TcpListener>> {
     let mut err = None;
     let mut succ = false;

actix-server/src/config.rs

@@ -8,7 +8,7 @@ use tokio_tcp::TcpStream;
 
 use crate::counter::CounterGuard;
 
-use super::server::bind_addr;
+use super::builder::bind_addr;
 use super::services::{
     BoxedServerService, InternalServiceFactory, ServerMessage, StreamService,
 };

actix-server/src/lib.rs

@@ -0,0 +1,33 @@
+//! General purpose tcp server
+
+mod accept;
+mod builder;
+mod config;
+mod counter;
+mod server;
+mod services;
+mod signals;
+pub mod ssl;
+mod worker;
+
+pub use self::builder::ServerBuilder;
+pub use self::config::{ServiceConfig, ServiceRuntime};
+pub use self::server::Server;
+pub use self::services::{ServerMessage, ServiceFactory, StreamServiceFactory};
+
+/// Socket id token
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub(crate) struct Token(usize);
+
+impl Token {
+    pub(crate) fn next(&mut self) -> Token {
+        let token = Token(self.0 + 1);
+        self.0 += 1;
+        token
+    }
+}
+
+/// Start server building process
+pub fn build() -> ServerBuilder {
+    ServerBuilder::default()
+}

actix-server/src/server.rs

@@ -0,0 +1,69 @@
+use futures::sync::mpsc::UnboundedSender;
+use futures::sync::oneshot;
+use futures::Future;
+
+use crate::builder::ServerBuilder;
+use crate::signals::Signal;
+
+pub(crate) enum ServerCommand {
+    WorkerDied(usize),
+    Pause(oneshot::Sender<()>),
+    Resume(oneshot::Sender<()>),
+    Signal(Signal),
+    /// Whether to try and shut down gracefully
+    Stop {
+        graceful: bool,
+        completion: Option<oneshot::Sender<()>>,
+    },
+}
+
+#[derive(Clone)]
+pub struct Server(UnboundedSender<ServerCommand>);
+
+impl Server {
+    pub(crate) fn new(tx: UnboundedSender<ServerCommand>) -> Self {
+        Server(tx)
+    }
+
+    /// Start server building process
+    pub fn build() -> ServerBuilder {
+        ServerBuilder::default()
+    }
+
+    pub(crate) fn signal(&self, sig: Signal) {
+        let _ = self.0.unbounded_send(ServerCommand::Signal(sig));
+    }
+
+    pub(crate) fn worker_died(&self, idx: usize) {
+        let _ = self.0.unbounded_send(ServerCommand::WorkerDied(idx));
+    }
+
+    /// Pause accepting incoming connections
+    ///
+    /// If socket contains some pending connection, they might be dropped.
+    /// All opened connection remains active.
+    pub fn pause(&self) -> impl Future<Item = (), Error = ()> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.0.unbounded_send(ServerCommand::Pause(tx));
+        rx.map_err(|_| ())
+    }
+
+    /// Resume accepting incoming connections
+    pub fn resume(&self) -> impl Future<Item = (), Error = ()> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.0.unbounded_send(ServerCommand::Resume(tx));
+        rx.map_err(|_| ())
+    }
+
+    /// Stop incoming connection processing, stop all workers and exit.
+    ///
+    /// If server starts with `spawn()` method, then spawned thread get terminated.
+    pub fn stop(&self, graceful: bool) -> impl Future<Item = (), Error = ()> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.0.unbounded_send(ServerCommand::Stop {
+            graceful,
+            completion: Some(tx),
+        });
+        rx.map_err(|_| ())
+    }
+}

actix-server/src/services.rs

@@ -1,11 +1,11 @@
 use std::net;
 use std::time::Duration;
 
+use actix_rt::spawn;
 use actix_service::{NewService, Service};
 use futures::future::{err, ok, FutureResult};
 use futures::{Future, Poll};
 use log::error;
-use tokio_current_thread::spawn;
 use tokio_reactor::Handle;
 use tokio_tcp::TcpStream;
 

actix-server/src/signals.rs

@@ -0,0 +1,116 @@
+use std::io;
+
+use actix_rt::spawn;
+use futures::stream::futures_unordered;
+use futures::{Async, Future, Poll, Stream};
+
+use crate::server::Server;
+
+/// Different types of process signals
+#[derive(PartialEq, Clone, Copy, Debug)]
+pub(crate) enum Signal {
+    /// SIGHUP
+    Hup,
+    /// SIGINT
+    Int,
+    /// SIGTERM
+    Term,
+    /// SIGQUIT
+    Quit,
+}
+
+pub(crate) struct Signals {
+    srv: Server,
+    #[cfg(not(unix))]
+    stream: SigStream,
+    #[cfg(unix)]
+    streams: Vec<SigStream>,
+}
+
+type SigStream = Box<Stream<Item = Signal, Error = io::Error>>;
+
+impl Signals {
+    pub(crate) fn start(srv: Server) {
+        let fut = {
+            #[cfg(not(unix))]
+            {
+                tokio_signal::ctrl_c().and_then(move |stream| Signals {
+                    srv,
+                    stream: Box::new(stream.map(|_| Signal::Int)),
+                })
+            }
+
+            #[cfg(unix)]
+            {
+                use tokio_signal::unix;
+
+                let mut sigs: Vec<Box<Future<Item = SigStream, Error = io::Error>>> =
+                    Vec::new();
+                sigs.push(Box::new(
+                    tokio_signal::unix::Signal::new(tokio_signal::unix::SIGINT).map(|stream| {
+                        let s: SigStream = Box::new(stream.map(|_| Signal::Int));
+                        s
+                    }),
+                ));
+                sigs.push(Box::new(
+                    tokio_signal::unix::Signal::new(tokio_signal::unix::SIGHUP).map(
+                        |stream: unix::Signal| {
+                            let s: SigStream = Box::new(stream.map(|_| Signal::Hup));
+                            s
+                        },
+                    ),
+                ));
+                sigs.push(Box::new(
+                    tokio_signal::unix::Signal::new(tokio_signal::unix::SIGTERM).map(
+                        |stream| {
+                            let s: SigStream = Box::new(stream.map(|_| Signal::Term));
+                            s
+                        },
+                    ),
+                ));
+                sigs.push(Box::new(
+                    tokio_signal::unix::Signal::new(tokio_signal::unix::SIGQUIT).map(
+                        |stream| {
+                            let s: SigStream = Box::new(stream.map(|_| Signal::Quit));
+                            s
+                        },
+                    ),
+                ));
+                futures_unordered(sigs)
+                    .collect()
+                    .map_err(|_| ())
+                    .and_then(move |streams| Signals { srv, streams })
+            }
+        };
+        spawn(fut);
+    }
+}
+
+impl Future for Signals {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        #[cfg(not(unix))]
+        loop {
+            match self.stream.poll() {
+                Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
+                Ok(Async::Ready(Some(sig))) => self.srv.signal(sig),
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+            }
+        }
+        #[cfg(unix)]
+        {
+            for s in &mut self.streams {
+                loop {
+                    match s.poll() {
+                        Ok(Async::Ready(None)) | Err(_) => return Ok(Async::Ready(())),
+                        Ok(Async::NotReady) => break,
+                        Ok(Async::Ready(Some(sig))) => self.srv.signal(sig),
+                    }
+                }
+            }
+            Ok(Async::NotReady)
+        }
+    }
+}

actix-server/src/ssl/mod.rs

@@ -1,12 +1,12 @@
 //! SSL Services
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use super::counter::Counter;
+use crate::counter::Counter;
 
 #[cfg(feature = "ssl")]
 mod openssl;
 #[cfg(feature = "ssl")]
-pub use self::openssl::{OpensslAcceptor, OpensslConnector};
+pub use self::openssl::OpensslAcceptor;
 
 #[cfg(feature = "tls")]
 mod nativetls;
@@ -28,7 +28,7 @@ pub fn max_concurrent_ssl_connect(num: usize) {
     MAX_CONN.store(num, Ordering::Relaxed);
 }
 
-pub(crate) const MAX_CONN: AtomicUsize = AtomicUsize::new(256);
+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

@@ -21,7 +21,7 @@ impl<T: AsyncRead + AsyncWrite> NativeTlsAcceptor<T> {
     /// Create `NativeTlsAcceptor` instance
     pub fn new(acceptor: TlsAcceptor) -> Self {
         NativeTlsAcceptor {
-            acceptor: acceptor.into(),
+            acceptor,
             io: PhantomData,
         }
     }

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/worker.rs

@@ -2,20 +2,17 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::{mem, net, time};
 
+use actix_rt::{spawn, Arbiter};
 use futures::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use futures::sync::oneshot;
 use futures::{future, Async, Future, Poll, Stream};
 use log::{error, info, trace};
-use tokio_current_thread::spawn;
 use tokio_timer::{sleep, Delay};
 
-use actix::msgs::StopArbiter;
+use crate::accept::AcceptNotify;
-use actix::{Arbiter, Message};
-
-use super::accept::AcceptNotify;
-use super::services::{BoxedServerService, InternalServiceFactory, ServerMessage};
-use super::Token;
 use crate::counter::Counter;
+use crate::services::{BoxedServerService, InternalServiceFactory, ServerMessage};
+use crate::Token;
 
 pub(crate) struct WorkerCommand(Conn);
 
@@ -26,14 +23,14 @@ pub(crate) struct StopCommand {
     result: oneshot::Sender<bool>,
 }
 
-#[derive(Debug, Message)]
+#[derive(Debug)]
 pub(crate) struct Conn {
     pub io: net::TcpStream,
     pub token: Token,
     pub peer: Option<net::SocketAddr>,
 }
 
-const MAX_CONNS: AtomicUsize = AtomicUsize::new(25600);
+static MAX_CONNS: AtomicUsize = AtomicUsize::new(25600);
 
 /// Sets the maximum per-worker number of concurrent connections.
 ///
@@ -167,7 +164,7 @@ impl Worker {
             future::join_all(fut)
                 .map_err(|e| {
                     error!("Can not start worker: {:?}", e);
-                    Arbiter::current().do_send(StopArbiter(0));
+                    Arbiter::current().stop();
                 })
                 .and_then(move |services| {
                     for item in services {
@@ -365,7 +362,7 @@ impl Future for Worker {
                 let num = num_connections();
                 if num == 0 {
                     let _ = tx.send(true);
-                    Arbiter::current().do_send(StopArbiter(0));
+                    Arbiter::current().stop();
                     return Ok(Async::Ready(()));
                 }
 
@@ -375,7 +372,7 @@ impl Future for Worker {
                     Async::Ready(_) => {
                         self.shutdown(true);
                         let _ = tx.send(false);
-                        Arbiter::current().do_send(StopArbiter(0));
+                        Arbiter::current().stop();
                         return Ok(Async::Ready(()));
                     }
                 }

actix-utils/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2018-12-09
+
+* Move utils services to separate crate

actix-utils/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "actix-utils"
+version = "0.1.0"
+authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+description = "Actix utils - various actix net related services"
+keywords = ["network", "framework", "async", "futures"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-utils/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+edition = "2018"
+workspace = "../"
+
+[lib]
+name = "actix_utils"
+path = "src/lib.rs"
+
+[dependencies]
+actix-service = "0.1.1"
+actix-codec = "0.1.0"
+actix-rt = "0.1.0"
+
+# io
+bytes = "0.4"
+futures = "0.1"
+tokio-timer = "0.2.8"

actix-utils/src/counter.rs

@@ -0,0 +1,78 @@
+use std::cell::Cell;
+use std::rc::Rc;
+
+use futures::task::AtomicTask;
+
+#[derive(Clone)]
+/// Simple counter with ability to notify task on reaching specific number
+///
+/// Counter could be cloned, total ncount is shared across all clones.
+pub struct Counter(Rc<CounterInner>);
+
+struct CounterInner {
+    count: Cell<usize>,
+    capacity: usize,
+    task: AtomicTask,
+}
+
+impl Counter {
+    /// Create `Counter` instance and set max value.
+    pub fn new(capacity: usize) -> Self {
+        Counter(Rc::new(CounterInner {
+            capacity,
+            count: Cell::new(0),
+            task: AtomicTask::new(),
+        }))
+    }
+
+    pub fn get(&self) -> CounterGuard {
+        CounterGuard::new(self.0.clone())
+    }
+
+    /// Check if counter is not at capacity
+    pub fn available(&self) -> bool {
+        self.0.available()
+    }
+
+    /// Get total number of acquired counts
+    pub fn total(&self) -> usize {
+        self.0.count.get()
+    }
+}
+
+pub struct CounterGuard(Rc<CounterInner>);
+
+impl CounterGuard {
+    fn new(inner: Rc<CounterInner>) -> Self {
+        inner.inc();
+        CounterGuard(inner)
+    }
+}
+
+impl Drop for CounterGuard {
+    fn drop(&mut self) {
+        self.0.dec();
+    }
+}
+
+impl CounterInner {
+    fn inc(&self) {
+        let num = self.count.get() + 1;
+        self.count.set(num);
+        if num == self.capacity {
+            self.task.register();
+        }
+    }
+
+    fn dec(&self) {
+        let num = self.count.get();
+        self.count.set(num - 1);
+        if num == self.capacity {
+            self.task.notify();
+        }
+    }
+
+    fn available(&self) -> bool {
+        self.count.get() < self.capacity
+    }
+}

actix-utils/src/framed.rs

@@ -2,15 +2,12 @@
 use std::marker::PhantomData;
 use std::mem;
 
-use actix;
+use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
+use actix_rt::Arbiter;
 use actix_service::{IntoNewService, IntoService, NewService, Service};
 use futures::future::{ok, FutureResult};
 use futures::unsync::mpsc;
 use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
-use tokio_codec::{Decoder, Encoder};
-use tokio_io::{AsyncRead, AsyncWrite};
-
-use crate::codec::Framed;
 
 type Request<U> = <U as Decoder>::Item;
 type Response<U> = <U as Encoder>::Item;
@@ -257,7 +254,7 @@ where
             Ok(Async::Ready(_)) => {
                 if let Some(item) = self.request.take() {
                     let sender = self.write_tx.clone();
-                    actix::Arbiter::spawn(
+                    Arbiter::spawn(
                         self.service
                             .call(item)
                             .then(|item| sender.send(item).map(|_| ()).map_err(|_| ())),
@@ -282,7 +279,7 @@ where
                     match self.service.poll_ready() {
                         Ok(Async::Ready(_)) => {
                             let sender = self.write_tx.clone();
-                            actix::Arbiter::spawn(
+                            Arbiter::spawn(
                                 self.service
                                     .call(item)
                                     .then(|item| sender.send(item).map(|_| ()).map_err(|_| ())),

actix-utils/src/lib.rs

@@ -0,0 +1,14 @@
+//! Actix utils - various helper services
+mod cell;
+pub mod cloneable;
+pub mod counter;
+pub mod either;
+pub mod framed;
+pub mod inflight;
+pub mod keepalive;
+pub mod stream;
+pub mod time;
+pub mod timeout;
+
+#[derive(Copy, Clone, Debug)]
+pub enum Never {}

actix-utils/src/stream.rs

@@ -1,9 +1,9 @@
 use std::marker::PhantomData;
 
+use actix_rt::spawn;
 use actix_service::{IntoService, NewService, Service};
 use futures::unsync::mpsc;
 use futures::{future, Async, Future, Poll, Stream};
-use tokio_current_thread::spawn;
 
 pub struct StreamDispatcher<S: Stream, T> {
     stream: S,

actix-utils/src/time.rs

@@ -1,9 +1,9 @@
 use std::time::{Duration, Instant};
 
+use actix_rt::spawn;
 use actix_service::{NewService, Service};
 use futures::future::{ok, FutureResult};
 use futures::{Async, Future, Poll};
-use tokio_current_thread::spawn;
 use tokio_timer::sleep;
 
 use super::cell::Cell;

examples/basic.rs

@@ -1,29 +1,20 @@
 //! simple composite service
 //! build: cargo run --example basic --features "ssl"
 //! to test: curl https://127.0.0.1:8443/ -k
-extern crate actix;
-extern crate actix_net;
-extern crate env_logger;
-extern crate futures;
-extern crate openssl;
-extern crate tokio_io;
-extern crate tokio_openssl;
-extern crate tokio_tcp;
-
 use std::sync::{
     atomic::{AtomicUsize, Ordering},
     Arc,
 };
 use std::{env, fmt};
 
+use actix_codec::{AsyncRead, AsyncWrite};
+use actix_rt::System;
+use actix_server::Server;
+use actix_service::{IntoNewService, NewService};
 use futures::{future, Future};
 use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
-use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_openssl::SslAcceptorExt;
 
-use actix_net::server::Server;
-use actix_net::service::{IntoNewService, NewServiceExt};
-
 /// Simple logger service, it just prints fact of the new connections
 fn logger<T: AsyncRead + AsyncWrite + fmt::Debug>(
     stream: T,
@@ -36,7 +27,7 @@ fn main() {
     env::set_var("RUST_LOG", "actix_net=trace");
     env_logger::init();
 
-    let sys = actix::System::new("test");
+    let sys = System::new("test");
 
     // load ssl keys
     let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
@@ -53,7 +44,7 @@ fn main() {
     // bind socket address and start workers. By default server uses number of
     // available logical cpu as threads count. actix net start separate
     // instances of service pipeline in each worker.
-    Server::default()
+    Server::build()
         .bind(
             // configure service pipeline
             "basic",

examples/ssl.rs

@@ -1,22 +1,14 @@
-extern crate actix;
-extern crate actix_net;
-extern crate futures;
-extern crate openssl;
-extern crate tokio_io;
-extern crate tokio_tcp;
-
 use std::sync::{
     atomic::{AtomicUsize, Ordering},
     Arc,
 };
 
+use actix_codec::{AsyncRead, AsyncWrite};
+use actix_rt::System;
+use actix_server::{ssl, Server};
+use actix_service::NewService;
 use futures::{future, Future};
 use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
-use tokio_io::{AsyncRead, AsyncWrite};
-
-use actix_net::server::Server;
-use actix_net::service::NewServiceExt;
-use actix_net::ssl;
 
 #[derive(Debug)]
 struct ServiceState {
@@ -33,7 +25,7 @@ fn service<T: AsyncRead + AsyncWrite>(
 }
 
 fn main() {
-    let sys = actix::System::new("test");
+    let sys = System::new("test");
 
     // load ssl keys
     let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
@@ -48,7 +40,7 @@ fn main() {
     let openssl = ssl::OpensslAcceptor::new(builder.build());
 
     // server start mutiple workers, it runs supplied `Fn` in each worker.
-    Server::default()
+    Server::build()
         .bind("test-ssl", "0.0.0.0:8443", move || {
             let num = num.clone();
 

src/codec/framed2.rs

@@ -1,313 +0,0 @@
-#![allow(deprecated)]
-
-use std::fmt;
-use std::io::{self, Read, Write};
-
-use bytes::BytesMut;
-use futures::{Poll, Sink, StartSend, Stream};
-use tokio_codec::{Decoder, Encoder};
-use tokio_io::{AsyncRead, AsyncWrite};
-
-use super::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2};
-use super::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2};
-
-/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
-/// the `Encoder` and `Decoder` traits to encode and decode frames.
-///
-/// You can create a `Framed` instance by using the `AsyncRead::framed` adapter.
-pub struct Framed2<T, D, E> {
-    inner: FramedRead2<FramedWrite2<Fuse2<T, D, E>>>,
-}
-
-pub struct Fuse2<T, D, E>(pub T, pub D, pub E);
-
-impl<T, D, E> Framed2<T, D, E>
-where
-    T: AsyncRead + AsyncWrite,
-    D: Decoder,
-    E: Encoder,
-{
-    /// Provides a `Stream` and `Sink` interface for reading and writing to this
-    /// `Io` object, using `Decode` and `Encode` to read and write the raw data.
-    ///
-    /// Raw I/O objects work with byte sequences, but higher-level code usually
-    /// wants to batch these into meaningful chunks, called "frames". This
-    /// method layers framing on top of an I/O object, by using the `Codec`
-    /// traits to handle encoding and decoding of messages frames. Note that
-    /// the incoming and outgoing frame types may be distinct.
-    ///
-    /// This function returns a *single* object that is both `Stream` and
-    /// `Sink`; grouping this into a single object is often useful for layering
-    /// things like gzip or TLS, which require both read and write access to the
-    /// underlying object.
-    ///
-    /// If you want to work more directly with the streams and sink, consider
-    /// calling `split` on the `Framed` returned by this method, which will
-    /// break them into separate objects, allowing them to interact more easily.
-    pub fn new(inner: T, decoder: D, encoder: E) -> Framed2<T, D, E> {
-        Framed2 {
-            inner: framed_read2(framed_write2(Fuse2(inner, decoder, encoder))),
-        }
-    }
-}
-
-impl<T, D, E> Framed2<T, D, E> {
-    /// Provides a `Stream` and `Sink` interface for reading and writing to this
-    /// `Io` object, using `Decode` and `Encode` to read and write the raw data.
-    ///
-    /// Raw I/O objects work with byte sequences, but higher-level code usually
-    /// wants to batch these into meaningful chunks, called "frames". This
-    /// method layers framing on top of an I/O object, by using the `Codec`
-    /// traits to handle encoding and decoding of messages frames. Note that
-    /// the incoming and outgoing frame types may be distinct.
-    ///
-    /// This function returns a *single* object that is both `Stream` and
-    /// `Sink`; grouping this into a single object is often useful for layering
-    /// things like gzip or TLS, which require both read and write access to the
-    /// underlying object.
-    ///
-    /// This objects takes a stream and a readbuffer and a writebuffer. These
-    /// field can be obtained from an existing `Framed` with the
-    /// `into_parts` method.
-    ///
-    /// If you want to work more directly with the streams and sink, consider
-    /// calling `split` on the `Framed` returned by this method, which will
-    /// break them into separate objects, allowing them to interact more easily.
-    pub fn from_parts(parts: FramedParts2<T, D, E>) -> Framed2<T, D, E> {
-        Framed2 {
-            inner: framed_read2_with_buffer(
-                framed_write2_with_buffer(
-                    Fuse2(parts.io, parts.decoder, parts.encoder),
-                    parts.write_buf,
-                ),
-                parts.read_buf,
-            ),
-        }
-    }
-
-    /// Returns a reference to the underlying I/O stream wrapped by
-    /// `Frame`.
-    ///
-    /// Note that care should be taken to not tamper with the underlying stream
-    /// of data coming in as it may corrupt the stream of frames otherwise
-    /// being worked with.
-    pub fn get_ref(&self) -> &T {
-        &self.inner.get_ref().get_ref().0
-    }
-
-    /// Returns a mutable reference to the underlying I/O stream wrapped by
-    /// `Frame`.
-    ///
-    /// Note that care should be taken to not tamper with the underlying stream
-    /// of data coming in as it may corrupt the stream of frames otherwise
-    /// being worked with.
-    pub fn get_mut(&mut self) -> &mut T {
-        &mut self.inner.get_mut().get_mut().0
-    }
-
-    /// Returns a reference to the underlying decoder.
-    pub fn decocer(&self) -> &D {
-        &self.inner.get_ref().get_ref().1
-    }
-
-    /// Returns a mutable reference to the underlying decoder.
-    pub fn decoder_mut(&mut self) -> &mut D {
-        &mut self.inner.get_mut().get_mut().1
-    }
-
-    /// Returns a reference to the underlying encoder.
-    pub fn encoder(&self) -> &E {
-        &self.inner.get_ref().get_ref().2
-    }
-
-    /// Returns a mutable reference to the underlying codec.
-    pub fn encoder_mut(&mut self) -> &mut E {
-        &mut self.inner.get_mut().get_mut().2
-    }
-
-    /// Consumes the `Frame`, returning its underlying I/O stream.
-    ///
-    /// Note that care should be taken to not tamper with the underlying stream
-    /// of data coming in as it may corrupt the stream of frames otherwise
-    /// being worked with.
-    pub fn into_inner(self) -> T {
-        self.inner.into_inner().into_inner().0
-    }
-
-    /// Consume the `Frame`, returning `Frame` with different codec.
-    pub fn switch_encoder<E2>(self, encoder: E2) -> Framed2<T, D, E2> {
-        let (inner, read_buf) = self.inner.into_parts();
-        let (inner, write_buf) = inner.into_parts();
-
-        Framed2 {
-            inner: framed_read2_with_buffer(
-                framed_write2_with_buffer(Fuse2(inner.0, inner.1, encoder), write_buf),
-                read_buf,
-            ),
-        }
-    }
-
-    /// Consumes the `Frame`, returning its underlying I/O stream, the buffer
-    /// with unprocessed data, and the codec.
-    ///
-    /// Note that care should be taken to not tamper with the underlying stream
-    /// of data coming in as it may corrupt the stream of frames otherwise
-    /// being worked with.
-    pub fn into_parts(self) -> FramedParts2<T, D, E> {
-        let (inner, read_buf) = self.inner.into_parts();
-        let (inner, write_buf) = inner.into_parts();
-
-        FramedParts2 {
-            io: inner.0,
-            decoder: inner.1,
-            encoder: inner.2,
-            read_buf: read_buf,
-            write_buf: write_buf,
-            _priv: (),
-        }
-    }
-}
-
-impl<T, D, E> Stream for Framed2<T, D, E>
-where
-    T: AsyncRead,
-    D: Decoder,
-{
-    type Item = D::Item;
-    type Error = D::Error;
-
-    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-        self.inner.poll()
-    }
-}
-
-impl<T, D, E> Sink for Framed2<T, D, E>
-where
-    T: AsyncWrite,
-    E: Encoder,
-    E::Error: From<io::Error>,
-{
-    type SinkItem = E::Item;
-    type SinkError = E::Error;
-
-    fn start_send(
-        &mut self,
-        item: Self::SinkItem,
-    ) -> StartSend<Self::SinkItem, Self::SinkError> {
-        self.inner.get_mut().start_send(item)
-    }
-
-    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
-        self.inner.get_mut().poll_complete()
-    }
-
-    fn close(&mut self) -> Poll<(), Self::SinkError> {
-        self.inner.get_mut().close()
-    }
-}
-
-impl<T, D, E> fmt::Debug for Framed2<T, D, E>
-where
-    T: fmt::Debug,
-    D: fmt::Debug,
-    E: fmt::Debug,
-{
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Framed2")
-            .field("io", &self.inner.get_ref().get_ref().0)
-            .field("decoder", &self.inner.get_ref().get_ref().1)
-            .field("encoder", &self.inner.get_ref().get_ref().2)
-            .finish()
-    }
-}
-
-// ===== impl Fuse2 =====
-
-impl<T: Read, D, E> Read for Fuse2<T, D, E> {
-    fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
-        self.0.read(dst)
-    }
-}
-
-impl<T: AsyncRead, D, E> AsyncRead for Fuse2<T, D, E> {
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
-        self.0.prepare_uninitialized_buffer(buf)
-    }
-}
-
-impl<T: Write, D, E> Write for Fuse2<T, D, E> {
-    fn write(&mut self, src: &[u8]) -> io::Result<usize> {
-        self.0.write(src)
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        self.0.flush()
-    }
-}
-
-impl<T: AsyncWrite, D, E> AsyncWrite for Fuse2<T, D, E> {
-    fn shutdown(&mut self) -> Poll<(), io::Error> {
-        self.0.shutdown()
-    }
-}
-
-impl<T, D: Decoder, E> Decoder for Fuse2<T, D, E> {
-    type Item = D::Item;
-    type Error = D::Error;
-
-    fn decode(&mut self, buffer: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        self.1.decode(buffer)
-    }
-
-    fn decode_eof(&mut self, buffer: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        self.1.decode_eof(buffer)
-    }
-}
-
-impl<T, D, E: Encoder> Encoder for Fuse2<T, D, E> {
-    type Item = E::Item;
-    type Error = E::Error;
-
-    fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
-        self.2.encode(item, dst)
-    }
-}
-
-/// `FramedParts` contains an export of the data of a Framed transport.
-/// It can be used to construct a new `Framed` with a different codec.
-/// It contains all current buffers and the inner transport.
-#[derive(Debug)]
-pub struct FramedParts2<T, D, E> {
-    /// The inner transport used to read bytes to and write bytes to
-    pub io: T,
-
-    /// The decoder
-    pub decoder: D,
-
-    /// The encoder
-    pub encoder: E,
-
-    /// The buffer with read but unprocessed data.
-    pub read_buf: BytesMut,
-
-    /// A buffer with unprocessed data which are not written yet.
-    pub write_buf: BytesMut,
-
-    /// This private field allows us to add additional fields in the future in a
-    /// backwards compatible way.
-    _priv: (),
-}
-
-impl<T, D, E> FramedParts2<T, D, E> {
-    /// Create a new, default, `FramedParts`
-    pub fn new(io: T, decoder: D, encoder: E) -> FramedParts2<T, D, E> {
-        FramedParts2 {
-            io,
-            decoder,
-            encoder,
-            read_buf: BytesMut::new(),
-            write_buf: BytesMut::new(),
-            _priv: (),
-        }
-    }
-}

src/lib.rs

@@ -1,32 +0,0 @@
-//! Actix net - framework for the compisible network services for Rust.
-//!
-//! ## Package feature
-//!
-//! * `tls` - enables ssl support via `native-tls` crate
-//! * `ssl` - enables ssl support via `openssl` crate
-//! * `rust-tls` - enables ssl support via `rustls` crate
-// #![warn(missing_docs)]
-
-#![allow(
-    clippy::declare_interior_mutable_const,
-    clippy::borrow_interior_mutable_const
-)]
-
-mod cell;
-pub mod cloneable;
-pub mod codec;
-pub mod connector;
-pub mod counter;
-pub mod either;
-pub mod framed;
-pub mod inflight;
-pub mod keepalive;
-pub mod resolver;
-pub mod server;
-pub mod ssl;
-pub mod stream;
-pub mod time;
-pub mod timeout;
-
-#[derive(Copy, Clone, Debug)]
-pub enum Never {}

src/server/mod.rs

@@ -1,48 +0,0 @@
-//! General purpose networking server
-
-use actix::Message;
-
-mod accept;
-mod config;
-mod server;
-mod services;
-mod worker;
-
-pub use self::config::{ServiceConfig, ServiceRuntime};
-pub use self::server::Server;
-pub use self::services::{ServerMessage, ServiceFactory, StreamServiceFactory};
-
-/// Pause accepting incoming connections
-///
-/// If socket contains some pending connection, they might be dropped.
-/// All opened connection remains active.
-#[derive(Message)]
-pub struct PauseServer;
-
-/// Resume accepting incoming connections
-#[derive(Message)]
-pub struct ResumeServer;
-
-/// Stop incoming connection processing, stop all workers and exit.
-///
-/// If server starts with `spawn()` method, then spawned thread get terminated.
-pub struct StopServer {
-    /// Whether to try and shut down gracefully
-    pub graceful: bool,
-}
-
-impl Message for StopServer {
-    type Result = Result<(), ()>;
-}
-
-/// Socket id token
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub(crate) struct Token(usize);
-
-impl Token {
-    pub(crate) fn next(&mut self) -> Token {
-        let token = Token(self.0 + 1);
-        self.0 += 1;
-        token
-    }
-}