Framed::is_write_buf_empty() checks if write buffer is flushed

CHANGES.md

@@ -1,5 +1,54 @@
 # Changes
 
+## [0.2.3] - 2018-11-17
+
+### Added
+
+* Framed::is_write_buf_empty() checks if write buffer is flushed
+
+## [0.2.2] - 2018-11-14
+
+### Added
+
+* Add low/high caps to Framed
+
+### Changed
+
+* Refactor Connector and Resolver services
+
+
+### Fixed
+
+* Fix wrong service to socket binding
+
+
+## [0.2.0] - 2018-11-08
+
+### Added
+
+* Timeout service
+
+* Added ServiceConfig and ServiceRuntime for server service configuration
+
+
+### Changed
+
+* Connector has been refactored
+
+* timer and LowResTimer renamed to time and LowResTime
+
+* Refactored `Server::configure()` method
+
+
+## [0.1.1] - 2018-10-10
+
+### Changed
+
+- Set actix min version - 0.7.5
+
+- Set trust-dns min version
+
+
 ## [0.1.0] - 2018-10-08
 
 * Initial impl

Cargo.toml

@@ -1,13 +1,13 @@
 [package]
 name = "actix-net"
-version = "0.1.0"
+version = "0.2.3"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix net - framework for the compisible network services for Rust (experimental)"
 readme = "README.md"
 keywords = ["network", "framework", "async", "futures"]
 homepage = "https://actix.rs"
 repository = "https://github.com/actix/actix-net.git"
-documentation = "https://actix.rs/api/actix-net/stable/actix_net/"
+documentation = "https://docs.rs/actix-net/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
@@ -39,7 +39,7 @@ rust-tls = ["rustls", "tokio-rustls", "webpki", "webpki-roots"]
 cell = []
 
 [dependencies]
-actix = "0.7.0"
+actix = "0.7.6"
 
 log = "0.4"
 num_cpus = "1.0"
@@ -58,7 +58,8 @@ tokio-timer = "0.2"
 tokio-reactor = "0.1"
 tokio-current-thread = "0.1"
 tower-service = "0.1"
-trust-dns-resolver = "0.10.0-alpha.2"
+trust-dns-proto = "^0.5.0"
+trust-dns-resolver = "^0.10.0"
 
 # native-tls
 native-tls = { version="0.2", optional = true }

examples/ssl.rs

@@ -24,7 +24,8 @@ struct ServiceState {
 }
 
 fn service<T: AsyncRead + AsyncWrite>(
-    st: &mut ServiceState, _: T,
+    st: &mut ServiceState,
+    _: T,
 ) -> impl Future<Item = (), Error = ()> {
     let num = st.num.fetch_add(1, Ordering::Relaxed);
     println!("got ssl connection {:?}", num);

src/codec/bcodec.rs

@@ -0,0 +1,33 @@
+use std::io;
+
+use bytes::{Bytes, BytesMut};
+use tokio_codec::{Decoder, Encoder};
+
+/// Bytes codec.
+///
+/// Reads/Writes chunks of bytes from a stream.
+#[derive(Debug, Copy, Clone)]
+pub struct BytesCodec;
+
+impl Encoder for BytesCodec {
+    type Item = Bytes;
+    type Error = io::Error;
+
+    fn encode(&mut self, item: Bytes, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        dst.extend_from_slice(&item[..]);
+        Ok(())
+    }
+}
+
+impl Decoder for BytesCodec {
+    type Item = BytesMut;
+    type Error = io::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if src.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(src.take()))
+        }
+    }
+}

src/codec/framed.rs

@@ -11,6 +11,9 @@ 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};
 
+const LW: usize = 1024;
+const HW: usize = 8 * 1024;
+
 /// A unified `Stream` and `Sink` interface to an underlying I/O object, using
 /// the `Encoder` and `Decoder` traits to encode and decode frames.
 ///
@@ -45,9 +48,25 @@ where
     /// break them into separate objects, allowing them to interact more easily.
     pub fn new(inner: T, codec: U) -> Framed<T, U> {
         Framed {
-            inner: framed_read2(framed_write2(Fuse(inner, codec))),
+            inner: framed_read2(framed_write2(Fuse(inner, codec), LW, HW)),
         }
     }
+
+    /// Same as `Framed::new()` with ability to specify write buffer low/high capacity watermarks.
+    pub fn new_with_caps(inner: T, codec: U, lw: usize, hw: usize) -> Framed<T, U> {
+        debug_assert!((lw < hw) && hw != 0);
+        Framed {
+            inner: framed_read2(framed_write2(Fuse(inner, codec), lw, hw)),
+        }
+    }
+
+    /// Force send item
+    pub fn force_send(
+        &mut self,
+        item: <U as Encoder>::Item,
+    ) -> Result<(), <U as Encoder>::Error> {
+        self.inner.get_mut().force_send(item)
+    }
 }
 
 impl<T, U> Framed<T, U> {
@@ -75,7 +94,12 @@ impl<T, U> Framed<T, U> {
     pub fn from_parts(parts: FramedParts<T, U>) -> Framed<T, U> {
         Framed {
             inner: framed_read2_with_buffer(
-                framed_write2_with_buffer(Fuse(parts.io, parts.codec), parts.write_buf),
+                framed_write2_with_buffer(
+                    Fuse(parts.io, parts.codec),
+                    parts.write_buf,
+                    parts.write_buf_lw,
+                    parts.write_buf_hw,
+                ),
                 parts.read_buf,
             ),
         }
@@ -111,6 +135,16 @@ impl<T, U> Framed<T, U> {
         &mut self.inner.get_mut().get_mut().0
     }
 
+    /// Check if write buffer is empty.
+    pub fn is_write_buf_empty(&self) -> bool {
+        self.inner.get_ref().is_empty()
+    }
+
+    /// Check if write buffer is full.
+    pub fn is_write_buf_full(&self) -> bool {
+        self.inner.get_ref().is_full()
+    }
+
     /// Consumes the `Frame`, returning its underlying I/O stream.
     ///
     /// Note that care should be taken to not tamper with the underlying stream
@@ -123,11 +157,27 @@ impl<T, U> Framed<T, U> {
     /// Consume the `Frame`, returning `Frame` with different codec.
     pub fn into_framed<U2>(self, codec: U2) -> Framed<T, U2> {
         let (inner, read_buf) = self.inner.into_parts();
-        let (inner, write_buf) = inner.into_parts();
+        let (inner, write_buf, lw, hw) = inner.into_parts();
 
         Framed {
             inner: framed_read2_with_buffer(
-                framed_write2_with_buffer(Fuse(inner.0, codec), write_buf),
+                framed_write2_with_buffer(Fuse(inner.0, codec), write_buf, lw, hw),
+                read_buf,
+            ),
+        }
+    }
+
+    /// Consume the `Frame`, returning `Frame` with different codec.
+    pub fn map_codec<F, U2>(self, f: F) -> Framed<T, U2>
+    where
+        F: Fn(U) -> U2,
+    {
+        let (inner, read_buf) = self.inner.into_parts();
+        let (inner, write_buf, lw, hw) = inner.into_parts();
+
+        Framed {
+            inner: framed_read2_with_buffer(
+                framed_write2_with_buffer(Fuse(inner.0, f(inner.1)), write_buf, lw, hw),
                 read_buf,
             ),
         }
@@ -141,13 +191,15 @@ impl<T, U> Framed<T, U> {
     /// being worked with.
     pub fn into_parts(self) -> FramedParts<T, U> {
         let (inner, read_buf) = self.inner.into_parts();
-        let (inner, write_buf) = inner.into_parts();
+        let (inner, write_buf, write_buf_lw, write_buf_hw) = inner.into_parts();
 
         FramedParts {
             io: inner.0,
             codec: inner.1,
-            read_buf: read_buf,
+            read_buf,
-            write_buf: write_buf,
+            write_buf,
+            write_buf_lw,
+            write_buf_hw,
             _priv: (),
         }
     }
@@ -176,7 +228,8 @@ where
     type SinkError = U::Error;
 
     fn start_send(
-        &mut self, item: Self::SinkItem,
+        &mut self,
+        item: Self::SinkItem,
     ) -> StartSend<Self::SinkItem, Self::SinkError> {
         self.inner.get_mut().start_send(item)
     }
@@ -272,6 +325,12 @@ pub struct FramedParts<T, U> {
     /// A buffer with unprocessed data which are not written yet.
     pub write_buf: BytesMut,
 
+    /// A buffer low watermark capacity
+    pub write_buf_lw: usize,
+
+    /// A buffer high watermark capacity
+    pub write_buf_hw: usize,
+
     /// This private field allows us to add additional fields in the future in a
     /// backwards compatible way.
     _priv: (),
@@ -285,6 +344,8 @@ impl<T, U> FramedParts<T, U> {
             codec,
             read_buf: BytesMut::new(),
             write_buf: BytesMut::new(),
+            write_buf_lw: LW,
+            write_buf_hw: HW,
             _priv: (),
         }
     }

src/codec/framed2.rs

@@ -191,7 +191,8 @@ where
     type SinkError = E::Error;
 
     fn start_send(
-        &mut self, item: Self::SinkItem,
+        &mut self,
+        item: Self::SinkItem,
     ) -> StartSend<Self::SinkItem, Self::SinkError> {
         self.inner.get_mut().start_send(item)
     }

src/codec/framed_read.rs

@@ -98,7 +98,8 @@ where
     type SinkError = T::SinkError;
 
     fn start_send(
-        &mut self, item: Self::SinkItem,
+        &mut self,
+        item: Self::SinkItem,
     ) -> StartSend<Self::SinkItem, Self::SinkError> {
         self.inner.inner.0.start_send(item)
     }

src/codec/framed_write.rs

@@ -16,20 +16,19 @@ pub struct FramedWrite<T, E> {
 pub struct FramedWrite2<T> {
     inner: T,
     buffer: BytesMut,
+    low_watermark: usize,
+    high_watermark: usize,
 }
 
-const INITIAL_CAPACITY: usize = 8 * 1024;
-const BACKPRESSURE_BOUNDARY: usize = INITIAL_CAPACITY;
-
 impl<T, E> FramedWrite<T, E>
 where
     T: AsyncWrite,
     E: Encoder,
 {
     /// Creates a new `FramedWrite` with the given `encoder`.
-    pub fn new(inner: T, encoder: E) -> FramedWrite<T, E> {
+    pub fn new(inner: T, encoder: E, lw: usize, hw: usize) -> FramedWrite<T, E> {
         FramedWrite {
-            inner: framed_write2(Fuse(inner, encoder)),
+            inner: framed_write2(Fuse(inner, encoder), lw, hw),
         }
     }
 }
@@ -73,6 +72,26 @@ impl<T, E> FramedWrite<T, E> {
     pub fn encoder_mut(&mut self) -> &mut E {
         &mut self.inner.inner.1
     }
+
+    /// Check if write buffer is full
+    pub fn is_full(&self) -> bool {
+        self.inner.is_full()
+    }
+
+    /// Check if write buffer is empty.
+    pub fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+}
+
+impl<T, E> FramedWrite<T, E>
+where
+    E: Encoder,
+{
+    /// Force send item
+    pub fn force_send(&mut self, item: E::Item) -> Result<(), E::Error> {
+        self.inner.force_send(item)
+    }
 }
 
 impl<T, E> Sink for FramedWrite<T, E>
@@ -124,21 +143,34 @@ where
 
 // ===== impl FramedWrite2 =====
 
-pub fn framed_write2<T>(inner: T) -> FramedWrite2<T> {
+pub fn framed_write2<T>(
+    inner: T,
+    low_watermark: usize,
+    high_watermark: usize,
+) -> FramedWrite2<T> {
     FramedWrite2 {
-        inner: inner,
+        inner,
-        buffer: BytesMut::with_capacity(INITIAL_CAPACITY),
+        low_watermark,
+        high_watermark,
+        buffer: BytesMut::with_capacity(high_watermark),
     }
 }
 
-pub fn framed_write2_with_buffer<T>(inner: T, mut buf: BytesMut) -> FramedWrite2<T> {
+pub fn framed_write2_with_buffer<T>(
-    if buf.capacity() < INITIAL_CAPACITY {
+    inner: T,
-        let bytes_to_reserve = INITIAL_CAPACITY - buf.capacity();
+    mut buffer: BytesMut,
-        buf.reserve(bytes_to_reserve);
+    low_watermark: usize,
+    high_watermark: usize,
+) -> FramedWrite2<T> {
+    if buffer.capacity() < high_watermark {
+        let bytes_to_reserve = high_watermark - buffer.capacity();
+        buffer.reserve(bytes_to_reserve);
     }
     FramedWrite2 {
-        inner: inner,
+        inner,
-        buffer: buf,
+        buffer,
+        low_watermark,
+        high_watermark,
     }
 }
 
@@ -151,13 +183,40 @@ impl<T> FramedWrite2<T> {
         self.inner
     }
 
-    pub fn into_parts(self) -> (T, BytesMut) {
+    pub fn into_parts(self) -> (T, BytesMut, usize, usize) {
-        (self.inner, self.buffer)
+        (
+            self.inner,
+            self.buffer,
+            self.low_watermark,
+            self.high_watermark,
+        )
     }
 
     pub fn get_mut(&mut self) -> &mut T {
         &mut self.inner
     }
+
+    pub fn is_full(&self) -> bool {
+        self.buffer.len() >= self.high_watermark
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.buffer.is_empty()
+    }
+}
+
+impl<T> FramedWrite2<T>
+where
+    T: Encoder,
+{
+    pub fn force_send(&mut self, item: T::Item) -> Result<(), T::Error> {
+        let len = self.buffer.len();
+        if len < self.low_watermark {
+            self.buffer.reserve(self.high_watermark - len)
+        }
+        self.inner.encode(item, &mut self.buffer)?;
+        Ok(())
+    }
 }
 
 impl<T> Sink for FramedWrite2<T>
@@ -168,18 +227,16 @@ where
     type SinkError = T::Error;
 
     fn start_send(&mut self, item: T::Item) -> StartSend<T::Item, T::Error> {
-        // If the buffer is already over 8KiB, then attempt to flush it. If after
+        // Check the buffer capacity
-        // flushing it's *still* over 8KiB, then apply backpressure (reject the
+        let len = self.buffer.len();
-        // send).
+        if len >= self.high_watermark {
-        if self.buffer.len() >= BACKPRESSURE_BOUNDARY {
+            return Ok(AsyncSink::NotReady(item));
-            try!(self.poll_complete());
+        }
-
+        if len < self.low_watermark {
-            if self.buffer.len() >= BACKPRESSURE_BOUNDARY {
+            self.buffer.reserve(self.high_watermark - len)
-                return Ok(AsyncSink::NotReady(item));
-            }
         }
 
-        try!(self.inner.encode(item, &mut self.buffer));
+        self.inner.encode(item, &mut self.buffer)?;
 
         Ok(AsyncSink::Ready)
     }

src/codec/mod.rs

@@ -12,12 +12,14 @@
 
 #![deny(missing_docs, missing_debug_implementations, warnings)]
 
+mod bcodec;
 mod framed;
-mod framed2;
+// 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::framed2::{Framed2, FramedParts2};
 pub use self::framed_read::FramedRead;
 pub use self::framed_write::FramedWrite;

src/connector.rs

@@ -1,6 +1,8 @@
 use std::collections::VecDeque;
-use std::io;
+use std::marker::PhantomData;
-use std::net::SocketAddr;
+use std::net::{IpAddr, SocketAddr};
+use std::time::Duration;
+use std::{fmt, io};
 
 use futures::{
     future::{ok, FutureResult},
@@ -10,41 +12,115 @@ use tokio_tcp::{ConnectFuture, TcpStream};
 use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
 use trust_dns_resolver::system_conf::read_system_conf;
 
-use super::resolver::{HostAware, Resolver, ResolverError, ResolverFuture};
+use super::resolver::{RequestHost, ResolveError, Resolver, ResolverFuture};
 use super::service::{NewService, Service};
 
+/// Port of the request
+pub trait RequestPort {
+    fn port(&self) -> u16;
+}
+
 // #[derive(Fail, Debug)]
 #[derive(Debug)]
 pub enum ConnectorError {
     /// Failed to resolve the hostname
     // #[fail(display = "Failed resolving hostname: {}", _0)]
-    Resolver(ResolverError),
+    Resolver(ResolveError),
 
-    /// Not dns records
+    /// No dns records
-    // #[fail(display = "Invalid input: {}", _0)]
+    // #[fail(display = "No dns records found for the input")]
     NoRecords,
 
+    /// Connecting took too long
+    // #[fail(display = "Timeout out while establishing connection")]
+    Timeout,
+
+    /// Invalid input
+    InvalidInput,
+
     /// Connection io error
     // #[fail(display = "{}", _0)]
     IoError(io::Error),
 }
 
-impl From<ResolverError> for ConnectorError {
+impl From<ResolveError> for ConnectorError {
-    fn from(err: ResolverError) -> Self {
+    fn from(err: ResolveError) -> Self {
         ConnectorError::Resolver(err)
     }
 }
 
-pub struct ConnectionInfo {
+impl From<io::Error> for ConnectorError {
+    fn from(err: io::Error) -> Self {
+        ConnectorError::IoError(err)
+    }
+}
+
+/// Connect request
+#[derive(Eq, PartialEq, Debug, Hash)]
+pub struct Connect {
     pub host: String,
-    pub addr: SocketAddr,
+    pub port: u16,
+    pub timeout: Duration,
 }
 
-pub struct Connector<T = String> {
+impl Connect {
-    resolver: Resolver<T>,
+    /// Create new `Connect` instance.
+    pub fn new<T: AsRef<str>>(host: T, port: u16) -> Connect {
+        Connect {
+            port,
+            host: host.as_ref().to_owned(),
+            timeout: Duration::from_secs(1),
+        }
+    }
+
+    /// Create `Connect` instance by spliting the string by ':' and convert the second part to u16
+    pub fn with<T: AsRef<str>>(host: T) -> Result<Connect, ConnectorError> {
+        let mut parts_iter = host.as_ref().splitn(2, ':');
+        let host = parts_iter.next().ok_or(ConnectorError::InvalidInput)?;
+        let port_str = parts_iter.next().unwrap_or("");
+        let port = port_str
+            .parse::<u16>()
+            .map_err(|_| ConnectorError::InvalidInput)?;
+        Ok(Connect {
+            port,
+            host: host.to_owned(),
+            timeout: Duration::from_secs(1),
+        })
+    }
+
+    /// Set connect timeout
+    ///
+    /// By default timeout is set to a 1 second.
+    pub fn timeout(mut self, timeout: Duration) -> Connect {
+        self.timeout = timeout;
+        self
+    }
 }
 
-impl<T: HostAware> Default for Connector<T> {
+impl RequestHost for Connect {
+    fn host(&self) -> &str {
+        &self.host
+    }
+}
+
+impl RequestPort for Connect {
+    fn port(&self) -> u16 {
+        self.port
+    }
+}
+
+impl fmt::Display for Connect {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{}:{}", self.host, self.port)
+    }
+}
+
+/// Tcp connector
+pub struct Connector {
+    resolver: Resolver<Connect>,
+}
+
+impl Default for Connector {
     fn default() -> Self {
         let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
             (cfg, opts)
@@ -56,51 +132,37 @@ impl<T: HostAware> Default for Connector<T> {
     }
 }
 
-impl<T: HostAware> Connector<T> {
+impl Connector {
+    /// Create new connector with resolver configuration
     pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
         Connector {
             resolver: Resolver::new(cfg, opts),
         }
     }
 
+    /// Create new connector with custom resolver
     pub fn with_resolver(
-        resolver: Resolver<T>,
+        resolver: Resolver<Connect>,
-    ) -> impl Service<
+    ) -> impl Service<Request = Connect, Response = (Connect, TcpStream), Error = ConnectorError>
-        Request = T,
+                 + Clone {
-        Response = (T, ConnectionInfo, TcpStream),
-        Error = ConnectorError,
-    > + Clone {
         Connector { resolver }
     }
 
-    pub fn new_service<E>() -> impl NewService<
+    /// Create new default connector service
-        Request = T,
-        Response = (T, ConnectionInfo, TcpStream),
-        Error = ConnectorError,
-        InitError = E,
-    > + Clone {
-        || -> FutureResult<Connector<T>, E> { ok(Connector::default()) }
-    }
-
     pub fn new_service_with_config<E>(
-        cfg: ResolverConfig, opts: ResolverOpts,
+        cfg: ResolverConfig,
+        opts: ResolverOpts,
     ) -> impl NewService<
-        Request = T,
+        Request = Connect,
-        Response = (T, ConnectionInfo, TcpStream),
+        Response = (Connect, TcpStream),
         Error = ConnectorError,
         InitError = E,
     > + Clone {
-        move || -> FutureResult<Connector<T>, E> { ok(Connector::new(cfg.clone(), opts)) }
+        move || -> FutureResult<Connector, E> { ok(Connector::new(cfg.clone(), opts)) }
-    }
-
-    pub fn change_request<T2: HostAware>(&self) -> Connector<T2> {
-        Connector {
-            resolver: self.resolver.change_request(),
-        }
     }
 }
 
-impl<T> Clone for Connector<T> {
+impl Clone for Connector {
     fn clone(&self) -> Self {
         Connector {
             resolver: self.resolver.clone(),
@@ -108,11 +170,11 @@ impl<T> Clone for Connector<T> {
     }
 }
 
-impl<T: HostAware> Service for Connector<T> {
+impl Service for Connector {
-    type Request = T;
+    type Request = Connect;
-    type Response = (T, ConnectionInfo, TcpStream);
+    type Response = (Connect, TcpStream);
     type Error = ConnectorError;
-    type Future = ConnectorFuture<T>;
+    type Future = ConnectorFuture;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         Ok(Async::Ready(()))
@@ -127,25 +189,25 @@ impl<T: HostAware> Service for Connector<T> {
 }
 
 #[doc(hidden)]
-pub struct ConnectorFuture<T: HostAware> {
+pub struct ConnectorFuture {
-    fut: ResolverFuture<T>,
+    fut: ResolverFuture<Connect>,
-    fut2: Option<TcpConnector<T>>,
+    fut2: Option<TcpConnectorResponse<Connect>>,
 }
 
-impl<T: HostAware> Future for ConnectorFuture<T> {
+impl Future for ConnectorFuture {
-    type Item = (T, ConnectionInfo, TcpStream);
+    type Item = (Connect, TcpStream);
     type Error = ConnectorError;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         if let Some(ref mut fut) = self.fut2 {
-            return fut.poll();
+            return fut.poll().map_err(ConnectorError::from);
         }
         match self.fut.poll().map_err(ConnectorError::from)? {
-            Async::Ready((req, host, addrs)) => {
+            Async::Ready((req, addrs)) => {
                 if addrs.is_empty() {
                     Err(ConnectorError::NoRecords)
                 } else {
-                    self.fut2 = Some(TcpConnector::new(req, host, addrs));
+                    self.fut2 = Some(TcpConnectorResponse::new(req, addrs));
                     self.poll()
                 }
             }
@@ -154,26 +216,101 @@ impl<T: HostAware> Future for ConnectorFuture<T> {
     }
 }
 
-#[derive(Clone)]
+/// Tcp stream connector service
-pub struct DefaultConnector<T: HostAware>(Connector<T>);
+pub struct TcpConnector<T: RequestPort>(PhantomData<T>);
 
-impl<T: HostAware> Default for DefaultConnector<T> {
+impl<T: RequestPort> Default for TcpConnector<T> {
+    fn default() -> TcpConnector<T> {
+        TcpConnector(PhantomData)
+    }
+}
+
+impl<T: RequestPort> Service for TcpConnector<T> {
+    type Request = (T, VecDeque<IpAddr>);
+    type Response = (T, TcpStream);
+    type Error = io::Error;
+    type Future = TcpConnectorResponse<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, (req, addrs): Self::Request) -> Self::Future {
+        TcpConnectorResponse::new(req, addrs)
+    }
+}
+
+#[doc(hidden)]
+/// Tcp stream connector response future
+pub struct TcpConnectorResponse<T: RequestPort> {
+    port: u16,
+    req: Option<T>,
+    addr: Option<SocketAddr>,
+    addrs: VecDeque<IpAddr>,
+    stream: Option<ConnectFuture>,
+}
+
+impl<T: RequestPort> TcpConnectorResponse<T> {
+    pub fn new(req: T, addrs: VecDeque<IpAddr>) -> TcpConnectorResponse<T> {
+        TcpConnectorResponse {
+            addrs,
+            port: req.port(),
+            req: Some(req),
+            addr: None,
+            stream: None,
+        }
+    }
+}
+
+impl<T: RequestPort> Future for TcpConnectorResponse<T> {
+    type Item = (T, TcpStream);
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        // connect
+        loop {
+            if let Some(new) = self.stream.as_mut() {
+                match new.poll() {
+                    Ok(Async::Ready(sock)) => {
+                        return Ok(Async::Ready((self.req.take().unwrap(), sock)))
+                    }
+                    Ok(Async::NotReady) => return Ok(Async::NotReady),
+                    Err(err) => {
+                        if self.addrs.is_empty() {
+                            return Err(err);
+                        }
+                    }
+                }
+            }
+
+            // try to connect
+            let addr = SocketAddr::new(self.addrs.pop_front().unwrap(), self.port);
+            self.stream = Some(TcpStream::connect(&addr));
+            self.addr = Some(addr)
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct DefaultConnector(Connector);
+
+impl Default for DefaultConnector {
     fn default() -> Self {
         DefaultConnector(Connector::default())
     }
 }
 
-impl<T: HostAware> DefaultConnector<T> {
+impl DefaultConnector {
     pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
         DefaultConnector(Connector::new(cfg, opts))
     }
 }
 
-impl<T: HostAware> Service for DefaultConnector<T> {
+impl Service for DefaultConnector {
-    type Request = T;
+    type Request = Connect;
     type Response = TcpStream;
     type Error = ConnectorError;
-    type Future = DefaultConnectorFuture<T>;
+    type Future = DefaultConnectorFuture;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         self.0.poll_ready()
@@ -187,73 +324,15 @@ impl<T: HostAware> Service for DefaultConnector<T> {
 }
 
 #[doc(hidden)]
-pub struct DefaultConnectorFuture<T: HostAware> {
+pub struct DefaultConnectorFuture {
-    fut: ConnectorFuture<T>,
+    fut: ConnectorFuture,
 }
 
-impl<T: HostAware> Future for DefaultConnectorFuture<T> {
+impl Future for DefaultConnectorFuture {
     type Item = TcpStream;
     type Error = ConnectorError;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        Ok(Async::Ready(try_ready!(self.fut.poll()).2))
+        Ok(Async::Ready(try_ready!(self.fut.poll()).1))
-    }
-}
-
-#[doc(hidden)]
-/// Tcp stream connector
-pub struct TcpConnector<T> {
-    req: Option<T>,
-    host: Option<String>,
-    addr: Option<SocketAddr>,
-    addrs: VecDeque<SocketAddr>,
-    stream: Option<ConnectFuture>,
-}
-
-impl<T> TcpConnector<T> {
-    pub fn new(req: T, host: String, addrs: VecDeque<SocketAddr>) -> TcpConnector<T> {
-        TcpConnector {
-            addrs,
-            req: Some(req),
-            host: Some(host),
-            addr: None,
-            stream: None,
-        }
-    }
-}
-
-impl<T> Future for TcpConnector<T> {
-    type Item = (T, ConnectionInfo, TcpStream);
-    type Error = ConnectorError;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        // connect
-        loop {
-            if let Some(new) = self.stream.as_mut() {
-                match new.poll() {
-                    Ok(Async::Ready(sock)) => {
-                        return Ok(Async::Ready((
-                            self.req.take().unwrap(),
-                            ConnectionInfo {
-                                host: self.host.take().unwrap(),
-                                addr: self.addr.take().unwrap(),
-                            },
-                            sock,
-                        )))
-                    }
-                    Ok(Async::NotReady) => return Ok(Async::NotReady),
-                    Err(err) => {
-                        if self.addrs.is_empty() {
-                            return Err(ConnectorError::IoError(err));
-                        }
-                    }
-                }
-            }
-
-            // try to connect
-            let addr = self.addrs.pop_front().unwrap();
-            self.stream = Some(TcpStream::connect(&addr));
-            self.addr = Some(addr)
-        }
     }
 }

src/keepalive.rs

@@ -6,13 +6,13 @@ use futures::{Async, Future, Poll};
 use tokio_timer::Delay;
 
 use super::service::{NewService, Service};
-use super::timer::{LowResTimer, LowResTimerService};
+use super::time::{LowResTime, LowResTimeService};
 use super::Never;
 
 pub struct KeepAlive<R, E, F> {
     f: F,
     ka: Duration,
-    timer: LowResTimer,
+    time: LowResTime,
     _t: PhantomData<(R, E)>,
 }
 
@@ -20,11 +20,11 @@ impl<R, E, F> KeepAlive<R, E, F>
 where
     F: Fn() -> E + Clone,
 {
-    pub fn new(ka: Duration, timer: LowResTimer, f: F) -> Self {
+    pub fn new(ka: Duration, time: LowResTime, f: F) -> Self {
         KeepAlive {
             f,
             ka,
-            timer,
+            time,
             _t: PhantomData,
         }
     }
@@ -38,7 +38,7 @@ where
         KeepAlive {
             f: self.f.clone(),
             ka: self.ka,
-            timer: self.timer.clone(),
+            time: self.time.clone(),
             _t: PhantomData,
         }
     }
@@ -58,7 +58,7 @@ where
     fn new_service(&self) -> Self::Future {
         ok(KeepAliveService::new(
             self.ka,
-            self.timer.timer(),
+            self.time.timer(),
             self.f.clone(),
         ))
     }
@@ -67,7 +67,7 @@ where
 pub struct KeepAliveService<R, E, F> {
     f: F,
     ka: Duration,
-    timer: LowResTimerService,
+    time: LowResTimeService,
     delay: Delay,
     expire: Instant,
     _t: PhantomData<(R, E)>,
@@ -77,14 +77,14 @@ impl<R, E, F> KeepAliveService<R, E, F>
 where
     F: Fn() -> E,
 {
-    pub fn new(ka: Duration, mut timer: LowResTimerService, f: F) -> Self {
+    pub fn new(ka: Duration, time: LowResTimeService, f: F) -> Self {
-        let expire = timer.now() + ka;
+        let expire = time.now() + ka;
         KeepAliveService {
             f,
             ka,
-            timer,
+            time,
-            delay: Delay::new(expire),
             expire,
+            delay: Delay::new(expire),
             _t: PhantomData,
         }
     }
@@ -102,11 +102,11 @@ where
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         match self.delay.poll() {
             Ok(Async::Ready(_)) => {
-                let now = self.timer.now();
+                let now = self.time.now();
                 if self.expire <= now {
                     Err((self.f)())
                 } else {
-                    self.delay = Delay::new(self.expire);
+                    self.delay.reset(self.expire);
                     let _ = self.delay.poll();
                     Ok(Async::Ready(()))
                 }
@@ -117,7 +117,7 @@ where
     }
 
     fn call(&mut self, req: Self::Request) -> Self::Future {
-        self.expire = self.timer.now() + self.ka;
+        self.expire = self.time.now() + self.ka;
         ok(req)
     }
 }

src/lib.rs

@@ -69,7 +69,8 @@ pub mod server;
 pub mod service;
 pub mod ssl;
 pub mod stream;
-pub mod timer;
+pub mod time;
+pub mod timeout;
 
 #[derive(Copy, Clone, Debug)]
 pub enum Never {}

src/resolver.rs

@@ -1,29 +1,24 @@
 use std::collections::VecDeque;
 use std::marker::PhantomData;
-use std::net::SocketAddr;
+use std::net::IpAddr;
 
 use futures::{Async, Future, Poll};
 
 use tokio_current_thread::spawn;
 use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
-use trust_dns_resolver::error::ResolveError;
+pub use trust_dns_resolver::error::ResolveError;
 use trust_dns_resolver::lookup_ip::LookupIpFuture;
 use trust_dns_resolver::system_conf::read_system_conf;
 use trust_dns_resolver::{AsyncResolver, Background};
 
 use super::service::Service;
 
-#[derive(Debug)]
+/// Host name of the request
-pub enum ResolverError {
+pub trait RequestHost {
-    Resolve(ResolveError),
-    InvalidInput,
-}
-
-pub trait HostAware {
     fn host(&self) -> &str;
 }
 
-impl HostAware for String {
+impl RequestHost for String {
     fn host(&self) -> &str {
         self.as_ref()
     }
@@ -34,7 +29,7 @@ pub struct Resolver<T = String> {
     req: PhantomData<T>,
 }
 
-impl<T: HostAware> Default for Resolver<T> {
+impl<T: RequestHost> Default for Resolver<T> {
     fn default() -> Self {
         let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
             (cfg, opts)
@@ -46,7 +41,8 @@ impl<T: HostAware> Default for Resolver<T> {
     }
 }
 
-impl<T: HostAware> Resolver<T> {
+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);
@@ -56,7 +52,8 @@ impl<T: HostAware> Resolver<T> {
         }
     }
 
-    pub fn change_request<T2: HostAware>(&self) -> Resolver<T2> {
+    /// Change type of resolver request.
+    pub fn into_request<T2: RequestHost>(&self) -> Resolver<T2> {
         Resolver {
             resolver: self.resolver.clone(),
             req: PhantomData,
@@ -73,10 +70,10 @@ impl<T> Clone for Resolver<T> {
     }
 }
 
-impl<T: HostAware> Service for Resolver<T> {
+impl<T: RequestHost> Service for Resolver<T> {
     type Request = T;
-    type Response = (T, String, VecDeque<SocketAddr>);
+    type Response = (T, VecDeque<IpAddr>);
-    type Error = ResolverError;
+    type Error = ResolveError;
     type Future = ResolverFuture<T>;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
@@ -84,7 +81,7 @@ impl<T: HostAware> Service for Resolver<T> {
     }
 
     fn call(&mut self, req: Self::Request) -> Self::Future {
-        ResolverFuture::new(req, 0, &self.resolver)
+        ResolverFuture::new(req, &self.resolver)
     }
 }
 
@@ -92,78 +89,37 @@ impl<T: HostAware> Service for Resolver<T> {
 /// Resolver future
 pub struct ResolverFuture<T> {
     req: Option<T>,
-    port: u16,
     lookup: Option<Background<LookupIpFuture>>,
-    addrs: Option<VecDeque<SocketAddr>>,
+    addrs: Option<VecDeque<IpAddr>>,
-    error: Option<ResolverError>,
-    host: Option<String>,
 }
 
-impl<T: HostAware> ResolverFuture<T> {
+impl<T: RequestHost> ResolverFuture<T> {
-    pub fn new(addr: T, port: u16, resolver: &AsyncResolver) -> Self {
+    pub fn new(addr: T, resolver: &AsyncResolver) -> Self {
         // we need to do dns resolution
-        match ResolverFuture::<T>::parse(addr.host(), port) {
+        let lookup = Some(resolver.lookup_ip(addr.host()));
-            Ok((host, port)) => {
+        ResolverFuture {
-                let lookup = Some(resolver.lookup_ip(host.as_str()));
+            lookup,
-                ResolverFuture {
+            req: Some(addr),
-                    port,
+            addrs: None,
-                    lookup,
-                    req: Some(addr),
-                    host: Some(host.to_owned()),
-                    addrs: None,
-                    error: None,
-                }
-            }
-            Err(err) => ResolverFuture {
-                port,
-                req: None,
-                host: None,
-                lookup: None,
-                addrs: None,
-                error: Some(err),
-            },
         }
     }
-
-    fn parse(addr: &str, port: u16) -> Result<(String, u16), ResolverError> {
-        // split the string by ':' and convert the second part to u16
-        let mut parts_iter = addr.splitn(2, ':');
-        let host = parts_iter.next().ok_or(ResolverError::InvalidInput)?;
-        let port_str = parts_iter.next().unwrap_or("");
-        let port: u16 = port_str.parse().unwrap_or(port);
-
-        Ok((host.to_owned(), port))
-    }
 }
 
-impl<T: HostAware> Future for ResolverFuture<T> {
+impl<T: RequestHost> Future for ResolverFuture<T> {
-    type Item = (T, String, VecDeque<SocketAddr>);
+    type Item = (T, VecDeque<IpAddr>);
-    type Error = ResolverError;
+    type Error = ResolveError;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        if let Some(err) = self.error.take() {
+        if let Some(addrs) = self.addrs.take() {
-            Err(err)
+            Ok(Async::Ready((self.req.take().unwrap(), addrs)))
-        } else if let Some(addrs) = self.addrs.take() {
-            Ok(Async::Ready((
-                self.req.take().unwrap(),
-                self.host.take().unwrap(),
-                addrs,
-            )))
         } else {
             match self.lookup.as_mut().unwrap().poll() {
                 Ok(Async::NotReady) => Ok(Async::NotReady),
-                Ok(Async::Ready(ips)) => {
+                Ok(Async::Ready(ips)) => Ok(Async::Ready((
-                    let addrs: VecDeque<_> = ips
+                    self.req.take().unwrap(),
-                        .iter()
+                    ips.iter().collect(),
-                        .map(|ip| SocketAddr::new(ip, self.port))
+                ))),
-                        .collect();
+                Err(err) => Err(err),
-                    Ok(Async::Ready((
-                        self.req.take().unwrap(),
-                        self.host.take().unwrap(),
-                        addrs,
-                    )))
-                }
-                Err(err) => Err(ResolverError::Resolve(err)),
             }
         }
     }

src/server/accept.rs

@@ -23,7 +23,6 @@ pub(crate) enum Command {
 struct ServerSocketInfo {
     addr: net::SocketAddr,
     token: Token,
-    handler: Token,
     sock: mio::net::TcpListener,
     timeout: Option<Instant>,
 }
@@ -87,7 +86,9 @@ impl AcceptLoop {
     }
 
     pub(crate) fn start(
-        &mut self, socks: Vec<(Token, net::TcpListener)>, workers: Vec<WorkerClient>,
+        &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");
 
@@ -135,9 +136,12 @@ fn connection_error(e: &io::Error) -> bool {
 impl Accept {
     #![cfg_attr(feature = "cargo-clippy", allow(too_many_arguments))]
     pub(crate) fn start(
-        rx: sync_mpsc::Receiver<Command>, cmd_reg: mio::Registration,
+        rx: sync_mpsc::Receiver<Command>,
-        notify_reg: mio::Registration, socks: Vec<(Token, net::TcpListener)>,
+        cmd_reg: mio::Registration,
-        srv: mpsc::UnboundedSender<ServerCommand>, workers: Vec<WorkerClient>,
+        notify_reg: mio::Registration,
+        socks: Vec<(Token, net::TcpListener)>,
+        srv: mpsc::UnboundedSender<ServerCommand>,
+        workers: Vec<WorkerClient>,
     ) {
         let sys = System::current();
 
@@ -173,8 +177,10 @@ impl Accept {
     }
 
     fn new(
-        rx: sync_mpsc::Receiver<Command>, socks: Vec<(Token, net::TcpListener)>,
+        rx: sync_mpsc::Receiver<Command>,
-        workers: Vec<WorkerClient>, srv: mpsc::UnboundedSender<ServerCommand>,
+        socks: Vec<(Token, net::TcpListener)>,
+        workers: Vec<WorkerClient>,
+        srv: mpsc::UnboundedSender<ServerCommand>,
     ) -> Accept {
         // Create a poll instance
         let poll = match mio::Poll::new() {
@@ -184,7 +190,7 @@ impl Accept {
 
         // Start accept
         let mut sockets = Slab::new();
-        for (idx, (hnd_token, lst)) in socks.into_iter().enumerate() {
+        for (hnd_token, lst) in socks.into_iter() {
             let addr = lst.local_addr().unwrap();
             let server = mio::net::TcpListener::from_std(lst)
                 .expect("Can not create mio::net::TcpListener");
@@ -205,7 +211,6 @@ impl Accept {
             entry.insert(ServerSocketInfo {
                 addr,
                 token: hnd_token,
-                handler: Token(idx),
                 sock: server,
                 timeout: None,
             });
@@ -243,9 +248,11 @@ impl Accept {
             for event in events.iter() {
                 let token = event.token();
                 match token {
-                    CMD => if !self.process_cmd() {
+                    CMD => {
-                        return;
+                        if !self.process_cmd() {
-                    },
+                            return;
+                        }
+                    }
                     TIMER => self.process_timer(),
                     NOTIFY => self.backpressure(false),
                     _ => {
@@ -427,7 +434,6 @@ impl Accept {
                     Ok((io, addr)) => Conn {
                         io,
                         token: info.token,
-                        handler: info.handler,
                         peer: Some(addr),
                     },
                     Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return,

src/server/config.rs

@@ -0,0 +1,213 @@
+use std::collections::HashMap;
+use std::{fmt, io, net};
+
+use futures::future::{join_all, Future};
+use tokio_tcp::TcpStream;
+
+use counter::CounterGuard;
+use service::{IntoNewService, NewService};
+
+use super::server::bind_addr;
+use super::services::{
+    BoxedServerService, InternalServiceFactory, ServerMessage, StreamService,
+};
+use super::Token;
+
+pub struct ServiceConfig {
+    pub(super) services: Vec<(String, net::TcpListener)>,
+    pub(super) rt: Box<ServiceRuntimeConfiguration>,
+}
+
+impl ServiceConfig {
+    pub(super) fn new() -> ServiceConfig {
+        ServiceConfig {
+            services: Vec::new(),
+            rt: Box::new(not_configured),
+        }
+    }
+
+    /// Add new service to server
+    pub fn bind<U, N: AsRef<str>>(&mut self, name: N, addr: U) -> io::Result<&mut Self>
+    where
+        U: net::ToSocketAddrs,
+    {
+        let sockets = bind_addr(addr)?;
+
+        for lst in sockets {
+            self.listen(name.as_ref(), lst);
+        }
+
+        Ok(self)
+    }
+
+    /// Add new service to server
+    pub fn listen<N: AsRef<str>>(&mut self, name: N, lst: net::TcpListener) -> &mut Self {
+        self.services.push((name.as_ref().to_string(), lst));
+        self
+    }
+
+    /// Register service configuration function
+    pub fn rt<F>(&mut self, f: F) -> io::Result<()>
+    where
+        F: Fn(&mut ServiceRuntime) + Send + Clone + 'static,
+    {
+        self.rt = Box::new(f);
+        Ok(())
+    }
+}
+
+pub(super) struct ConfiguredService {
+    rt: Box<ServiceRuntimeConfiguration>,
+    names: HashMap<Token, String>,
+    services: HashMap<String, Token>,
+}
+
+impl ConfiguredService {
+    pub(super) fn new(rt: Box<ServiceRuntimeConfiguration>) -> Self {
+        ConfiguredService {
+            rt,
+            names: HashMap::new(),
+            services: HashMap::new(),
+        }
+    }
+
+    pub(super) fn stream(&mut self, token: Token, name: String) {
+        self.names.insert(token, name.clone());
+        self.services.insert(name, token);
+    }
+}
+
+impl InternalServiceFactory for ConfiguredService {
+    fn name(&self, token: Token) -> &str {
+        &self.names[&token]
+    }
+
+    fn clone_factory(&self) -> Box<InternalServiceFactory> {
+        Box::new(Self {
+            rt: self.rt.clone(),
+            names: self.names.clone(),
+            services: self.services.clone(),
+        })
+    }
+
+    fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
+        // configure services
+        let mut rt = ServiceRuntime::new(self.services.clone());
+        self.rt.configure(&mut rt);
+        rt.validate();
+
+        // construct services
+        let mut fut = Vec::new();
+        for (token, ns) in rt.services {
+            fut.push(ns.new_service().map(move |service| (token, service)));
+        }
+
+        Box::new(join_all(fut).map_err(|e| {
+            error!("Can not construct service: {:?}", e);
+        }))
+    }
+}
+
+pub(super) trait ServiceRuntimeConfiguration: Send {
+    fn clone(&self) -> Box<ServiceRuntimeConfiguration>;
+
+    fn configure(&self, &mut ServiceRuntime);
+}
+
+impl<F> ServiceRuntimeConfiguration for F
+where
+    F: Fn(&mut ServiceRuntime) + Send + Clone + 'static,
+{
+    fn clone(&self) -> Box<ServiceRuntimeConfiguration> {
+        Box::new(self.clone())
+    }
+
+    fn configure(&self, rt: &mut ServiceRuntime) {
+        (self)(rt)
+    }
+}
+
+fn not_configured(_: &mut ServiceRuntime) {
+    error!("Service is not configured");
+}
+
+pub struct ServiceRuntime {
+    names: HashMap<String, Token>,
+    services: HashMap<Token, BoxedNewService>,
+}
+
+impl ServiceRuntime {
+    fn new(names: HashMap<String, Token>) -> Self {
+        ServiceRuntime {
+            names,
+            services: HashMap::new(),
+        }
+    }
+
+    fn validate(&self) {
+        for (name, token) in &self.names {
+            if !self.services.contains_key(&token) {
+                error!("Service {:?} is not configured", name);
+            }
+        }
+    }
+
+    pub fn service<T, F>(&mut self, name: &str, service: F)
+    where
+        F: IntoNewService<T>,
+        T: NewService<Request = TcpStream, Response = ()> + 'static,
+        T::Future: 'static,
+        T::Service: 'static,
+        T::InitError: fmt::Debug,
+    {
+        // let name = name.to_owned();
+        if let Some(token) = self.names.get(name) {
+            self.services.insert(
+                token.clone(),
+                Box::new(ServiceFactory {
+                    inner: service.into_new_service(),
+                }),
+            );
+        } else {
+            panic!("Unknown service: {:?}", name);
+        }
+    }
+}
+
+type BoxedNewService = Box<
+    NewService<
+        Request = (Option<CounterGuard>, ServerMessage),
+        Response = (),
+        Error = (),
+        InitError = (),
+        Service = BoxedServerService,
+        Future = Box<Future<Item = BoxedServerService, Error = ()>>,
+    >,
+>;
+
+struct ServiceFactory<T> {
+    inner: T,
+}
+
+impl<T> NewService for ServiceFactory<T>
+where
+    T: NewService<Request = TcpStream, Response = ()>,
+    T::Future: 'static,
+    T::Service: 'static,
+    T::Error: 'static,
+    T::InitError: fmt::Debug + 'static,
+{
+    type Request = (Option<CounterGuard>, ServerMessage);
+    type Response = ();
+    type Error = ();
+    type InitError = ();
+    type Service = BoxedServerService;
+    type Future = Box<Future<Item = BoxedServerService, Error = ()>>;
+
+    fn new_service(&self) -> Self::Future {
+        Box::new(self.inner.new_service().map_err(|_| ()).map(|s| {
+            let service: BoxedServerService = Box::new(StreamService::new(s));
+            service
+        }))
+    }
+}

src/server/mod.rs

@@ -3,10 +3,12 @@
 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};
 
@@ -34,5 +36,13 @@ impl Message for StopServer {
 }
 
 /// Socket id token
-#[derive(Clone, Copy, Debug)]
+#[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
+    }
+}

src/server/server.rs

@@ -12,6 +12,7 @@ use actix::{
 };
 
 use super::accept::{AcceptLoop, AcceptNotify, Command};
+use super::config::{ConfiguredService, ServiceConfig};
 use super::services::{InternalServiceFactory, StreamNewService, StreamServiceFactory};
 use super::services::{ServiceFactory, ServiceNewService};
 use super::worker::{self, Worker, WorkerAvailability, WorkerClient};
@@ -24,6 +25,7 @@ pub(crate) enum ServerCommand {
 /// Server
 pub struct Server {
     threads: usize,
+    token: Token,
     workers: Vec<(usize, WorkerClient)>,
     services: Vec<Box<InternalServiceFactory>>,
     sockets: Vec<(Token, net::TcpListener)>,
@@ -45,6 +47,7 @@ impl Server {
     pub fn new() -> Server {
         Server {
             threads: num_cpus::get(),
+            token: Token(0),
             workers: Vec::new(),
             services: Vec::new(),
             sockets: Vec::new(),
@@ -113,12 +116,24 @@ impl Server {
     /// process
     ///
     /// This function is useful for moving parts of configuration to a
-    /// different module or event library.
+    /// different module or even library.
-    pub fn configure<F>(self, cfg: F) -> Server
+    pub fn configure<F>(mut self, f: F) -> io::Result<Server>
     where
-        F: Fn(Server) -> Server,
+        F: Fn(&mut ServiceConfig) -> io::Result<()>,
     {
-        cfg(self)
+        let mut cfg = ServiceConfig::new();
+
+        f(&mut cfg)?;
+
+        let mut srv = ConfiguredService::new(cfg.rt);
+        for (name, lst) in cfg.services {
+            let token = self.token.next();
+            srv.stream(token, name);
+            self.sockets.push((token, lst));
+        }
+        self.services.push(Box::new(srv));
+
+        Ok(self)
     }
 
     /// Add new service to server
@@ -129,36 +144,53 @@ impl Server {
     {
         let sockets = bind_addr(addr)?;
 
+        let token = self.token.next();
+        self.services.push(StreamNewService::create(
+            name.as_ref().to_string(),
+            token,
+            factory,
+        ));
+
         for lst in sockets {
-            self = self.listen(name.as_ref(), lst, factory.clone())
+            self.sockets.push((token, lst));
         }
         Ok(self)
     }
 
     /// Add new service to server
     pub fn listen<F, N: AsRef<str>>(
-        mut self, name: N, lst: net::TcpListener, factory: F,
+        mut self,
+        name: N,
+        lst: net::TcpListener,
+        factory: F,
     ) -> Self
     where
         F: StreamServiceFactory,
     {
-        let token = Token(self.services.len());
+        let token = self.token.next();
-        self.services
+        self.services.push(StreamNewService::create(
-            .push(StreamNewService::create(name.as_ref().to_string(), factory));
+            name.as_ref().to_string(),
+            token,
+            factory,
+        ));
         self.sockets.push((token, lst));
         self
     }
 
     /// Add new service to server
     pub fn listen2<F, N: AsRef<str>>(
-        mut self, name: N, lst: net::TcpListener, factory: F,
+        mut self,
+        name: N,
+        lst: net::TcpListener,
+        factory: F,
     ) -> Self
     where
         F: ServiceFactory,
     {
-        let token = Token(self.services.len());
+        let token = self.token.next();
         self.services.push(ServiceNewService::create(
             name.as_ref().to_string(),
+            token,
             factory,
         ));
         self.sockets.push((token, lst));
@@ -243,15 +275,16 @@ impl Server {
     }
 
     fn start_worker(&self, idx: usize, notify: AcceptNotify) -> WorkerClient {
-        let (tx, rx) = unbounded();
+        let (tx1, rx1) = unbounded();
+        let (tx2, rx2) = unbounded();
         let timeout = self.shutdown_timeout;
         let avail = WorkerAvailability::new(notify);
-        let worker = WorkerClient::new(idx, tx, avail.clone());
+        let worker = WorkerClient::new(idx, tx1, tx2, avail.clone());
         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 || {
-            Worker::start(rx, services, avail, timeout.clone());
+            Worker::start(rx1, rx2, services, avail, timeout.clone());
             Ok::<_, ()>(())
         }));
 
@@ -395,7 +428,7 @@ impl StreamHandler<ServerCommand, ()> for Server {
     }
 }
 
-fn bind_addr<S: net::ToSocketAddrs>(addr: S) -> io::Result<Vec<net::TcpListener>> {
+pub(super) fn bind_addr<S: net::ToSocketAddrs>(addr: S) -> io::Result<Vec<net::TcpListener>> {
     let mut err = None;
     let mut succ = false;
     let mut sockets = Vec::new();

src/server/services.rs

@@ -7,6 +7,7 @@ use tokio_current_thread::spawn;
 use tokio_reactor::Handle;
 use tokio_tcp::TcpStream;
 
+use super::Token;
 use counter::CounterGuard;
 use service::{NewService, Service};
 
@@ -33,11 +34,11 @@ pub trait ServiceFactory: Send + Clone + 'static {
 }
 
 pub(crate) trait InternalServiceFactory: Send {
-    fn name(&self) -> &str;
+    fn name(&self, token: Token) -> &str;
 
     fn clone_factory(&self) -> Box<InternalServiceFactory>;
 
-    fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>>;
+    fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>>;
 }
 
 pub(crate) type BoxedServerService = Box<
@@ -54,7 +55,7 @@ pub(crate) struct StreamService<T> {
 }
 
 impl<T> StreamService<T> {
-    fn new(service: T) -> Self {
+    pub(crate) fn new(service: T) -> Self {
         StreamService { service }
     }
 }
@@ -133,14 +134,15 @@ where
 pub(crate) struct ServiceNewService<F: ServiceFactory> {
     name: String,
     inner: F,
+    token: Token,
 }
 
 impl<F> ServiceNewService<F>
 where
     F: ServiceFactory,
 {
-    pub(crate) fn create(name: String, inner: F) -> Box<InternalServiceFactory> {
+    pub(crate) fn create(name: String, token: Token, inner: F) -> Box<InternalServiceFactory> {
-        Box::new(Self { name, inner })
+        Box::new(Self { name, inner, token })
     }
 }
 
@@ -148,7 +150,7 @@ impl<F> InternalServiceFactory for ServiceNewService<F>
 where
     F: ServiceFactory,
 {
-    fn name(&self) -> &str {
+    fn name(&self, _: Token) -> &str {
         &self.name
     }
 
@@ -156,10 +158,12 @@ where
         Box::new(Self {
             name: self.name.clone(),
             inner: self.inner.clone(),
+            token: self.token,
         })
     }
 
-    fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>> {
+    fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
+        let token = self.token;
         Box::new(
             self.inner
                 .create()
@@ -167,7 +171,7 @@ where
                 .map_err(|_| ())
                 .map(move |inner| {
                     let service: BoxedServerService = Box::new(ServerService::new(inner));
-                    service
+                    vec![(token, service)]
                 }),
         )
     }
@@ -176,14 +180,15 @@ where
 pub(crate) struct StreamNewService<F: StreamServiceFactory> {
     name: String,
     inner: F,
+    token: Token,
 }
 
 impl<F> StreamNewService<F>
 where
     F: StreamServiceFactory,
 {
-    pub(crate) fn create(name: String, inner: F) -> Box<InternalServiceFactory> {
+    pub(crate) fn create(name: String, token: Token, inner: F) -> Box<InternalServiceFactory> {
-        Box::new(Self { name, inner })
+        Box::new(Self { name, token, inner })
     }
 }
 
@@ -191,7 +196,7 @@ impl<F> InternalServiceFactory for StreamNewService<F>
 where
     F: StreamServiceFactory,
 {
-    fn name(&self) -> &str {
+    fn name(&self, _: Token) -> &str {
         &self.name
     }
 
@@ -199,10 +204,12 @@ where
         Box::new(Self {
             name: self.name.clone(),
             inner: self.inner.clone(),
+            token: self.token,
         })
     }
 
-    fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>> {
+    fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
+        let token = self.token;
         Box::new(
             self.inner
                 .create()
@@ -210,22 +217,22 @@ where
                 .map_err(|_| ())
                 .map(move |inner| {
                     let service: BoxedServerService = Box::new(StreamService::new(inner));
-                    service
+                    vec![(token, service)]
                 }),
         )
     }
 }
 
 impl InternalServiceFactory for Box<InternalServiceFactory> {
-    fn name(&self) -> &str {
+    fn name(&self, token: Token) -> &str {
-        self.as_ref().name()
+        self.as_ref().name(token)
     }
 
     fn clone_factory(&self) -> Box<InternalServiceFactory> {
         self.as_ref().clone_factory()
     }
 
-    fn create(&self) -> Box<Future<Item = BoxedServerService, Error = ()>> {
+    fn create(&self) -> Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>> {
         self.as_ref().create()
     }
 }

src/server/worker.rs

@@ -16,17 +16,18 @@ use super::services::{BoxedServerService, InternalServiceFactory, ServerMessage}
 use super::Token;
 use counter::Counter;
 
-pub(crate) enum WorkerCommand {
+pub(crate) struct WorkerCommand(Conn);
-    Message(Conn),
+
-    /// Stop worker message. Returns `true` on successful shutdown
+/// Stop worker message. Returns `true` on successful shutdown
-    /// and `false` if some connections still alive.
+/// and `false` if some connections still alive.
-    Stop(bool, oneshot::Sender<bool>),
+pub(crate) struct StopCommand {
+    graceful: bool,
+    result: oneshot::Sender<bool>,
 }
 
 #[derive(Debug, Message)]
 pub(crate) struct Conn {
     pub io: net::TcpStream,
-    pub handler: Token,
     pub token: Token,
     pub peer: Option<net::SocketAddr>,
 }
@@ -55,24 +56,30 @@ thread_local! {
 #[derive(Clone)]
 pub(crate) struct WorkerClient {
     pub idx: usize,
-    tx: UnboundedSender<WorkerCommand>,
+    tx1: UnboundedSender<WorkerCommand>,
+    tx2: UnboundedSender<StopCommand>,
     avail: WorkerAvailability,
 }
 
 impl WorkerClient {
     pub fn new(
-        idx: usize, tx: UnboundedSender<WorkerCommand>, avail: WorkerAvailability,
+        idx: usize,
+        tx1: UnboundedSender<WorkerCommand>,
+        tx2: UnboundedSender<StopCommand>,
+        avail: WorkerAvailability,
     ) -> Self {
-        WorkerClient { idx, tx, avail }
+        WorkerClient {
+            idx,
+            tx1,
+            tx2,
+            avail,
+        }
     }
 
     pub fn send(&self, msg: Conn) -> Result<(), Conn> {
-        self.tx
+        self.tx1
-            .unbounded_send(WorkerCommand::Message(msg))
+            .unbounded_send(WorkerCommand(msg))
-            .map_err(|e| match e.into_inner() {
+            .map_err(|msg| msg.into_inner().0)
-                WorkerCommand::Message(msg) => msg,
-                _ => panic!(),
-            })
     }
 
     pub fn available(&self) -> bool {
@@ -80,8 +87,8 @@ impl WorkerClient {
     }
 
     pub fn stop(&self, graceful: bool) -> oneshot::Receiver<bool> {
-        let (tx, rx) = oneshot::channel();
+        let (result, rx) = oneshot::channel();
-        let _ = self.tx.unbounded_send(WorkerCommand::Stop(graceful, tx));
+        let _ = self.tx2.unbounded_send(StopCommand { graceful, result });
         rx
     }
 }
@@ -118,7 +125,8 @@ impl WorkerAvailability {
 /// processing.
 pub(crate) struct Worker {
     rx: UnboundedReceiver<WorkerCommand>,
-    services: Vec<BoxedServerService>,
+    rx2: UnboundedReceiver<StopCommand>,
+    services: Vec<Option<(usize, BoxedServerService)>>,
     availability: WorkerAvailability,
     conns: Counter,
     factories: Vec<Box<InternalServiceFactory>>,
@@ -128,12 +136,16 @@ pub(crate) struct Worker {
 
 impl Worker {
     pub(crate) fn start(
-        rx: UnboundedReceiver<WorkerCommand>, factories: Vec<Box<InternalServiceFactory>>,
+        rx: UnboundedReceiver<WorkerCommand>,
-        availability: WorkerAvailability, shutdown_timeout: time::Duration,
+        rx2: UnboundedReceiver<StopCommand>,
+        factories: Vec<Box<InternalServiceFactory>>,
+        availability: WorkerAvailability,
+        shutdown_timeout: time::Duration,
     ) {
         availability.set(false);
         let mut wrk = MAX_CONNS_COUNTER.with(|conns| Worker {
             rx,
+            rx2,
             availability,
             factories,
             shutdown_timeout,
@@ -143,8 +155,12 @@ impl Worker {
         });
 
         let mut fut = Vec::new();
-        for factory in &wrk.factories {
+        for (idx, factory) in wrk.factories.iter().enumerate() {
-            fut.push(factory.create());
+            fut.push(factory.create().map(move |res| {
+                res.into_iter()
+                    .map(|(t, s)| (idx, t, s))
+                    .collect::<Vec<_>>()
+            }));
         }
         spawn(
             future::join_all(fut)
@@ -152,7 +168,14 @@ impl Worker {
                     error!("Can not start worker: {:?}", e);
                     Arbiter::current().do_send(StopArbiter(0));
                 }).and_then(move |services| {
-                    wrk.services.extend(services);
+                    for item in services {
+                        for (idx, token, service) in item {
+                            while token.0 >= wrk.services.len() {
+                                wrk.services.push(None);
+                            }
+                            wrk.services[token.0] = Some((idx, service));
+                        }
+                    }
                     wrk
                 }),
         );
@@ -161,33 +184,42 @@ impl Worker {
     fn shutdown(&mut self, force: bool) {
         if force {
             self.services.iter_mut().for_each(|h| {
-                let _ = h.call((None, ServerMessage::ForceShutdown));
+                if let Some(h) = h {
+                    let _ = h.1.call((None, ServerMessage::ForceShutdown));
+                }
             });
         } else {
             let timeout = self.shutdown_timeout;
             self.services.iter_mut().for_each(move |h| {
-                let _ = h.call((None, ServerMessage::Shutdown(timeout.clone())));
+                if let Some(h) = h {
+                    let _ = h.1.call((None, ServerMessage::Shutdown(timeout.clone())));
+                }
             });
         }
     }
 
-    fn check_readiness(&mut self, trace: bool) -> Result<bool, usize> {
+    fn check_readiness(&mut self, trace: bool) -> Result<bool, (Token, usize)> {
         let mut ready = self.conns.available();
         let mut failed = None;
-        for (idx, service) in self.services.iter_mut().enumerate() {
+        for (token, service) in &mut self.services.iter_mut().enumerate() {
-            match service.poll_ready() {
+            if let Some(service) = service {
-                Ok(Async::Ready(_)) => {
+                match service.1.poll_ready() {
-                    if trace {
+                    Ok(Async::Ready(_)) => {
-                        trace!("Service {:?} is available", self.factories[idx].name());
+                        if trace {
+                            trace!(
+                                "Service {:?} is available",
+                                self.factories[service.0].name(Token(token))
+                            );
+                        }
+                    }
+                    Ok(Async::NotReady) => ready = false,
+                    Err(_) => {
+                        error!(
+                            "Service {:?} readiness check returned error, restarting",
+                            self.factories[service.0].name(Token(token))
+                        );
+                        failed = Some((Token(token), service.0));
                     }
-                }
-                Ok(Async::NotReady) => ready = false,
-                Err(_) => {
-                    error!(
-                        "Service {:?} readiness check returned error, restarting",
-                        self.factories[idx].name()
-                    );
-                    failed = Some(idx);
                 }
             }
         }
@@ -203,7 +235,11 @@ enum WorkerState {
     None,
     Available,
     Unavailable(Vec<Conn>),
-    Restarting(usize, Box<Future<Item = BoxedServerService, Error = ()>>),
+    Restarting(
+        usize,
+        Token,
+        Box<Future<Item = Vec<(Token, BoxedServerService)>, Error = ()>>,
+    ),
     Shutdown(Delay, Delay, oneshot::Sender<bool>),
 }
 
@@ -212,6 +248,39 @@ impl Future for Worker {
     type Error = ();
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        // `StopWorker` message handler
+        match self.rx2.poll() {
+            Ok(Async::Ready(Some(StopCommand { graceful, result }))) => {
+                self.availability.set(false);
+                let num = num_connections();
+                if num == 0 {
+                    info!("Shutting down worker, 0 connections");
+                    let _ = result.send(true);
+                    return Ok(Async::Ready(()));
+                } else if graceful {
+                    self.shutdown(false);
+                    let num = num_connections();
+                    if num != 0 {
+                        info!("Graceful worker shutdown, {} connections", num);
+                        self.state = WorkerState::Shutdown(
+                            sleep(time::Duration::from_secs(1)),
+                            sleep(self.shutdown_timeout),
+                            result,
+                        );
+                    } else {
+                        let _ = result.send(true);
+                        return Ok(Async::Ready(()));
+                    }
+                } else {
+                    info!("Force shutdown worker, {} connections", num);
+                    self.shutdown(true);
+                    let _ = result.send(false);
+                    return Ok(Async::Ready(()));
+                }
+            }
+            _ => (),
+        }
+
         let state = mem::replace(&mut self.state, WorkerState::None);
 
         match state {
@@ -225,7 +294,10 @@ impl Future for Worker {
                             match self.check_readiness(false) {
                                 Ok(true) => {
                                     let guard = self.conns.get();
-                                    let _ = self.services[msg.handler.0]
+                                    let _ = self.services[msg.token.0]
+                                        .as_mut()
+                                        .expect("actix net bug")
+                                        .1
                                         .call((Some(guard), ServerMessage::Connect(msg.io)));
                                 }
                                 Ok(false) => {
@@ -233,13 +305,14 @@ impl Future for Worker {
                                     self.state = WorkerState::Unavailable(conns);
                                     return self.poll();
                                 }
-                                Err(idx) => {
+                                Err((token, idx)) => {
                                     trace!(
                                         "Service {:?} failed, restarting",
-                                        self.factories[idx].name()
+                                        self.factories[idx].name(token)
                                     );
                                     self.state = WorkerState::Restarting(
                                         idx,
+                                        token,
                                         self.factories[idx].create(),
                                     );
                                     return self.poll();
@@ -253,32 +326,38 @@ impl Future for Worker {
                         self.state = WorkerState::Unavailable(conns);
                         return Ok(Async::NotReady);
                     }
-                    Err(idx) => {
+                    Err((token, idx)) => {
                         trace!(
                             "Service {:?} failed, restarting",
-                            self.factories[idx].name()
+                            self.factories[idx].name(token)
                         );
-                        self.state = WorkerState::Restarting(idx, self.factories[idx].create());
+                        self.state =
+                            WorkerState::Restarting(idx, token, self.factories[idx].create());
                         return self.poll();
                     }
                 }
             }
-            WorkerState::Restarting(idx, mut fut) => {
+            WorkerState::Restarting(idx, token, mut fut) => {
                 match fut.poll() {
-                    Ok(Async::Ready(service)) => {
+                    Ok(Async::Ready(item)) => {
-                        trace!(
+                        for (token, service) in item {
-                            "Service {:?} has been restarted",
+                            trace!(
-                            self.factories[idx].name()
+                                "Service {:?} has been restarted",
-                        );
+                                self.factories[idx].name(token)
-                        self.services[idx] = service;
+                            );
-                        self.state = WorkerState::Unavailable(Vec::new());
+                            self.services[token.0] = Some((idx, service));
+                            self.state = WorkerState::Unavailable(Vec::new());
+                        }
                     }
                     Ok(Async::NotReady) => {
-                        self.state = WorkerState::Restarting(idx, fut);
+                        self.state = WorkerState::Restarting(idx, token, fut);
                         return Ok(Async::NotReady);
                     }
                     Err(_) => {
-                        panic!("Can not restart {:?} service", self.factories[idx].name());
+                        panic!(
+                            "Can not restart {:?} service",
+                            self.factories[idx].name(token)
+                        );
                     }
                 }
                 return self.poll();
@@ -317,11 +396,14 @@ impl Future for Worker {
                 loop {
                     match self.rx.poll() {
                         // handle incoming tcp stream
-                        Ok(Async::Ready(Some(WorkerCommand::Message(msg)))) => {
+                        Ok(Async::Ready(Some(WorkerCommand(msg)))) => {
                             match self.check_readiness(false) {
                                 Ok(true) => {
                                     let guard = self.conns.get();
-                                    let _ = self.services[msg.handler.0]
+                                    let _ = self.services[msg.token.0]
+                                        .as_mut()
+                                        .expect("actix net bug")
+                                        .1
                                         .call((Some(guard), ServerMessage::Connect(msg.io)));
                                     continue;
                                 }
@@ -330,49 +412,21 @@ impl Future for Worker {
                                     self.availability.set(false);
                                     self.state = WorkerState::Unavailable(vec![msg]);
                                 }
-                                Err(idx) => {
+                                Err((token, idx)) => {
                                     trace!(
                                         "Service {:?} failed, restarting",
-                                        self.factories[idx].name()
+                                        self.factories[idx].name(token)
                                     );
                                     self.availability.set(false);
                                     self.state = WorkerState::Restarting(
                                         idx,
+                                        token,
                                         self.factories[idx].create(),
                                     );
                                 }
                             }
                             return self.poll();
                         }
-                        // `StopWorker` message handler
-                        Ok(Async::Ready(Some(WorkerCommand::Stop(graceful, tx)))) => {
-                            self.availability.set(false);
-                            let num = num_connections();
-                            if num == 0 {
-                                info!("Shutting down worker, 0 connections");
-                                let _ = tx.send(true);
-                                return Ok(Async::Ready(()));
-                            } else if graceful {
-                                self.shutdown(false);
-                                let num = num_connections();
-                                if num != 0 {
-                                    info!("Graceful worker shutdown, {} connections", num);
-                                    break Some(WorkerState::Shutdown(
-                                        sleep(time::Duration::from_secs(1)),
-                                        sleep(self.shutdown_timeout),
-                                        tx,
-                                    ));
-                                } else {
-                                    let _ = tx.send(true);
-                                    return Ok(Async::Ready(()));
-                                }
-                            } else {
-                                info!("Force shutdown worker, {} connections", num);
-                                self.shutdown(true);
-                                let _ = tx.send(false);
-                                return Ok(Async::Ready(()));
-                            }
-                        }
                         Ok(Async::NotReady) => {
                             self.state = WorkerState::Available;
                             return Ok(Async::NotReady);
@@ -383,7 +437,5 @@ impl Future for Worker {
             }
             WorkerState::None => panic!(),
         };
-
-        Ok(Async::NotReady)
     }
 }

src/service/mod.rs

@@ -27,7 +27,9 @@ pub trait ServiceExt: Service {
     /// Apply function to specified service and use it as a next service in
     /// chain.
     fn apply<S, I, F, R>(
-        self, service: I, f: F,
+        self,
+        service: I,
+        f: F,
     ) -> AndThen<Self, Apply<S, F, R, Self::Response>>
     where
         Self: Sized,
@@ -120,7 +122,9 @@ pub trait ServiceExt: Service {
 
 pub trait NewServiceExt: NewService {
     fn apply<S, I, F, R>(
-        self, service: I, f: F,
+        self,
+        service: I,
+        f: F,
     ) -> AndThenNewService<Self, ApplyNewService<S, F, R, Self::Response>>
     where
         Self: Sized,

src/ssl/openssl.rs

@@ -6,8 +6,8 @@ use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_openssl::{AcceptAsync, ConnectAsync, SslAcceptorExt, SslConnectorExt, SslStream};
 
 use super::MAX_CONN_COUNTER;
-use connector::ConnectionInfo;
 use counter::{Counter, CounterGuard};
+use resolver::RequestHost;
 use service::{NewService, Service};
 
 /// Support `SSL` connections via openssl package
@@ -102,113 +102,95 @@ impl<T: AsyncRead + AsyncWrite> Future for OpensslAcceptorServiceFut<T> {
 }
 
 /// Openssl connector factory
-pub struct OpensslConnector<T, Io, E> {
+pub struct OpensslConnector<R, T, E> {
     connector: SslConnector,
-    t: PhantomData<T>,
+    _t: PhantomData<(R, T, E)>,
-    io: PhantomData<Io>,
-    _e: PhantomData<E>,
 }
 
-impl<T, Io, E> OpensslConnector<T, Io, E> {
+impl<R, T, E> OpensslConnector<R, T, E> {
     pub fn new(connector: SslConnector) -> Self {
         OpensslConnector {
             connector,
-            t: PhantomData,
+            _t: PhantomData,
-            io: PhantomData,
-            _e: PhantomData,
         }
     }
 }
 
-impl<T, Io: AsyncRead + AsyncWrite> OpensslConnector<T, Io, ()> {
+impl<R: RequestHost, T: AsyncRead + AsyncWrite> OpensslConnector<R, T, ()> {
     pub fn service(
         connector: SslConnector,
-    ) -> impl Service<
+    ) -> impl Service<Request = (R, T), Response = (R, SslStream<T>), Error = Error> {
-        Request = (T, ConnectionInfo, Io),
-        Response = (T, ConnectionInfo, SslStream<Io>),
-        Error = Error,
-    > {
         OpensslConnectorService {
             connector: connector,
-            t: PhantomData,
+            _t: PhantomData,
-            io: PhantomData,
         }
     }
 }
 
-impl<T, Io, E> Clone for OpensslConnector<T, Io, E> {
+impl<R, T, E> Clone for OpensslConnector<R, T, E> {
     fn clone(&self) -> Self {
         Self {
             connector: self.connector.clone(),
-            t: PhantomData,
+            _t: PhantomData,
-            io: PhantomData,
-            _e: PhantomData,
         }
     }
 }
 
-impl<T, Io: AsyncRead + AsyncWrite, E> NewService for OpensslConnector<T, Io, E> {
+impl<R: RequestHost, T: AsyncRead + AsyncWrite, E> NewService for OpensslConnector<R, T, E> {
-    type Request = (T, ConnectionInfo, Io);
+    type Request = (R, T);
-    type Response = (T, ConnectionInfo, SslStream<Io>);
+    type Response = (R, SslStream<T>);
     type Error = Error;
-    type Service = OpensslConnectorService<T, Io>;
+    type Service = OpensslConnectorService<R, T>;
     type InitError = E;
     type Future = FutureResult<Self::Service, Self::InitError>;
 
     fn new_service(&self) -> Self::Future {
         ok(OpensslConnectorService {
             connector: self.connector.clone(),
-            t: PhantomData,
+            _t: PhantomData,
-            io: PhantomData,
         })
     }
 }
 
-pub struct OpensslConnectorService<T, Io> {
+pub struct OpensslConnectorService<R, T> {
     connector: SslConnector,
-    t: PhantomData<T>,
+    _t: PhantomData<(R, T)>,
-    io: PhantomData<Io>,
 }
 
-impl<T, Io: AsyncRead + AsyncWrite> Service for OpensslConnectorService<T, Io> {
+impl<R: RequestHost, T: AsyncRead + AsyncWrite> Service for OpensslConnectorService<R, T> {
-    type Request = (T, ConnectionInfo, Io);
+    type Request = (R, T);
-    type Response = (T, ConnectionInfo, SslStream<Io>);
+    type Response = (R, SslStream<T>);
     type Error = Error;
-    type Future = ConnectAsyncExt<T, Io>;
+    type Future = ConnectAsyncExt<R, T>;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         Ok(Async::Ready(()))
     }
 
-    fn call(&mut self, (req, info, stream): Self::Request) -> Self::Future {
+    fn call(&mut self, (req, stream): Self::Request) -> Self::Future {
         ConnectAsyncExt {
-            fut: SslConnectorExt::connect_async(&self.connector, &info.host, stream),
+            fut: SslConnectorExt::connect_async(&self.connector, req.host(), stream),
             req: Some(req),
-            host: Some(info),
         }
     }
 }
 
-pub struct ConnectAsyncExt<T, Io> {
+pub struct ConnectAsyncExt<R, T> {
-    fut: ConnectAsync<Io>,
+    req: Option<R>,
-    req: Option<T>,
+    fut: ConnectAsync<T>,
-    host: Option<ConnectionInfo>,
 }
 
-impl<T, Io> Future for ConnectAsyncExt<T, Io>
+impl<R, T> Future for ConnectAsyncExt<R, T>
 where
-    Io: AsyncRead + AsyncWrite,
+    R: RequestHost,
+    T: AsyncRead + AsyncWrite,
 {
-    type Item = (T, ConnectionInfo, SslStream<Io>);
+    type Item = (R, SslStream<T>);
     type Error = Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         match self.fut.poll()? {
-            Async::Ready(stream) => Ok(Async::Ready((
+            Async::Ready(stream) => Ok(Async::Ready((self.req.take().unwrap(), stream))),
-                self.req.take().unwrap(),
-                self.host.take().unwrap(),
-                stream,
-            ))),
             Async::NotReady => Ok(Async::NotReady),
         }
     }

src/stream.rs

@@ -2,7 +2,7 @@ use std::marker::PhantomData;
 
 use futures::unsync::mpsc;
 use futures::{future, Async, Future, Poll, Stream};
-use tokio::executor::current_thread::spawn;
+use tokio_current_thread::spawn;
 
 use super::service::{IntoService, NewService, Service};
 
@@ -103,6 +103,12 @@ impl<T> TakeItem<T> {
     }
 }
 
+impl<T> Clone for TakeItem<T> {
+    fn clone(&self) -> TakeItem<T> {
+        TakeItem { _t: PhantomData }
+    }
+}
+
 impl<T: Stream> NewService for TakeItem<T> {
     type Request = T;
     type Response = (Option<T::Item>, T);
@@ -121,6 +127,12 @@ pub struct TakeItemService<T> {
     _t: PhantomData<T>,
 }
 
+impl<T> Clone for TakeItemService<T> {
+    fn clone(&self) -> TakeItemService<T> {
+        TakeItemService { _t: PhantomData }
+    }
+}
+
 impl<T: Stream> Service for TakeItemService<T> {
     type Request = T;
     type Response = (Option<T::Item>, T);

src/time.rs

@@ -10,7 +10,7 @@ use super::service::{NewService, Service};
 use super::Never;
 
 #[derive(Clone, Debug)]
-pub struct LowResTimer(Cell<Inner>);
+pub struct LowResTime(Cell<Inner>);
 
 #[derive(Debug)]
 struct Inner {
@@ -27,28 +27,28 @@ impl Inner {
     }
 }
 
-impl LowResTimer {
+impl LowResTime {
-    pub fn with(resolution: Duration) -> LowResTimer {
+    pub fn with(resolution: Duration) -> LowResTime {
-        LowResTimer(Cell::new(Inner::new(resolution)))
+        LowResTime(Cell::new(Inner::new(resolution)))
     }
 
-    pub fn timer(&self) -> LowResTimerService {
+    pub fn timer(&self) -> LowResTimeService {
-        LowResTimerService(self.0.clone())
+        LowResTimeService(self.0.clone())
     }
 }
 
-impl Default for LowResTimer {
+impl Default for LowResTime {
     fn default() -> Self {
-        LowResTimer(Cell::new(Inner::new(Duration::from_secs(1))))
+        LowResTime(Cell::new(Inner::new(Duration::from_secs(1))))
     }
 }
 
-impl NewService for LowResTimer {
+impl NewService for LowResTime {
     type Request = ();
     type Response = Instant;
     type Error = Never;
     type InitError = Never;
-    type Service = LowResTimerService;
+    type Service = LowResTimeService;
     type Future = FutureResult<Self::Service, Self::InitError>;
 
     fn new_service(&self) -> Self::Future {
@@ -57,16 +57,16 @@ impl NewService for LowResTimer {
 }
 
 #[derive(Clone, Debug)]
-pub struct LowResTimerService(Cell<Inner>);
+pub struct LowResTimeService(Cell<Inner>);
 
-impl LowResTimerService {
+impl LowResTimeService {
-    pub fn with(resolution: Duration) -> LowResTimerService {
+    pub fn with(resolution: Duration) -> LowResTimeService {
-        LowResTimerService(Cell::new(Inner::new(resolution)))
+        LowResTimeService(Cell::new(Inner::new(resolution)))
     }
 
     /// Get current time. This function has to be called from
     /// future's poll method, otherwise it panics.
-    pub fn now(&mut self) -> Instant {
+    pub fn now(&self) -> Instant {
         let cur = self.0.borrow().current.clone();
         if let Some(cur) = cur {
             cur
@@ -88,7 +88,7 @@ impl LowResTimerService {
     }
 }
 
-impl Service for LowResTimerService {
+impl Service for LowResTimeService {
     type Request = ();
     type Response = Instant;
     type Error = Never;

src/timeout.rs

@@ -0,0 +1,162 @@
+//! Service that applies a timeout to requests.
+//!
+//! If the response does not complete within the specified timeout, the response
+//! will be aborted.
+use std::fmt;
+use std::time::Duration;
+
+use futures::{Async, Future, Poll};
+use tokio_timer::{clock, Delay};
+
+use service::{NewService, Service};
+
+/// Applies a timeout to requests.
+#[derive(Debug)]
+pub struct Timeout<T: NewService + Clone> {
+    inner: T,
+    timeout: Duration,
+}
+
+/// Timeout error
+pub enum TimeoutError<E> {
+    /// Service error
+    Service(E),
+    /// Service call timeout
+    Timeout,
+}
+
+impl<E: fmt::Debug> fmt::Debug for TimeoutError<E> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            TimeoutError::Service(e) => write!(f, "TimeoutError::Service({:?})", e),
+            TimeoutError::Timeout => write!(f, "TimeoutError::Timeout"),
+        }
+    }
+}
+
+impl<T> Timeout<T>
+where
+    T: NewService + Clone,
+{
+    pub fn new(timeout: Duration, inner: T) -> Self {
+        Timeout { inner, timeout }
+    }
+}
+
+impl<T> NewService for Timeout<T>
+where
+    T: NewService + Clone,
+{
+    type Request = T::Request;
+    type Response = T::Response;
+    type Error = TimeoutError<T::Error>;
+    type InitError = T::InitError;
+    type Service = TimeoutService<T::Service>;
+    type Future = TimeoutFut<T>;
+
+    fn new_service(&self) -> Self::Future {
+        TimeoutFut {
+            fut: self.inner.new_service(),
+            timeout: self.timeout.clone(),
+        }
+    }
+}
+
+/// `Timeout` response future
+#[derive(Debug)]
+pub struct TimeoutFut<T: NewService> {
+    fut: T::Future,
+    timeout: Duration,
+}
+
+impl<T> Future for TimeoutFut<T>
+where
+    T: NewService,
+{
+    type Item = TimeoutService<T::Service>;
+    type Error = T::InitError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let service = try_ready!(self.fut.poll());
+        Ok(Async::Ready(TimeoutService::new(self.timeout, service)))
+    }
+}
+
+/// Applies a timeout to requests.
+#[derive(Debug)]
+pub struct TimeoutService<T> {
+    inner: T,
+    timeout: Duration,
+}
+
+impl<T> TimeoutService<T> {
+    pub fn new(timeout: Duration, inner: T) -> Self {
+        TimeoutService { inner, timeout }
+    }
+}
+
+impl<T> Clone for TimeoutService<T>
+where
+    T: Clone,
+{
+    fn clone(&self) -> Self {
+        TimeoutService {
+            inner: self.inner.clone(),
+            timeout: self.timeout,
+        }
+    }
+}
+
+impl<T> Service for TimeoutService<T>
+where
+    T: Service,
+{
+    type Request = T::Request;
+    type Response = T::Response;
+    type Error = TimeoutError<T::Error>;
+    type Future = TimeoutServiceResponse<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        self.inner
+            .poll_ready()
+            .map_err(|e| TimeoutError::Service(e))
+    }
+
+    fn call(&mut self, request: Self::Request) -> Self::Future {
+        TimeoutServiceResponse {
+            fut: self.inner.call(request),
+            sleep: Delay::new(clock::now() + self.timeout),
+        }
+    }
+}
+
+/// `TimeoutService` response future
+#[derive(Debug)]
+pub struct TimeoutServiceResponse<T: Service> {
+    fut: T::Future,
+    sleep: Delay,
+}
+
+impl<T> Future for TimeoutServiceResponse<T>
+where
+    T: Service,
+{
+    type Item = T::Response;
+    type Error = TimeoutError<T::Error>;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        // First, try polling the future
+        match self.fut.poll() {
+            Ok(Async::Ready(v)) => return Ok(Async::Ready(v)),
+            Ok(Async::NotReady) => {}
+            Err(e) => return Err(TimeoutError::Service(e)),
+        }
+
+        // Now check the sleep
+        match self.sleep.poll() {
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Ok(Async::Ready(_)) => Err(TimeoutError::Timeout),
+            Err(_) => Err(TimeoutError::Timeout),
+        }
+    }
+}