update deps

prepare actix-server release
refactor ioframe dispatcher
2025-08-13 02:47:08 +02:00 · 2019-12-11 14:47:30 +06:00 · 2019-12-11 14:43:26 +06:00 · 2019-12-11 14:36:11 +06:00 · 2019-12-11 14:36:00 +06:00 · 2019-12-11 12:42:07 +06:00
53 changed files with 1138 additions and 1097 deletions
--- a/README.md
+++ b/README.md
@@ -30,14 +30,16 @@ fn main() -> io::Result<()> {
                let num = num.clone();
                let acceptor = acceptor.clone();

-                // service for converting incoming TcpStream to a SslStream<TcpStream>
-                fn_service(move |stream: Io<tokio_tcp::TcpStream>| {
-                    SslAcceptorExt::accept_async(&acceptor, stream.into_parts().0)
-                        .map_err(|e| println!("Openssl error: {}", e))
-                })
-                // .and_then() combinator uses other service to convert incoming `Request` to a
-                // `Response` and then uses that response as an input for next
-                // service. in this case, on success we use `logger` service
+                // construct transformation pipeline
+                pipeline(
+                    // service for converting incoming TcpStream to a SslStream<TcpStream>
+                    fn_service(move |stream: actix_rt::net::TcpStream| async move {
+                        SslAcceptorExt::accept_async(&acceptor, stream.into_parts().0).await
+                            .map_err(|e| println!("Openssl error: {}", e))
+                    }))
+                // .and_then() combinator chains result of previos service call to argument
+                /// for next service calll. in this case, on success we chain
+                /// ssl stream to the `logger` service.
                .and_then(fn_service(logger))
                // Next service counts number of connections
                .and_then(move |_| {
--- a/actix-codec/CHANGES.md
+++ b/actix-codec/CHANGES.md
@@ -1,5 +1,13 @@
 # Changes

+## [0.2.0] - 2019-12-10
+
+* Use specific futures dependencies
+
+## [0.2.0-alpha.4]
+
+* Fix buffer remaining capacity calcualtion
+
 ## [0.2.0-alpha.3]

 * Use tokio 0.2
@@ -10,17 +18,14 @@

 * Migrated to `std::future`

-
 ## [0.1.2] - 2019-03-27

 * Added `Framed::map_io()` method.

-
 ## [0.1.1] - 2019-03-06

 * Added `FramedParts::with_read_buffer()` method.

-
 ## [0.1.0] - 2018-12-09

 * Move codec to separate crate
--- a/actix-codec/Cargo.toml
+++ b/actix-codec/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-codec"
-version = "0.2.0-alpha.3"
+version = "0.2.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,6 @@ 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 = ".."

@@ -20,7 +19,8 @@ path = "src/lib.rs"
 [dependencies]
 bitflags = "1.2.1"
 bytes = "0.5.2"
-futures = "0.3.1"
+futures-core = "0.3.1"
+futures-sink = "0.3.1"
 tokio = { version = "0.2.4", default-features=false }
 tokio-util = { version = "0.2.0", default-features=false, features=["codec"] }
 log = "0.4"
--- a/actix-codec/src/framed.rs
+++ b/actix-codec/src/framed.rs
@@ -2,8 +2,9 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::{fmt, io};

-use bytes::{BufMut, BytesMut};
-use futures::{ready, Sink, Stream};
+use bytes::BytesMut;
+use futures_core::{ready, Stream};
+use futures_sink::Sink;

 use crate::{AsyncRead, AsyncWrite, Decoder, Encoder};

@@ -19,8 +20,6 @@ bitflags::bitflags! {

 /// 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 Framed<T, U> {
    io: T,
    codec: U,
@@ -49,10 +48,6 @@ where
    /// `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(io: T, codec: U) -> Framed<T, U> {
        Framed {
            io,
@@ -82,10 +77,6 @@ impl<T, U> Framed<T, U> {
    /// 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: FramedParts<T, U>) -> Framed<T, U> {
        Framed {
            io: parts.io,
@@ -136,15 +127,6 @@ impl<T, U> Framed<T, U> {
        self.write_buf.len() >= HW
    }

-    /// 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.io
-    }
-
    /// Consume the `Frame`, returning `Frame` with different codec.
    pub fn into_framed<U2>(self, codec: U2) -> Framed<T, U2> {
        Framed {
@@ -208,7 +190,7 @@ impl<T, U> Framed<T, U> {
        T: AsyncWrite,
        U: Encoder,
    {
-        let remaining = self.write_buf.remaining_mut();
+        let remaining = self.write_buf.capacity() - self.write_buf.len();
        if remaining < LW {
            self.write_buf.reserve(HW - remaining);
        }
@@ -217,11 +199,14 @@ impl<T, U> Framed<T, U> {
        Ok(())
    }

-    /// Check if framed is able to write more data
-    pub fn is_ready(&self) -> bool {
+    /// Check if framed is able to write more data.
+    ///
+    /// `Framed` object considers ready if there is free space in write buffer.
+    pub fn is_write_ready(&self) -> bool {
        self.write_buf.len() < HW
    }

+    /// Try to read underlying I/O stream and decode item.
    pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
    where
        T: AsyncRead,
@@ -251,9 +236,7 @@ impl<T, U> Framed<T, U> {
                        return Poll::Ready(Some(Ok(frame)));
                    }
                    Err(e) => return Poll::Ready(Some(Err(e))),
-                    _ => {
-                        // Need more data
-                    }
+                    _ => (), // Need more data
                }

                self.flags.remove(Flags::READABLE);
@@ -262,7 +245,7 @@ impl<T, U> Framed<T, U> {
            debug_assert!(!self.flags.contains(Flags::EOF));

            // Otherwise, try to read more data and try again. Make sure we've got room
-            let remaining = self.read_buf.remaining_mut();
+            let remaining = self.read_buf.capacity() - self.read_buf.len();
            if remaining < LW {
                self.read_buf.reserve(HW - remaining)
            }
@@ -281,6 +264,7 @@ impl<T, U> Framed<T, U> {
        }
    }

+    /// Flush write buffer to underlying I/O stream.
    pub fn flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
    where
        T: AsyncWrite,
@@ -298,14 +282,12 @@ impl<T, U> Framed<T, U> {
            if n == 0 {
                return Poll::Ready(Err(io::Error::new(
                    io::ErrorKind::WriteZero,
-                    "failed to \
-                     write frame to transport",
+                    "failed to write frame to transport",
                )
                .into()));
            }

-            // TODO: Add a way to `bytes` to do this w/o returning the drained
-            // data.
+            // remove written data
            let _ = self.write_buf.split_to(n);
        }

@@ -316,6 +298,7 @@ impl<T, U> Framed<T, U> {
        Poll::Ready(Ok(()))
    }

+    /// Flush write buffer and shutdown underlying I/O stream.
    pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
    where
        T: AsyncWrite,
@@ -350,7 +333,7 @@ where
    type Error = U::Error;

    fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        if self.is_ready() {
+        if self.is_write_ready() {
            Poll::Ready(Ok(()))
        } else {
            Poll::Pending
--- a/actix-codec/src/lib.rs
+++ b/actix-codec/src/lib.rs
@@ -7,7 +7,6 @@
 //! [`AsyncRead`]: #
 //! [`AsyncWrite`]: #
 #![deny(rust_2018_idioms, warnings)]
-#![allow(clippy::type_complexity)]

 mod bcodec;
 mod framed;
--- a/actix-connect/Cargo.toml
+++ b/actix-connect/Cargo.toml
@@ -33,10 +33,10 @@ rustls = ["rust-tls", "tokio-rustls", "webpki"]
 uri = ["http"]

 [dependencies]
-actix-service = "1.0.0-alpha.3"
-actix-codec = "0.2.0-alpha.3"
-actix-utils = "1.0.0-alpha.3"
-actix-rt = "1.0.0-alpha.3"
+actix-service = "1.0.0"
+actix-codec = "0.2.0"
+actix-utils = "1.0.0"
+actix-rt = "1.0.0"
 derive_more = "0.99.2"
 either = "1.5.2"
 futures = "0.3.1"
@@ -55,4 +55,4 @@ webpki = { version = "0.21", optional = true }

 [dev-dependencies]
 bytes = "0.5.2"
-actix-testing = { version="1.0.0-alpha.2" }
+actix-testing = { version="1.0.0" }
--- a/actix-connect/tests/test_connect.rs
+++ b/actix-connect/tests/test_connect.rs
@@ -2,7 +2,7 @@ use std::io;

 use actix_codec::{BytesCodec, Framed};
 use actix_rt::net::TcpStream;
-use actix_service::{service_fn, Service, ServiceFactory};
+use actix_service::{fn_service, Service, ServiceFactory};
 use actix_testing::TestServer;
 use bytes::Bytes;
 use futures::SinkExt;
@@ -14,7 +14,7 @@ use actix_connect::Connect;
 #[actix_rt::test]
 async fn test_string() {
    let srv = TestServer::with(|| {
-        service_fn(|io: TcpStream| {
+        fn_service(|io: TcpStream| {
            async {
                let mut framed = Framed::new(io, BytesCodec);
                framed.send(Bytes::from_static(b"test")).await?;
@@ -33,7 +33,7 @@ async fn test_string() {
 #[actix_rt::test]
 async fn test_rustls_string() {
    let srv = TestServer::with(|| {
-        service_fn(|io: TcpStream| {
+        fn_service(|io: TcpStream| {
            async {
                let mut framed = Framed::new(io, BytesCodec);
                framed.send(Bytes::from_static(b"test")).await?;
@@ -51,7 +51,7 @@ async fn test_rustls_string() {
 #[actix_rt::test]
 async fn test_static_str() {
    let srv = TestServer::with(|| {
-        service_fn(|io: TcpStream| {
+        fn_service(|io: TcpStream| {
            async {
                let mut framed = Framed::new(io, BytesCodec);
                framed.send(Bytes::from_static(b"test")).await?;
@@ -75,7 +75,7 @@ async fn test_static_str() {
 #[actix_rt::test]
 async fn test_new_service() {
    let srv = TestServer::with(|| {
-        service_fn(|io: TcpStream| {
+        fn_service(|io: TcpStream| {
            async {
                let mut framed = Framed::new(io, BytesCodec);
                framed.send(Bytes::from_static(b"test")).await?;
@@ -100,7 +100,7 @@ async fn test_uri() {
    use std::convert::TryFrom;

    let srv = TestServer::with(|| {
-        service_fn(|io: TcpStream| {
+        fn_service(|io: TcpStream| {
            async {
                let mut framed = Framed::new(io, BytesCodec);
                framed.send(Bytes::from_static(b"test")).await?;
@@ -121,7 +121,7 @@ async fn test_rustls_uri() {
    use std::convert::TryFrom;

    let srv = TestServer::with(|| {
-        service_fn(|io: TcpStream| {
+        fn_service(|io: TcpStream| {
            async {
                let mut framed = Framed::new(io, BytesCodec);
                framed.send(Bytes::from_static(b"test")).await?;
--- a/actix-ioframe/Cargo.toml
+++ b/actix-ioframe/Cargo.toml
@@ -18,10 +18,10 @@ name = "actix_ioframe"
 path = "src/lib.rs"

 [dependencies]
-actix-service = "1.0.0-alpha.3"
-actix-codec = "0.2.0-alpha.2"
-actix-utils = "1.0.0-alpha.2"
-actix-rt = "1.0.0-alpha.2"
+actix-service = "1.0.0"
+actix-codec = "0.2.0"
+actix-utils = "1.0.0"
+actix-rt = "1.0.0"
 bytes = "0.5"
 either = "1.5.2"
 futures = "0.3.1"
@@ -30,4 +30,4 @@ log = "0.4"

 [dev-dependencies]
 actix-connect = "1.0.0-alpha.2"
-actix-testing = "1.0.0-alpha.2"
+actix-testing = "1.0.0"
--- a/actix-ioframe/src/cell.rs
+++ b/actix-ioframe/src/cell.rs
@@ -1,39 +0,0 @@
-//! Custom cell impl
-
-use std::cell::UnsafeCell;
-use std::fmt;
-use std::rc::Rc;
-
-pub(crate) struct Cell<T> {
-    inner: Rc<UnsafeCell<T>>,
-}
-
-impl<T> Clone for Cell<T> {
-    fn clone(&self) -> Self {
-        Self {
-            inner: self.inner.clone(),
-        }
-    }
-}
-
-impl<T: fmt::Debug> fmt::Debug for Cell<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        self.inner.fmt(f)
-    }
-}
-
-impl<T> Cell<T> {
-    pub fn new(inner: T) -> Self {
-        Self {
-            inner: Rc::new(UnsafeCell::new(inner)),
-        }
-    }
-
-    pub(crate) unsafe fn get_ref(&mut self) -> &T {
-        &*self.inner.as_ref().get()
-    }
-
-    pub(crate) unsafe fn get_mut(&mut self) -> &mut T {
-        &mut *self.inner.as_ref().get()
-    }
-}
--- a/actix-ioframe/src/connect.rs
+++ b/actix-ioframe/src/connect.rs
@@ -3,55 +3,51 @@ use std::pin::Pin;
 use std::task::{Context, Poll};

 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
-use actix_utils::mpsc;
 use futures::Stream;

-use crate::dispatcher::FramedMessage;
 use crate::sink::Sink;

-pub struct Connect<Io, St = (), Codec = ()> {
+pub struct Connect<Io, Codec, Err, St = ()>
+where
+    Codec: Encoder + Decoder,
+{
    io: Io,
+    sink: Sink<<Codec as Encoder>::Item, Err>,
    _t: PhantomData<(St, Codec)>,
 }

-impl<Io> Connect<Io>
+impl<Io, Codec, Err> Connect<Io, Codec, Err>
 where
    Io: AsyncRead + AsyncWrite,
+    Codec: Encoder + Decoder,
 {
-    pub(crate) fn new(io: Io) -> Self {
+    pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item, Err>) -> Self {
        Self {
            io,
+            sink,
            _t: PhantomData,
        }
    }

-    pub fn codec<Codec>(self, codec: Codec) -> ConnectResult<Io, (), Codec>
-    where
-        Codec: Encoder + Decoder,
-    {
-        let (tx, rx) = mpsc::channel();
-        let sink = Sink::new(tx);
-
+    pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec, Err> {
        ConnectResult {
            state: (),
+            sink: self.sink,
            framed: Framed::new(self.io, codec),
-            rx,
-            sink,
        }
    }
 }

 #[pin_project::pin_project]
-pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
+pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Err> {
    pub(crate) state: St,
    pub(crate) framed: Framed<Io, Codec>,
-    pub(crate) rx: mpsc::Receiver<FramedMessage<<Codec as Encoder>::Item>>,
-    pub(crate) sink: Sink<<Codec as Encoder>::Item>,
+    pub(crate) sink: Sink<<Codec as Encoder>::Item, Err>,
 }

-impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
+impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
    #[inline]
-    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
+    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, Err> {
        &self.sink
    }

@@ -66,17 +62,16 @@ impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
    }

    #[inline]
-    pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec> {
+    pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec, Err> {
        ConnectResult {
            state,
            framed: self.framed,
-            rx: self.rx,
            sink: self.sink,
        }
    }
 }

-impl<Io, St, Codec> Stream for ConnectResult<Io, St, Codec>
+impl<Io, St, Codec, Err> Stream for ConnectResult<Io, St, Codec, Err>
 where
    Io: AsyncRead + AsyncWrite,
    Codec: Encoder + Decoder,
@@ -88,7 +83,8 @@ where
    }
 }

-impl<Io, St, Codec> futures::Sink<<Codec as Encoder>::Item> for ConnectResult<Io, St, Codec>
+impl<Io, St, Codec, Err> futures::Sink<<Codec as Encoder>::Item>
+    for ConnectResult<Io, St, Codec, Err>
 where
    Io: AsyncRead + AsyncWrite,
    Codec: Encoder + Decoder,
@@ -96,7 +92,7 @@ where
    type Error = <Codec as Encoder>::Error;

    fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        if self.framed.is_ready() {
+        if self.framed.is_write_ready() {
            Poll::Ready(Ok(()))
        } else {
            Poll::Pending
--- a/actix-ioframe/src/dispatcher.rs
+++ b/actix-ioframe/src/dispatcher.rs
@@ -1,39 +1,34 @@
 //! Framed dispatcher service and related utilities
-use std::collections::VecDeque;
-use std::mem;
 use std::pin::Pin;
 use std::rc::Rc;
 use std::task::{Context, Poll};

 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, Service};
-use actix_utils::task::LocalWaker;
 use actix_utils::{mpsc, oneshot};
-use futures::future::ready;
-use futures::{FutureExt, Sink as FutureSink, Stream};
+use futures::{FutureExt, Stream};
 use log::debug;

-use crate::cell::Cell;
 use crate::error::ServiceError;
 use crate::item::Item;
 use crate::sink::Sink;

-type Request<S, U> = Item<S, U>;
+type Request<S, U, E> = Item<S, U, E>;
 type Response<U> = <U as Encoder>::Item;

-pub(crate) enum FramedMessage<T> {
-    Message(T),
-    Close,
+pub(crate) enum Message<T> {
+    Item(T),
    WaitClose(oneshot::Sender<()>),
+    Close,
 }

 /// FramedTransport - is a future that reads frames from Framed object
 /// and pass then to the service.
 #[pin_project::pin_project]
-pub(crate) struct FramedDispatcher<St, S, T, U>
+pub(crate) struct Dispatcher<St, S, T, U, E>
 where
    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
+    S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>,
    S::Error: 'static,
    S::Future: 'static,
    T: AsyncRead + AsyncWrite,
@@ -42,19 +37,19 @@ where
    <U as Encoder>::Error: std::fmt::Debug,
 {
    service: S,
-    sink: Sink<<U as Encoder>::Item>,
+    sink: Sink<<U as Encoder>::Item, E>,
    state: St,
    dispatch_state: FramedState<S, U>,
    framed: Framed<T, U>,
-    rx: Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
+    rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
+    tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, E>>,
    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
 }

-impl<St, S, T, U> FramedDispatcher<St, S, T, U>
+impl<St, S, T, U, E> Dispatcher<St, S, T, U, E>
 where
    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
+    S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>,
    S::Error: 'static,
    S::Future: 'static,
    T: AsyncRead + AsyncWrite,
@@ -66,22 +61,21 @@ where
        framed: Framed<T, U>,
        state: St,
        service: F,
-        rx: mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>,
-        sink: Sink<<U as Encoder>::Item>,
+        sink: Sink<<U as Encoder>::Item, E>,
+        rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
        disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
    ) -> Self {
-        FramedDispatcher {
+        let tx = rx.sender();
+
+        Dispatcher {
            framed,
            state,
            sink,
            disconnect,
-            rx: Some(rx),
+            rx,
+            tx,
            service: service.into_service(),
            dispatch_state: FramedState::Processing,
-            inner: Cell::new(FramedDispatcherInner {
-                buf: VecDeque::new(),
-                task: LocalWaker::new(),
-            }),
        }
    }
 }
@@ -116,17 +110,26 @@ impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
            }
        }
    }
+
+    fn take_error(&mut self) -> ServiceError<S::Error, U> {
+        match std::mem::replace(self, FramedState::Processing) {
+            FramedState::Error(err) => err,
+            _ => panic!(),
+        }
+    }
+
+    fn take_framed_error(&mut self) -> ServiceError<S::Error, U> {
+        match std::mem::replace(self, FramedState::Processing) {
+            FramedState::FramedError(err) => err,
+            _ => panic!(),
+        }
+    }
 }

-struct FramedDispatcherInner<I, E> {
-    buf: VecDeque<Result<I, E>>,
-    task: LocalWaker,
-}
-
-impl<St, S, T, U> FramedDispatcher<St, S, T, U>
+impl<St, S, T, U, E> Dispatcher<St, S, T, U, E>
 where
    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
+    S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>,
    S::Error: 'static,
    S::Future: 'static,
    T: AsyncRead + AsyncWrite,
@@ -134,263 +137,150 @@ where
    <U as Encoder>::Item: 'static,
    <U as Encoder>::Error: std::fmt::Debug,
 {
+    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool {
+        loop {
+            match self.service.poll_ready(cx) {
+                Poll::Ready(Ok(_)) => {
+                    let item = match self.framed.next_item(cx) {
+                        Poll::Ready(Some(Ok(el))) => el,
+                        Poll::Ready(Some(Err(err))) => {
+                            self.dispatch_state =
+                                FramedState::FramedError(ServiceError::Decoder(err));
+                            return true;
+                        }
+                        Poll::Pending => return false,
+                        Poll::Ready(None) => {
+                            log::trace!("Client disconnected");
+                            self.dispatch_state = FramedState::Stopping;
+                            return true;
+                        }
+                    };
+
+                    let tx = self.tx.clone();
+                    actix_rt::spawn(
+                        self.service
+                            .call(Item::new(self.state.clone(), self.sink.clone(), item))
+                            .map(move |item| {
+                                let item = match item {
+                                    Ok(Some(item)) => Ok(Message::Item(item)),
+                                    Ok(None) => return,
+                                    Err(err) => Err(err),
+                                };
+                                let _ = tx.send(item);
+                            }),
+                    );
+                }
+                Poll::Pending => return false,
+                Poll::Ready(Err(err)) => {
+                    self.dispatch_state = FramedState::Error(ServiceError::Service(err));
+                    return true;
+                }
+            }
+        }
+    }
+
+    /// write to framed object
+    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool {
+        loop {
+            while !self.framed.is_write_buf_full() {
+                match Pin::new(&mut self.rx).poll_next(cx) {
+                    Poll::Ready(Some(Ok(Message::Item(msg)))) => {
+                        if let Err(err) = self.framed.write(msg) {
+                            self.dispatch_state =
+                                FramedState::FramedError(ServiceError::Encoder(err));
+                            return true;
+                        }
+                    }
+                    Poll::Ready(Some(Ok(Message::Close))) => {
+                        self.dispatch_state.stop(None);
+                        return true;
+                    }
+                    Poll::Ready(Some(Ok(Message::WaitClose(tx)))) => {
+                        self.dispatch_state.stop(Some(tx));
+                        return true;
+                    }
+                    Poll::Ready(Some(Err(err))) => {
+                        self.dispatch_state = FramedState::Error(ServiceError::Service(err));
+                        return true;
+                    }
+                    Poll::Ready(None) | Poll::Pending => break,
+                }
+            }
+
+            if !self.framed.is_write_buf_empty() {
+                match self.framed.flush(cx) {
+                    Poll::Pending => break,
+                    Poll::Ready(Ok(_)) => (),
+                    Poll::Ready(Err(err)) => {
+                        debug!("Error sending data: {:?}", err);
+                        self.dispatch_state =
+                            FramedState::FramedError(ServiceError::Encoder(err));
+                        return true;
+                    }
+                }
+            } else {
+                break;
+            }
+        }
+        false
+    }
+
    pub(crate) fn poll(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Result<(), ServiceError<S::Error, U>>> {
-        let this = self;
-        unsafe { this.inner.get_ref().task.register(cx.waker()) };
-
-        poll(
-            cx,
-            &mut this.service,
-            &mut this.state,
-            &mut this.sink,
-            &mut this.framed,
-            &mut this.dispatch_state,
-            &mut this.rx,
-            &mut this.inner,
-            &mut this.disconnect,
-        )
-    }
-}
-
-fn poll<St, S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut St,
-    sink: &mut Sink<<U as Encoder>::Item>,
-    framed: &mut Framed<T, U>,
-    dispatch_state: &mut FramedState<S, U>,
-    rx: &mut Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: &mut Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
-    disconnect: &mut Option<Rc<dyn Fn(&mut St, bool)>>,
-) -> Poll<Result<(), ServiceError<S::Error, U>>>
-where
-    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    match mem::replace(dispatch_state, FramedState::Processing) {
-        FramedState::Processing => {
-            if poll_read(cx, srv, state, sink, framed, dispatch_state, inner)
-                || poll_write(cx, framed, dispatch_state, rx, inner)
-            {
-                poll(
-                    cx,
-                    srv,
-                    state,
-                    sink,
-                    framed,
-                    dispatch_state,
-                    rx,
-                    inner,
-                    disconnect,
-                )
-            } else {
-                Poll::Pending
-            }
-        }
-        FramedState::Error(err) => {
-            if framed.is_write_buf_empty()
-                || (poll_write(cx, framed, dispatch_state, rx, inner)
-                    || framed.is_write_buf_empty())
-            {
-                if let Some(ref disconnect) = disconnect {
-                    (&*disconnect)(&mut *state, true);
+        match self.dispatch_state {
+            FramedState::Processing => {
+                if self.poll_read(cx) || self.poll_write(cx) {
+                    self.poll(cx)
+                } else {
+                    Poll::Pending
                }
-                Poll::Ready(Err(err))
-            } else {
-                *dispatch_state = FramedState::Error(err);
-                Poll::Pending
            }
-        }
-        FramedState::FlushAndStop(mut vec) => {
-            if !framed.is_write_buf_empty() {
-                match Pin::new(framed).poll_flush(cx) {
-                    Poll::Ready(Err(err)) => {
-                        debug!("Error sending data: {:?}", err);
-                    }
-                    Poll::Pending => {
-                        *dispatch_state = FramedState::FlushAndStop(vec);
+            FramedState::Error(_) => {
+                // flush write buffer
+                if !self.framed.is_write_buf_empty() {
+                    if let Poll::Pending = self.framed.flush(cx) {
                        return Poll::Pending;
                    }
-                    Poll::Ready(_) => (),
                }
-            };
-            for tx in vec.drain(..) {
-                let _ = tx.send(());
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(&mut self.state, true);
+                }
+                Poll::Ready(Err(self.dispatch_state.take_error()))
            }
-            if let Some(ref disconnect) = disconnect {
-                (&*disconnect)(&mut *state, false);
-            }
-            Poll::Ready(Ok(()))
-        }
-        FramedState::FramedError(err) => {
-            if let Some(ref disconnect) = disconnect {
-                (&*disconnect)(&mut *state, true);
-            }
-            Poll::Ready(Err(err))
-        }
-        FramedState::Stopping => {
-            if let Some(ref disconnect) = disconnect {
-                (&*disconnect)(&mut *state, false);
-            }
-            Poll::Ready(Ok(()))
-        }
-    }
-}
-
-fn poll_read<St, S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut St,
-    sink: &mut Sink<<U as Encoder>::Item>,
-    framed: &mut Framed<T, U>,
-    dispatch_state: &mut FramedState<S, U>,
-    inner: &mut Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
-) -> bool
-where
-    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    loop {
-        match srv.poll_ready(cx) {
-            Poll::Ready(Ok(_)) => {
-                let item = match framed.next_item(cx) {
-                    Poll::Ready(Some(Ok(el))) => el,
-                    Poll::Ready(Some(Err(err))) => {
-                        *dispatch_state = FramedState::FramedError(ServiceError::Decoder(err));
-                        return true;
-                    }
-                    Poll::Pending => return false,
-                    Poll::Ready(None) => {
-                        log::trace!("Client disconnected");
-                        *dispatch_state = FramedState::Stopping;
-                        return true;
+            FramedState::FlushAndStop(ref mut vec) => {
+                if !self.framed.is_write_buf_empty() {
+                    match self.framed.flush(cx) {
+                        Poll::Ready(Err(err)) => {
+                            debug!("Error sending data: {:?}", err);
+                        }
+                        Poll::Pending => {
+                            return Poll::Pending;
+                        }
+                        Poll::Ready(_) => (),
                    }
                };
-
-                let mut cell = inner.clone();
-                actix_rt::spawn(srv.call(Item::new(state.clone(), sink.clone(), item)).then(
-                    move |item| {
-                        let item = match item {
-                            Ok(Some(item)) => Ok(item),
-                            Ok(None) => return ready(()),
-                            Err(err) => Err(err),
-                        };
-                        unsafe {
-                            let inner = cell.get_mut();
-                            inner.buf.push_back(item);
-                            inner.task.wake();
-                        }
-                        ready(())
-                    },
-                ));
+                for tx in vec.drain(..) {
+                    let _ = tx.send(());
+                }
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(&mut self.state, false);
+                }
+                Poll::Ready(Ok(()))
            }
-            Poll::Pending => return false,
-            Poll::Ready(Err(err)) => {
-                *dispatch_state = FramedState::Error(ServiceError::Service(err));
-                return true;
+            FramedState::FramedError(_) => {
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(&mut self.state, true);
+                }
+                Poll::Ready(Err(self.dispatch_state.take_framed_error()))
+            }
+            FramedState::Stopping => {
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(&mut self.state, false);
+                }
+                Poll::Ready(Ok(()))
            }
        }
    }
 }
-
-/// write to framed object
-fn poll_write<St, S, T, U>(
-    cx: &mut Context<'_>,
-    framed: &mut Framed<T, U>,
-    dispatch_state: &mut FramedState<S, U>,
-    rx: &mut Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: &mut Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
-) -> bool
-where
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    let inner = unsafe { inner.get_mut() };
-    let mut rx_done = rx.is_none();
-    let mut buf_empty = inner.buf.is_empty();
-    loop {
-        while !framed.is_write_buf_full() {
-            if !buf_empty {
-                match inner.buf.pop_front().unwrap() {
-                    Ok(msg) => {
-                        if let Err(err) = framed.write(msg) {
-                            *dispatch_state =
-                                FramedState::FramedError(ServiceError::Encoder(err));
-                            return true;
-                        }
-                        buf_empty = inner.buf.is_empty();
-                    }
-                    Err(err) => {
-                        *dispatch_state = FramedState::Error(ServiceError::Service(err));
-                        return true;
-                    }
-                }
-            }
-
-            if !rx_done && rx.is_some() {
-                match Pin::new(rx.as_mut().unwrap()).poll_next(cx) {
-                    Poll::Ready(Some(FramedMessage::Message(msg))) => {
-                        if let Err(err) = framed.write(msg) {
-                            *dispatch_state =
-                                FramedState::FramedError(ServiceError::Encoder(err));
-                            return true;
-                        }
-                    }
-                    Poll::Ready(Some(FramedMessage::Close)) => {
-                        dispatch_state.stop(None);
-                        return true;
-                    }
-                    Poll::Ready(Some(FramedMessage::WaitClose(tx))) => {
-                        dispatch_state.stop(Some(tx));
-                        return true;
-                    }
-                    Poll::Ready(None) => {
-                        rx_done = true;
-                        let _ = rx.take();
-                    }
-                    Poll::Pending => rx_done = true,
-                }
-            }
-
-            if rx_done && buf_empty {
-                break;
-            }
-        }
-
-        if !framed.is_write_buf_empty() {
-            match framed.flush(cx) {
-                Poll::Pending => break,
-                Poll::Ready(Err(err)) => {
-                    debug!("Error sending data: {:?}", err);
-                    *dispatch_state = FramedState::FramedError(ServiceError::Encoder(err));
-                    return true;
-                }
-                Poll::Ready(_) => (),
-            }
-        } else {
-            break;
-        }
-    }
-
-    false
-}
--- a/actix-ioframe/src/item.rs
+++ b/actix-ioframe/src/item.rs
@@ -5,19 +5,19 @@ use actix_codec::{Decoder, Encoder};

 use crate::sink::Sink;

-pub struct Item<St, Codec: Encoder + Decoder> {
+pub struct Item<St, Codec: Encoder + Decoder, E> {
    state: St,
-    sink: Sink<<Codec as Encoder>::Item>,
+    sink: Sink<<Codec as Encoder>::Item, E>,
    item: <Codec as Decoder>::Item,
 }

-impl<St, Codec> Item<St, Codec>
+impl<St, Codec, E> Item<St, Codec, E>
 where
    Codec: Encoder + Decoder,
 {
    pub(crate) fn new(
        state: St,
-        sink: Sink<<Codec as Encoder>::Item>,
+        sink: Sink<<Codec as Encoder>::Item, E>,
        item: <Codec as Decoder>::Item,
    ) -> Self {
        Item { state, sink, item }
@@ -34,7 +34,7 @@ where
    }

    #[inline]
-    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
+    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, E> {
        &self.sink
    }

@@ -44,12 +44,18 @@ where
    }

    #[inline]
-    pub fn into_parts(self) -> (St, Sink<<Codec as Encoder>::Item>, <Codec as Decoder>::Item) {
+    pub fn into_parts(
+        self,
+    ) -> (
+        St,
+        Sink<<Codec as Encoder>::Item, E>,
+        <Codec as Decoder>::Item,
+    ) {
        (self.state, self.sink, self.item)
    }
 }

-impl<St, Codec> Deref for Item<St, Codec>
+impl<St, Codec, E> Deref for Item<St, Codec, E>
 where
    Codec: Encoder + Decoder,
 {
@@ -61,7 +67,7 @@ where
    }
 }

-impl<St, Codec> DerefMut for Item<St, Codec>
+impl<St, Codec, E> DerefMut for Item<St, Codec, E>
 where
    Codec: Encoder + Decoder,
 {
@@ -71,12 +77,12 @@ where
    }
 }

-impl<St, Codec> fmt::Debug for Item<St, Codec>
+impl<St, Codec, E> fmt::Debug for Item<St, Codec, E>
 where
    Codec: Encoder + Decoder,
    <Codec as Decoder>::Item: fmt::Debug,
 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_tuple("FramedItem").field(&self.item).finish()
+        f.debug_tuple("Item").field(&self.item).finish()
    }
 }
--- a/actix-ioframe/src/lib.rs
+++ b/actix-ioframe/src/lib.rs
@@ -1,7 +1,6 @@
 #![deny(rust_2018_idioms, warnings)]
 #![allow(clippy::type_complexity, clippy::too_many_arguments)]

-mod cell;
 mod connect;
 mod dispatcher;
 mod error;
--- a/actix-ioframe/src/service.rs
+++ b/actix-ioframe/src/service.rs
@@ -6,17 +6,20 @@ use std::task::{Context, Poll};

 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder};
 use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
+use actix_utils::mpsc;
 use either::Either;
 use futures::future::{FutureExt, LocalBoxFuture};
 use pin_project::project;

 use crate::connect::{Connect, ConnectResult};
-use crate::dispatcher::FramedDispatcher;
+use crate::dispatcher::{Dispatcher, Message};
 use crate::error::ServiceError;
 use crate::item::Item;
+use crate::sink::Sink;

-type RequestItem<S, U> = Item<S, U>;
+type RequestItem<S, U, E> = Item<S, U, E>;
 type ResponseItem<U> = Option<<U as Encoder>::Item>;
+type ServiceResult<U, E> = Result<Message<<U as Encoder>::Item>, E>;

 /// Service builder - structure that follows the builder pattern
 /// for building instances for framed services.
@@ -34,11 +37,15 @@ impl<St: Clone, Codec> Builder<St, Codec> {
    }

    /// Construct framed handler service with specified connect service
-    pub fn service<Io, C, F>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec>
+    pub fn service<Io, C, F, E>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec, E>
    where
        F: IntoService<C>,
        Io: AsyncRead + AsyncWrite,
-        C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
+        C: Service<
+            Request = Connect<Io, Codec, E>,
+            Response = ConnectResult<Io, St, Codec, E>,
+            Error = E,
+        >,
        Codec: Decoder + Encoder,
    {
        ServiceBuilder {
@@ -49,16 +56,17 @@ impl<St: Clone, Codec> Builder<St, Codec> {
    }

    /// Construct framed handler new service with specified connect service
-    pub fn factory<Io, C, F>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec>
+    pub fn factory<Io, C, F, E>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec, E>
    where
        F: IntoServiceFactory<C>,
+        E: 'static,
        Io: AsyncRead + AsyncWrite,
        C: ServiceFactory<
            Config = (),
-            Request = Connect<Io>,
-            Response = ConnectResult<Io, St, Codec>,
+            Request = Connect<Io, Codec, E>,
+            Response = ConnectResult<Io, St, Codec, E>,
+            Error = E,
        >,
-        C::Error: 'static,
        C::Future: 'static,
        Codec: Decoder + Encoder,
    {
@@ -70,17 +78,20 @@ impl<St: Clone, Codec> Builder<St, Codec> {
    }
 }

-pub struct ServiceBuilder<St, C, Io, Codec> {
+pub struct ServiceBuilder<St, C, Io, Codec, Err> {
    connect: C,
    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
-    _t: PhantomData<(St, Io, Codec)>,
+    _t: PhantomData<(St, Io, Codec, Err)>,
 }

-impl<St, C, Io, Codec> ServiceBuilder<St, C, Io, Codec>
+impl<St, C, Io, Codec, Err> ServiceBuilder<St, C, Io, Codec, Err>
 where
    St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
+    >,
    Io: AsyncRead + AsyncWrite,
    Codec: Decoder + Encoder,
    <Codec as Encoder>::Item: 'static,
@@ -98,16 +109,16 @@ where
    }

    /// Provide stream items handler service and construct service factory.
-    pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec>
+    pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec, Err>
    where
        F: IntoServiceFactory<T>,
        T: ServiceFactory<
-                Config = St,
-                Request = RequestItem<St, Codec>,
-                Response = ResponseItem<Codec>,
-                Error = C::Error,
-                InitError = C::Error,
-            > + 'static,
+            Config = St,
+            Request = RequestItem<St, Codec, Err>,
+            Response = ResponseItem<Codec>,
+            Error = Err,
+            InitError = Err,
+        >,
    {
        FramedServiceImpl {
            connect: self.connect,
@@ -118,22 +129,23 @@ where
    }
 }

-pub struct NewServiceBuilder<St, C, Io, Codec> {
+pub struct NewServiceBuilder<St, C, Io, Codec, Err> {
    connect: C,
    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
-    _t: PhantomData<(St, Io, Codec)>,
+    _t: PhantomData<(St, Io, Codec, Err)>,
 }

-impl<St, C, Io, Codec> NewServiceBuilder<St, C, Io, Codec>
+impl<St, C, Io, Codec, Err> NewServiceBuilder<St, C, Io, Codec, Err>
 where
    St: Clone,
    Io: AsyncRead + AsyncWrite,
+    Err: 'static,
    C: ServiceFactory<
        Config = (),
-        Request = Connect<Io>,
-        Response = ConnectResult<Io, St, Codec>,
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
    >,
-    C::Error: 'static,
    C::Future: 'static,
    Codec: Decoder + Encoder,
    <Codec as Encoder>::Item: 'static,
@@ -150,15 +162,15 @@ where
        self
    }

-    pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Cfg>
+    pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Err, Cfg>
    where
        F: IntoServiceFactory<T>,
        T: ServiceFactory<
                Config = St,
-                Request = RequestItem<St, Codec>,
+                Request = RequestItem<St, Codec, Err>,
                Response = ResponseItem<Codec>,
-                Error = C::Error,
-                InitError = C::Error,
+                Error = Err,
+                InitError = Err,
            > + 'static,
    {
        FramedService {
@@ -170,32 +182,34 @@ where
    }
 }

-pub struct FramedService<St, C, T, Io, Codec, Cfg> {
+pub struct FramedService<St, C, T, Io, Codec, Err, Cfg> {
    connect: C,
    handler: Rc<T>,
    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
-    _t: PhantomData<(St, Io, Codec, Cfg)>,
+    _t: PhantomData<(St, Io, Codec, Err, Cfg)>,
 }

-impl<St, C, T, Io, Codec, Cfg> ServiceFactory for FramedService<St, C, T, Io, Codec, Cfg>
+impl<St, C, T, Io, Codec, Err, Cfg> ServiceFactory
+    for FramedService<St, C, T, Io, Codec, Err, Cfg>
 where
    St: Clone + 'static,
    Io: AsyncRead + AsyncWrite,
    C: ServiceFactory<
        Config = (),
-        Request = Connect<Io>,
-        Response = ConnectResult<Io, St, Codec>,
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
    >,
-    C::Error: 'static,
    C::Future: 'static,
    T: ServiceFactory<
            Config = St,
-            Request = RequestItem<St, Codec>,
+            Request = RequestItem<St, Codec, Err>,
            Response = ResponseItem<Codec>,
-            Error = C::Error,
-            InitError = C::Error,
+            Error = Err,
+            InitError = Err,
        > + 'static,
    <T::Service as Service>::Future: 'static,
+    Err: 'static,
    Codec: Decoder + Encoder,
    <Codec as Encoder>::Item: 'static,
    <Codec as Encoder>::Error: std::fmt::Debug,
@@ -205,7 +219,7 @@ where
    type Response = ();
    type Error = ServiceError<C::Error, Codec>;
    type InitError = C::InitError;
-    type Service = FramedServiceImpl<St, C::Service, T, Io, Codec>;
+    type Service = FramedServiceImpl<St, C::Service, T, Io, Codec, Err>;
    type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;

    fn new_service(&self, _: Cfg) -> Self::Future {
@@ -227,25 +241,29 @@ where
    }
 }

-pub struct FramedServiceImpl<St, C, T, Io, Codec> {
+pub struct FramedServiceImpl<St, C, T, Io, Codec, Err> {
    connect: C,
    handler: Rc<T>,
    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
-    _t: PhantomData<(St, Io, Codec)>,
+    _t: PhantomData<(St, Io, Codec, Err)>,
 }

-impl<St, C, T, Io, Codec> Service for FramedServiceImpl<St, C, T, Io, Codec>
+impl<St, C, T, Io, Codec, Err> Service for FramedServiceImpl<St, C, T, Io, Codec, Err>
 where
    St: Clone,
    Io: AsyncRead + AsyncWrite,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
+    >,
+    Err: 'static,
    T: ServiceFactory<
        Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<St, Codec, Err>,
        Response = ResponseItem<Codec>,
-        Error = C::Error,
-        InitError = C::Error,
+        Error = Err,
+        InitError = Err,
    >,
    <T::Service as Service>::Future: 'static,
    Codec: Decoder + Encoder,
@@ -254,36 +272,43 @@ where
 {
    type Request = Io;
    type Response = ();
-    type Error = ServiceError<C::Error, Codec>;
-    type Future = FramedServiceImplResponse<St, Io, Codec, C, T>;
+    type Error = ServiceError<Err, Codec>;
+    type Future = FramedServiceImplResponse<St, Io, Codec, Err, C, T>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.connect.poll_ready(cx).map_err(|e| e.into())
    }

    fn call(&mut self, req: Io) -> Self::Future {
+        let (tx, rx) = mpsc::channel();
+        let sink = Sink::new(tx);
        FramedServiceImplResponse {
            inner: FramedServiceImplResponseInner::Connect(
-                self.connect.call(Connect::new(req)),
+                self.connect.call(Connect::new(req, sink.clone())),
                self.handler.clone(),
                self.disconnect.clone(),
+                Some(rx),
            ),
        }
    }
 }

 #[pin_project::pin_project]
-pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
+pub struct FramedServiceImplResponse<St, Io, Codec, Err, C, T>
 where
    St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
+    >,
+    Err: 'static,
    T: ServiceFactory<
        Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<St, Codec, Err>,
        Response = ResponseItem<Codec>,
-        Error = C::Error,
-        InitError = C::Error,
+        Error = Err,
+        InitError = Err,
    >,
    <T::Service as Service>::Future: 'static,
    Io: AsyncRead + AsyncWrite,
@@ -292,20 +317,24 @@ where
    <Codec as Encoder>::Error: std::fmt::Debug,
 {
    #[pin]
-    inner: FramedServiceImplResponseInner<St, Io, Codec, C, T>,
+    inner: FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>,
 }

-impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
+impl<St, Io, Codec, Err, C, T> Future for FramedServiceImplResponse<St, Io, Codec, Err, C, T>
 where
    St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
+    >,
+    Err: 'static,
    T: ServiceFactory<
        Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<St, Codec, Err>,
        Response = ResponseItem<Codec>,
-        Error = C::Error,
-        InitError = C::Error,
+        Error = Err,
+        InitError = Err,
    >,
    <T::Service as Service>::Future: 'static,
    Io: AsyncRead + AsyncWrite,
@@ -313,7 +342,7 @@ where
    <Codec as Encoder>::Item: 'static,
    <Codec as Encoder>::Error: std::fmt::Debug,
 {
-    type Output = Result<(), ServiceError<C::Error, Codec>>;
+    type Output = Result<(), ServiceError<Err, Codec>>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut this = self.as_mut().project();
@@ -331,17 +360,21 @@ where
 }

 #[pin_project::pin_project]
-enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
+enum FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>
 where
    St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
+    >,
+    Err: 'static,
    T: ServiceFactory<
        Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<St, Codec, Err>,
        Response = ResponseItem<Codec>,
-        Error = C::Error,
-        InitError = C::Error,
+        Error = Err,
+        InitError = Err,
    >,
    <T::Service as Service>::Future: 'static,
    Io: AsyncRead + AsyncWrite,
@@ -349,26 +382,36 @@ where
    <Codec as Encoder>::Item: 'static,
    <Codec as Encoder>::Error: std::fmt::Debug,
 {
-    Connect(#[pin] C::Future, Rc<T>, Option<Rc<dyn Fn(&mut St, bool)>>),
+    Connect(
+        #[pin] C::Future,
+        Rc<T>,
+        Option<Rc<dyn Fn(&mut St, bool)>>,
+        Option<mpsc::Receiver<ServiceResult<Codec, Err>>>,
+    ),
    Handler(
        #[pin] T::Future,
-        Option<ConnectResult<Io, St, Codec>>,
+        Option<ConnectResult<Io, St, Codec, Err>>,
        Option<Rc<dyn Fn(&mut St, bool)>>,
+        Option<mpsc::Receiver<ServiceResult<Codec, Err>>>,
    ),
-    Dispatcher(#[pin] FramedDispatcher<St, T::Service, Io, Codec>),
+    Dispatcher(Dispatcher<St, T::Service, Io, Codec, Err>),
 }

-impl<St, Io, Codec, C, T> FramedServiceImplResponseInner<St, Io, Codec, C, T>
+impl<St, Io, Codec, Err, C, T> FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>
 where
    St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<
+        Request = Connect<Io, Codec, Err>,
+        Response = ConnectResult<Io, St, Codec, Err>,
+        Error = Err,
+    >,
+    Err: 'static,
    T: ServiceFactory<
        Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<St, Codec, Err>,
        Response = ResponseItem<Codec>,
-        Error = C::Error,
-        InitError = C::Error,
+        Error = Err,
+        InitError = Err,
    >,
    <T::Service as Service>::Future: 'static,
    Io: AsyncRead + AsyncWrite,
@@ -381,42 +424,44 @@ where
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Either<
-        FramedServiceImplResponseInner<St, Io, Codec, C, T>,
-        Poll<Result<(), ServiceError<C::Error, Codec>>>,
+        FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>,
+        Poll<Result<(), ServiceError<Err, Codec>>>,
    > {
        #[project]
        match self.project() {
-            FramedServiceImplResponseInner::Connect(fut, handler, disconnect) => {
+            FramedServiceImplResponseInner::Connect(fut, handler, disconnect, rx) => {
                match fut.poll(cx) {
                    Poll::Ready(Ok(res)) => {
                        Either::Left(FramedServiceImplResponseInner::Handler(
                            handler.new_service(res.state.clone()),
                            Some(res),
                            disconnect.take(),
+                            rx.take(),
                        ))
                    }
                    Poll::Pending => Either::Right(Poll::Pending),
                    Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
                }
            }
-            FramedServiceImplResponseInner::Handler(fut, res, disconnect) => match fut.poll(cx)
-            {
-                Poll::Ready(Ok(handler)) => {
-                    let res = res.take().unwrap();
-                    Either::Left(FramedServiceImplResponseInner::Dispatcher(
-                        FramedDispatcher::new(
-                            res.framed,
-                            res.state,
-                            handler,
-                            res.rx,
-                            res.sink,
-                            disconnect.take(),
-                        ),
-                    ))
+            FramedServiceImplResponseInner::Handler(fut, res, disconnect, rx) => {
+                match fut.poll(cx) {
+                    Poll::Ready(Ok(handler)) => {
+                        let res = res.take().unwrap();
+                        Either::Left(FramedServiceImplResponseInner::Dispatcher(
+                            Dispatcher::new(
+                                res.framed,
+                                res.state,
+                                handler,
+                                res.sink,
+                                rx.take().unwrap(),
+                                disconnect.take(),
+                            ),
+                        ))
+                    }
+                    Poll::Pending => Either::Right(Poll::Pending),
+                    Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
                }
-                Poll::Pending => Either::Right(Poll::Pending),
-                Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
-            },
+            }
            FramedServiceImplResponseInner::Dispatcher(ref mut fut) => {
                Either::Right(fut.poll(cx))
            }
--- a/actix-ioframe/src/sink.rs
+++ b/actix-ioframe/src/sink.rs
@@ -3,41 +3,41 @@ use std::fmt;
 use actix_utils::{mpsc, oneshot};
 use futures::future::{Future, FutureExt};

-use crate::dispatcher::FramedMessage;
+use crate::dispatcher::Message;

-pub struct Sink<T>(mpsc::Sender<FramedMessage<T>>);
+pub struct Sink<T, E>(mpsc::Sender<Result<Message<T>, E>>);

-impl<T> Clone for Sink<T> {
+impl<T, E> Clone for Sink<T, E> {
    fn clone(&self) -> Self {
        Sink(self.0.clone())
    }
 }

-impl<T> Sink<T> {
-    pub(crate) fn new(tx: mpsc::Sender<FramedMessage<T>>) -> Self {
+impl<T, E> Sink<T, E> {
+    pub(crate) fn new(tx: mpsc::Sender<Result<Message<T>, E>>) -> Self {
        Sink(tx)
    }

    /// Close connection
    pub fn close(&self) {
-        let _ = self.0.send(FramedMessage::Close);
+        let _ = self.0.send(Ok(Message::Close));
    }

    /// Close connection
    pub fn wait_close(&self) -> impl Future<Output = ()> {
        let (tx, rx) = oneshot::channel();
-        let _ = self.0.send(FramedMessage::WaitClose(tx));
+        let _ = self.0.send(Ok(Message::WaitClose(tx)));

        rx.map(|_| ())
    }

    /// Send item
    pub fn send(&self, item: T) {
-        let _ = self.0.send(FramedMessage::Message(item));
+        let _ = self.0.send(Ok(Message::Item(item)));
    }
 }

-impl<T> fmt::Debug for Sink<T> {
+impl<T, E> fmt::Debug for Sink<T, E> {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("Sink").finish()
    }
--- a/actix-ioframe/tests/test_server.rs
+++ b/actix-ioframe/tests/test_server.rs
@@ -4,7 +4,7 @@ use std::time::Duration;

 use actix_codec::BytesCodec;
 use actix_rt::time::delay_for;
-use actix_service::{service_fn, Service};
+use actix_service::{fn_service, Service};
 use actix_testing::TestServer;
 use futures::future::ok;

@@ -22,22 +22,22 @@ async fn test_disconnect() -> std::io::Result<()> {
        let disconnect1 = disconnect1.clone();

        Builder::new()
-            .factory(service_fn(|conn: Connect<_>| {
+            .factory(fn_service(|conn: Connect<_, _, _>| {
                ok(conn.codec(BytesCodec).state(State))
            }))
            .disconnect(move |_, _| {
                disconnect1.store(true, Ordering::Relaxed);
            })
-            .finish(service_fn(|_t| ok(None)))
+            .finish(fn_service(|_t| ok(None)))
    });

    let mut client = Builder::new()
-        .service(|conn: Connect<_>| {
+        .service(|conn: Connect<_, _, _>| {
            let conn = conn.codec(BytesCodec).state(State);
            conn.sink().close();
            ok(conn)
        })
-        .finish(service_fn(|_t| ok(None)));
+        .finish(fn_service(|_t| ok(None)));

    let conn = actix_connect::default_connector()
        .call(actix_connect::Connect::with(String::new(), srv.addr()))
--- a/actix-macros/Cargo.toml
+++ b/actix-macros/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-macros"
-version = "0.1.0-alpha.1"
+version = "0.1.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime macros"
 repository = "https://github.com/actix/actix-net"
@@ -18,4 +18,4 @@ quote = "^1"
 syn = { version = "^1", features = ["full"] }

 [dev-dependencies]
-actix-rt = { version = "1.0.0-alpha.1" }
+actix-rt = { version = "1.0.0" }
--- a/actix-macros/src/lib.rs
+++ b/actix-macros/src/lib.rs
@@ -17,25 +17,26 @@ use quote::quote;
 #[proc_macro_attribute]
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main(_: TokenStream, item: TokenStream) -> TokenStream {
-    let input = syn::parse_macro_input!(item as syn::ItemFn);
-
-    let ret = &input.sig.output;
-    let name = &input.sig.ident;
-    let inputs = &input.sig.inputs;
-    let body = &input.block;
+    let mut input = syn::parse_macro_input!(item as syn::ItemFn);
    let attrs = &input.attrs;
+    let vis = &input.vis;
+    let sig = &mut input.sig;
+    let body = &input.block;
+    let name = &sig.ident;

-    if input.sig.asyncness.is_none() {
-        return syn::Error::new_spanned(input.sig.fn_token, "only async fn is supported")
+    if sig.asyncness.is_none() {
+        return syn::Error::new_spanned(sig.fn_token, "only async fn is supported")
            .to_compile_error()
            .into();
    }

+    sig.asyncness = None;
+
    (quote! {
        #(#attrs)*
-        fn #name(#inputs) #ret {
-            actix_rt::System::new("main")
-                .block_on(async { #body })
+        #vis #sig {
+            actix_rt::System::new(stringify!(#name))
+                .block_on(async move { #body })
        }
    })
    .into()
--- a/actix-rt/CHANGES.md
+++ b/actix-rt/CHANGES.md
@@ -1,5 +1,9 @@
 # Changes

+## [1.0.0] - 2019-12-11
+
+* Update dependencies
+
 ## [1.0.0-alpha.3] - 2019-12-07

 ### Fixed
--- a/actix-rt/Cargo.toml
+++ b/actix-rt/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-rt"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,6 @@ 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 = ".."

@@ -18,7 +17,7 @@ name = "actix_rt"
 path = "src/lib.rs"

 [dependencies]
-actix-macros = "0.1.0-alpha.1"
+actix-macros = "0.1.0"
 actix-threadpool = "0.3"
 futures = "0.3.1"
 copyless = "0.1.4"
--- a/actix-server/CHANGES.md
+++ b/actix-server/CHANGES.md
@@ -1,5 +1,18 @@
 # Changes

+## [1.0.0] - 2019-12-11
+
+### Changed
+
+* Use actix-net releases
+
+
+## [1.0.0-alpha.4] - 2019-12-08
+
+### Changed
+
+* Use actix-service 1.0.0-alpha.4
+
 ## [1.0.0-alpha.3] - 2019-12-07

 ### Changed
--- a/actix-server/Cargo.toml
+++ b/actix-server/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-server"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix server - General purpose tcp server"
 keywords = ["network", "framework", "async", "futures"]
@@ -21,13 +21,13 @@ path = "src/lib.rs"
 default = []

 [dependencies]
-actix-service = "1.0.0-alpha.3"
-actix-rt = "1.0.0-alpha.3"
-actix-codec = "0.2.0-alpha.3"
-actix-utils = "1.0.0-alpha.3"
+actix-service = "1.0.0"
+actix-rt = "1.0.0"
+actix-codec = "0.2.0"
+actix-utils = "1.0.0"

 log = "0.4"
-num_cpus = "1.0"
+num_cpus = "1.11"
 mio = "0.6.19"
 net2 = "0.2"
 futures = "0.3.1"
--- a/actix-server/src/config.rs
+++ b/actix-server/src/config.rs
@@ -142,7 +142,7 @@ impl InternalServiceFactory for ConfiguredService {
                    let name = names.remove(&token).unwrap().0;
                    res.push((
                        token,
-                        Box::new(StreamService::new(actix::service_fn2(
+                        Box::new(StreamService::new(actix::fn_service(
                            move |_: TcpStream| {
                                error!("Service {:?} is not configured", name);
                                ok::<_, ()>(())
--- a/actix-server/tests/test_server.rs
+++ b/actix-server/tests/test_server.rs
@@ -6,7 +6,7 @@ use std::{net, thread, time};
 use actix_codec::{BytesCodec, Framed};
 use actix_rt::net::TcpStream;
 use actix_server::Server;
-use actix_service::service_fn;
+use actix_service::fn_service;
 use bytes::Bytes;
 use futures::future::{lazy, ok};
 use futures::SinkExt;
@@ -31,7 +31,7 @@ fn test_bind() {
        let srv = Server::build()
            .workers(1)
            .disable_signals()
-            .bind("test", addr, move || service_fn(|_| ok::<_, ()>(())))
+            .bind("test", addr, move || fn_service(|_| ok::<_, ()>(())))
            .unwrap()
            .start();
        let _ = tx.send((srv, actix_rt::System::current()));
@@ -56,7 +56,7 @@ fn test_listen() {
        Server::build()
            .disable_signals()
            .workers(1)
-            .listen("test", lst, move || service_fn(|_| ok::<_, ()>(())))
+            .listen("test", lst, move || fn_service(|_| ok::<_, ()>(())))
            .unwrap()
            .start();
        let _ = tx.send(actix_rt::System::current());
@@ -82,7 +82,7 @@ fn test_start() {
            .backlog(100)
            .disable_signals()
            .bind("test", addr, move || {
-                service_fn(|io: TcpStream| {
+                fn_service(|io: TcpStream| {
                    async move {
                        let mut f = Framed::new(io, BytesCodec);
                        f.send(Bytes::from_static(b"test")).await.unwrap();
@@ -158,8 +158,8 @@ fn test_configure() {
                    .listen("addr3", lst)
                    .apply(move |rt| {
                        let num = num.clone();
-                        rt.service("addr1", service_fn(|_| ok::<_, ()>(())));
-                        rt.service("addr3", service_fn(|_| ok::<_, ()>(())));
+                        rt.service("addr1", fn_service(|_| ok::<_, ()>(())));
+                        rt.service("addr3", fn_service(|_| ok::<_, ()>(())));
                        rt.on_start(lazy(move |_| {
                            let _ = num.fetch_add(1, Relaxed);
                        }))
--- a/actix-service/CHANGES.md
+++ b/actix-service/CHANGES.md
@@ -1,5 +1,23 @@
 # Changes

+## [1.0.0] - 2019-12-11
+
+### Added
+
+* Add Clone impl for Apply service
+
+
+## [1.0.0-alpha.4] - 2019-12-08
+
+### Changed
+
+* Renamed `service_fn` to `fn_service`
+
+* Renamed `factory_fn` to `fn_factory`
+
+* Renamed `factory_fn_cfg` to `fn_factory_with_config`
+
+
 ## [1.0.0-alpha.3] - 2019-12-06

 ### Changed
--- a/actix-service/Cargo.toml
+++ b/actix-service/Cargo.toml
@@ -1,15 +1,14 @@
 [package]
 name = "actix-service"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
-description = "Actix Service"
+description = "Actix service"
 keywords = ["network", "framework", "async", "futures"]
 homepage = "https://actix.rs"
 repository = "https://github.com/actix/actix-net.git"
 documentation = "https://docs.rs/actix-service/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
-exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
 workspace = ".."

@@ -27,4 +26,4 @@ futures-util = "0.3.1"
 pin-project = "0.4.6"

 [dev-dependencies]
-actix-rt = "1.0.0-alpha.2"
+actix-rt = "1.0.0"
--- a/actix-service/src/and_then.rs
+++ b/actix-service/src/and_then.rs
@@ -258,7 +258,7 @@ mod tests {

    use futures_util::future::{lazy, ok, ready, Ready};

-    use crate::{factory_fn, pipeline, pipeline_factory, Service, ServiceFactory};
+    use crate::{fn_factory, pipeline, pipeline_factory, Service, ServiceFactory};

    struct Srv1(Rc<Cell<usize>>);

@@ -320,7 +320,7 @@ mod tests {
        let cnt = Rc::new(Cell::new(0));
        let cnt2 = cnt.clone();
        let new_srv =
-            pipeline_factory(factory_fn(move || ready(Ok::<_, ()>(Srv1(cnt2.clone())))))
+            pipeline_factory(fn_factory(move || ready(Ok::<_, ()>(Srv1(cnt2.clone())))))
                .and_then(move || ready(Ok(Srv2(cnt.clone()))));

        let mut srv = new_srv.new_service(()).await.unwrap();
--- a/actix-service/src/and_then_apply_fn.rs
+++ b/actix-service/src/and_then_apply_fn.rs
@@ -283,7 +283,7 @@ mod tests {

    use futures_util::future::{lazy, ok, Ready, TryFutureExt};

-    use crate::{pipeline, pipeline_factory, service_fn2, Service, ServiceFactory};
+    use crate::{fn_service, pipeline, pipeline_factory, Service, ServiceFactory};

    #[derive(Clone)]
    struct Srv;
@@ -318,7 +318,7 @@ mod tests {

    #[actix_rt::test]
    async fn test_service_factory() {
-        let new_srv = pipeline_factory(|| ok::<_, ()>(service_fn2(|r: &'static str| ok(r))))
+        let new_srv = pipeline_factory(|| ok::<_, ()>(fn_service(|r: &'static str| ok(r))))
            .and_then_apply_fn(
                || ok(Srv),
                |req: &'static str, s| s.call(()).map_ok(move |res| (req, res)),
--- a/actix-service/src/apply.rs
+++ b/actix-service/src/apply.rs
@@ -5,7 +5,7 @@ use std::task::{Context, Poll};

 use super::{IntoService, IntoServiceFactory, Service, ServiceFactory};

-/// Apply tranform function to a service
+/// Apply tranform function to a service.
 pub fn apply_fn<T, F, R, In, Out, Err, U>(service: U, f: F) -> Apply<T, F, R, In, Out, Err>
 where
    T: Service<Error = Err>,
@@ -16,7 +16,7 @@ where
    Apply::new(service.into_service(), f)
 }

-/// Create factory for `apply` service.
+/// Service factory that prodices `apply_fn` service.
 pub fn apply_fn_factory<T, F, R, In, Out, Err, U>(
    service: U,
    f: F,
@@ -56,6 +56,21 @@ where
    }
 }

+impl<T, F, R, In, Out, Err> Clone for Apply<T, F, R, In, Out, Err>
+where
+    T: Service<Error = Err> + Clone,
+    F: FnMut(In, &mut T) -> R + Clone,
+    R: Future<Output = Result<Out, Err>>,
+{
+    fn clone(&self) -> Self {
+        Apply {
+            service: self.service.clone(),
+            f: self.f.clone(),
+            r: PhantomData,
+        }
+    }
+}
+
 impl<T, F, R, In, Out, Err> Service for Apply<T, F, R, In, Out, Err>
 where
    T: Service<Error = Err>,
--- a/actix-service/src/apply_cfg.rs
+++ b/actix-service/src/apply_cfg.rs
@@ -6,7 +6,7 @@ use std::task::{Context, Poll};
 use crate::cell::Cell;
 use crate::{Service, ServiceFactory};

-/// Convert `Fn(&Config, &mut Service) -> Future<Service>` fn to a NewService
+/// Convert `Fn(Config, &mut Service1) -> Future<Service2>` fn to a service factory
 pub fn apply_cfg<F, C, T, R, S, E>(srv: T, f: F) -> ApplyConfigService<F, C, T, R, S, E>
 where
    F: FnMut(C, &mut T) -> R,
@@ -20,10 +20,11 @@ where
    }
 }

-/// Convert `Fn(&Config, &mut Service) -> Future<Service>` fn to a NewService
-/// Service get constructor from NewService.
+/// Convert `Fn(Config, &mut Service1) -> Future<Service2>` fn to a service factory
+///
+/// Service1 get constructed from `T` factory.
 pub fn apply_cfg_factory<F, C, T, R, S>(
-    srv: T,
+    factory: T,
    f: F,
 ) -> ApplyConfigServiceFactory<F, C, T, R, S>
 where
@@ -34,12 +35,12 @@ where
    S: Service,
 {
    ApplyConfigServiceFactory {
-        srv: Cell::new((srv, f)),
+        srv: Cell::new((factory, f)),
        _t: PhantomData,
    }
 }

-/// Convert `Fn(&Config) -> Future<Service>` fn to NewService\
+/// Convert `Fn(Config, &mut Server) -> Future<Service>` fn to NewService\
 pub struct ApplyConfigService<F, C, T, R, S, E>
 where
    F: FnMut(C, &mut T) -> R,
--- a/actix-service/src/boxed.rs
+++ b/actix-service/src/boxed.rs
@@ -13,7 +13,7 @@ pub type BoxService<Req, Res, Err> =

 pub struct BoxServiceFactory<C, Req, Res, Err, InitErr>(Inner<C, Req, Res, Err, InitErr>);

-/// Create boxed new service
+/// Create boxed service factory
 pub fn factory<T>(
    factory: T,
 ) -> BoxServiceFactory<T::Config, T::Request, T::Response, T::Error, T::InitError>
--- a/actix-service/src/fn_service.rs
+++ b/actix-service/src/fn_service.rs
@@ -7,7 +7,7 @@ use futures_util::future::{ok, Ready};
 use crate::{IntoService, IntoServiceFactory, Service, ServiceFactory};

 /// Create `ServiceFactory` for function that can act as a `Service`
-pub fn service_fn<F, Fut, Req, Res, Err, Cfg>(
+pub fn fn_service<F, Fut, Req, Res, Err, Cfg>(
    f: F,
 ) -> FnServiceFactory<F, Fut, Req, Res, Err, Cfg>
 where
@@ -17,16 +17,43 @@ where
    FnServiceFactory::new(f)
 }

-pub fn service_fn2<F, Fut, Req, Res, Err>(f: F) -> FnService<F, Fut, Req, Res, Err>
-where
-    F: FnMut(Req) -> Fut,
-    Fut: Future<Output = Result<Res, Err>>,
-{
-    FnService::new(f)
-}
-
 /// Create `ServiceFactory` for function that can produce services
-pub fn factory_fn<F, Cfg, Srv, Fut, Err>(f: F) -> FnServiceNoConfig<F, Cfg, Srv, Fut, Err>
+///
+/// # Example
+///
+/// ```rust
+/// use std::io;
+/// use actix_service::{fn_factory, fn_service, Service, ServiceFactory};
+/// use futures_util::future::ok;
+///
+/// /// Service that divides two usize values.
+/// async fn div((x, y): (usize, usize)) -> Result<usize, io::Error> {
+///     if y == 0 {
+///         Err(io::Error::new(io::ErrorKind::Other, "divide by zdro"))
+///     } else {
+///         Ok(x / y)
+///     }
+/// }
+///
+/// #[actix_rt::main]
+/// async fn main() -> io::Result<()> {
+///     // Create service factory that produces `div` services
+///     let factory = fn_factory(|| {
+///         ok::<_, io::Error>(fn_service(div))
+///     });
+///
+///     // construct new service
+///     let mut srv = factory.new_service(()).await?;
+///
+///     // now we can use `div` service
+///     let result = srv.call((10, 20)).await?;
+///
+///     println!("10 / 20 = {}", result);
+///
+///     Ok(())
+/// }
+/// ```
+pub fn fn_factory<F, Cfg, Srv, Fut, Err>(f: F) -> FnServiceNoConfig<F, Cfg, Srv, Fut, Err>
 where
    Srv: Service,
    F: Fn() -> Fut,
@@ -35,8 +62,39 @@ where
    FnServiceNoConfig::new(f)
 }

-/// Create `ServiceFactory` for function that can produce services with configuration
-pub fn factory_fn_cfg<F, Fut, Cfg, Srv, Err>(f: F) -> FnServiceConfig<F, Fut, Cfg, Srv, Err>
+/// Create `ServiceFactory` for function that accepts config argument and can produce services
+///
+/// Any function that has following form `Fn(Config) -> Future<Output = Service>` could
+/// act as a `ServiceFactory`.
+///
+/// # Example
+///
+/// ```rust
+/// use std::io;
+/// use actix_service::{fn_factory_with_config, fn_service, Service, ServiceFactory};
+/// use futures_util::future::ok;
+///
+/// #[actix_rt::main]
+/// async fn main() -> io::Result<()> {
+///     // Create service factory. factory uses config argument for
+///     // services it generates.
+///     let factory = fn_factory_with_config(|y: usize| {
+///         ok::<_, io::Error>(fn_service(move |x: usize| ok::<_, io::Error>(x * y)))
+///     });
+///
+///     // construct new service with config argument
+///     let mut srv = factory.new_service(10).await?;
+///
+///     let result = srv.call(10).await?;
+///     assert_eq!(result, 100);
+///
+///     println!("10 * 10 = {}", result);
+///     Ok(())
+/// }
+/// ```
+pub fn fn_factory_with_config<F, Fut, Cfg, Srv, Err>(
+    f: F,
+) -> FnServiceConfig<F, Fut, Cfg, Srv, Err>
 where
    F: Fn(Cfg) -> Fut,
    Fut: Future<Output = Result<Srv, Err>>,
@@ -132,6 +190,25 @@ where
    }
 }

+impl<F, Fut, Req, Res, Err> Service for FnServiceFactory<F, Fut, Req, Res, Err, ()>
+where
+    F: FnMut(Req) -> Fut + Clone,
+    Fut: Future<Output = Result<Res, Err>>,
+{
+    type Request = Req;
+    type Response = Res;
+    type Error = Err;
+    type Future = Fut;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: Self::Request) -> Self::Future {
+        (self.f)(req)
+    }
+}
+
 impl<F, Fut, Req, Res, Err, Cfg> ServiceFactory for FnServiceFactory<F, Fut, Req, Res, Err, Cfg>
 where
    F: FnMut(Req) -> Fut + Clone,
@@ -280,3 +357,47 @@ where
        FnServiceNoConfig::new(self)
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::task::Poll;
+
+    use futures_util::future::{lazy, ok};
+
+    use super::*;
+    use crate::{Service, ServiceFactory};
+
+    #[actix_rt::test]
+    async fn test_fn_service() {
+        let new_srv = fn_service(|()| ok::<_, ()>("srv"));
+
+        let mut srv = new_srv.new_service(()).await.unwrap();
+        let res = srv.call(()).await;
+        assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
+        assert!(res.is_ok());
+        assert_eq!(res.unwrap(), "srv");
+    }
+
+    #[actix_rt::test]
+    async fn test_fn_service_service() {
+        let mut srv = fn_service(|()| ok::<_, ()>("srv"));
+
+        let res = srv.call(()).await;
+        assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
+        assert!(res.is_ok());
+        assert_eq!(res.unwrap(), "srv");
+    }
+
+    #[actix_rt::test]
+    async fn test_fn_service_with_config() {
+        let new_srv = fn_factory_with_config(|cfg: usize| {
+            ok::<_, ()>(fn_service(move |()| ok::<_, ()>(("srv", cfg))))
+        });
+
+        let mut srv = new_srv.new_service(1).await.unwrap();
+        let res = srv.call(()).await;
+        assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
+        assert!(res.is_ok());
+        assert_eq!(res.unwrap(), ("srv", 1));
+    }
+}
--- a/actix-service/src/lib.rs
+++ b/actix-service/src/lib.rs
@@ -25,12 +25,48 @@ mod transform_err;

 pub use self::apply::{apply_fn, apply_fn_factory};
 pub use self::apply_cfg::{apply_cfg, apply_cfg_factory};
-pub use self::fn_service::{factory_fn, factory_fn_cfg, service_fn, service_fn2};
+pub use self::fn_service::{fn_factory, fn_factory_with_config, fn_service};
 pub use self::map_config::{map_config, unit_config};
 pub use self::pipeline::{pipeline, pipeline_factory, Pipeline, PipelineFactory};
 pub use self::transform::{apply, Transform};

 /// An asynchronous function from `Request` to a `Response`.
+///
+/// `Service` represents a service that represanting interation, taking requests and giving back
+/// replies. You can think about service as a function with one argument and result as a return
+/// type. In general form it looks like `async fn(Req) -> Result<Res, Err>`. `Service`
+/// trait just generalizing form of this function. Each parameter described as an assotiated type.
+///
+/// Services provides a symmetric and uniform API, same abstractions represents
+/// clients and servers. Services describe only `transforamtion` operation
+/// which encorouge to simplify api surface and phrases `value transformation`.
+/// That leads to simplier design of each service. That also allows better testability
+/// and better composition.
+///
+/// Services could be represented in several different forms. In general,
+/// Service is a type that implements `Service` trait.
+///
+/// ```rust,ignore
+/// struct MyService;
+///
+/// impl Service for MyService {
+///      type Request = u8;
+///      type Response = u64;
+///      type Error = MyError;
+///      type Future = Pin<Box<Future<Output=Result<Self::Response, Self::Error>>>;
+///
+///      fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { ... }
+///
+///      fn call(&mut self) -> Self::Future { ... }
+/// }
+/// ```
+///
+/// Service can have mutable state that influence computation.
+/// This service could be rewritten as a simple function:
+///
+/// ```rust,ignore
+/// async fn my_service(req: u8) -> Result<u64, MyError>;
+/// ```
 pub trait Service {
    /// Requests handled by the service.
    type Request;
@@ -53,6 +89,12 @@ pub trait Service {
    /// This is a **best effort** implementation. False positives are permitted.
    /// It is permitted for the service to return `Ready` from a `poll_ready`
    /// call and the next invocation of `call` results in an error.
+    ///
+    /// There are several notes to consider:
+    ///
+    /// 1. `.poll_ready()` might be called on different task from actual service call.
+    ///
+    /// 2. In case of chained services, `.poll_ready()` get called for all services at once.
    fn poll_ready(&mut self, ctx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>>;

    /// Process the request and return the response asynchronously.
@@ -287,7 +329,7 @@ pub trait IntoServiceFactory<T>
 where
    T: ServiceFactory,
 {
-    /// Convert `Self` an `ServiceFactory`
+    /// Convert `Self` to a `ServiceFactory`
    fn into_factory(self) -> T;
 }

--- a/actix-service/src/map_config.rs
+++ b/actix-service/src/map_config.rs
@@ -2,7 +2,10 @@ use std::marker::PhantomData;

 use super::ServiceFactory;

-/// Adapt external config to a config for provided new service
+/// Adapt external config argument to a config for provided service factory
+///
+/// Note that this function consumes the receiving service factory and returns
+/// a wrapped version of it.
 pub fn map_config<T, F, C>(factory: T, f: F) -> MapConfig<T, F, C>
 where
    T: ServiceFactory,
--- a/actix-service/src/pipeline.rs
+++ b/actix-service/src/pipeline.rs
@@ -9,6 +9,7 @@ use crate::map_init_err::MapInitErr;
 use crate::then::{ThenService, ThenServiceFactory};
 use crate::{IntoService, IntoServiceFactory, Service, ServiceFactory};

+/// Contruct new pipeline with one service in pipeline chain.
 pub fn pipeline<F, T>(service: F) -> Pipeline<T>
 where
    F: IntoService<T>,
@@ -19,6 +20,7 @@ where
    }
 }

+/// Contruct new pipeline factory with one service factory.
 pub fn pipeline_factory<T, F>(factory: F) -> PipelineFactory<T>
 where
    T: ServiceFactory,
@@ -29,7 +31,7 @@ where
    }
 }

-/// Pipeline service
+/// Pipeline service - pipeline allows to compose multiple service into one service.
 pub struct Pipeline<T> {
    service: T,
 }
@@ -159,7 +161,7 @@ impl<T: Service> Service for Pipeline<T> {
    }
 }

-/// Pipeline constructor
+/// Pipeline factory
 pub struct PipelineFactory<T> {
    factory: T,
 }
--- a/actix-service/src/transform.rs
+++ b/actix-service/src/transform.rs
@@ -7,23 +7,88 @@ use std::task::{Context, Poll};
 use crate::transform_err::TransformMapInitErr;
 use crate::{IntoServiceFactory, Service, ServiceFactory};

-/// Apply transform to a service. Function returns
-/// services factory that in initialization creates
-/// service and applies transform to this service.
-pub fn apply<T, S, U>(t: T, service: U) -> ApplyTransform<T, S>
+/// Apply transform to a service.
+pub fn apply<T, S, U>(t: T, factory: U) -> ApplyTransform<T, S>
 where
    S: ServiceFactory,
    T: Transform<S::Service, InitError = S::InitError>,
    U: IntoServiceFactory<S>,
 {
-    ApplyTransform::new(t, service.into_factory())
+    ApplyTransform::new(t, factory.into_factory())
 }

+/// The `Transform` trait defines the interface of a service factory that wraps inner service
+/// during construction.
+///
+/// Transform(middleware) wraps inner service and runs during
+/// inbound and/or outbound processing in the request/response lifecycle.
+/// It may modify request and/or response.
+///
+/// For example, timeout transform:
+///
+/// ```rust,ignore
+/// pub struct Timeout<S> {
+///     service: S,
+///     timeout: Duration,
+/// }
+///
+/// impl<S> Service for Timeout<S>
+/// where
+///     S: Service,
+/// {
+///     type Request = S::Request;
+///     type Response = S::Response;
+///     type Error = TimeoutError<S::Error>;
+///     type Future = TimeoutServiceResponse<S>;
+///
+///     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+///         ready!(self.service.poll_ready(cx)).map_err(TimeoutError::Service)
+///     }
+///
+///     fn call(&mut self, req: S::Request) -> Self::Future {
+///         TimeoutServiceResponse {
+///             fut: self.service.call(req),
+///             sleep: Delay::new(clock::now() + self.timeout),
+///         }
+///     }
+/// }
+/// ```
+///
+/// Timeout service in above example is decoupled from underlying service implementation
+/// and could be applied to any service.
+///
 /// The `Transform` trait defines the interface of a Service factory. `Transform`
 /// is often implemented for middleware, defining how to construct a
 /// middleware Service. A Service that is constructed by the factory takes
 /// the Service that follows it during execution as a parameter, assuming
 /// ownership of the next Service.
+///
+/// Factory for `Timeout` middleware from the above example could look like this:
+///
+/// ```rust,,ignore
+/// pub struct TimeoutTransform {
+///     timeout: Duration,
+/// }
+///
+/// impl<S> Transform<S> for TimeoutTransform<E>
+/// where
+///     S: Service,
+/// {
+///     type Request = S::Request;
+///     type Response = S::Response;
+///     type Error = TimeoutError<S::Error>;
+///     type InitError = S::Error;
+///     type Transform = Timeout<S>;
+///     type Future = Ready<Result<Self::Transform, Self::InitError>>;
+///
+///     fn new_transform(&self, service: S) -> Self::Future {
+///         ok(TimeoutService {
+///             service,
+///             timeout: self.timeout,
+///         })
+///     }
+/// }
+/// ```
 pub trait Transform<S> {
    /// Requests handled by the service.
    type Request;
@@ -41,13 +106,13 @@ pub trait Transform<S> {
        Error = Self::Error,
    >;

-    /// Errors produced while building a service.
+    /// Errors produced while building a transform service.
    type InitError;

    /// The future response value.
    type Future: Future<Output = Result<Self::Transform, Self::InitError>>;

-    /// Creates and returns a new Service component, asynchronously
+    /// Creates and returns a new Transform component, asynchronously
    fn new_transform(&self, service: S) -> Self::Future;

    /// Map this transforms's factory error to a different error,
--- a/actix-testing/Cargo.toml
+++ b/actix-testing/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-testing"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix testing utils"
 keywords = ["network", "framework", "async", "futures"]
@@ -17,10 +17,10 @@ name = "actix_testing"
 path = "src/lib.rs"

 [dependencies]
-actix-rt = "1.0.0-alpha.3"
-actix-macros = "0.1.0-alpha.1"
-actix-server = "1.0.0-alpha.3"
-actix-service = "1.0.0-alpha.3"
+actix-rt = "1.0.0"
+actix-macros = "0.1.0"
+actix-server = "1.0.0"
+actix-service = "1.0.0"

 log = "0.4"
 net2 = "0.2"
--- a/actix-testing/src/lib.rs
+++ b/actix-testing/src/lib.rs
@@ -20,12 +20,12 @@ pub use actix_macros::test;
 /// # Examples
 ///
 /// ```rust
-/// use actix_service::{service_fn};
+/// use actix_service::fn_service;
 /// use actix_testing::TestServer;
 ///
 /// #[actix_rt::main]
 /// async fn main() {
-///     let srv = TestServer::with(|| service_fn(
+///     let srv = TestServer::with(|| fn_service(
 ///         |sock| async move {
 ///             println!("New connection: {:?}", sock);
 ///             Ok::<_, ()>(())
--- a/actix-tls/Cargo.toml
+++ b/actix-tls/Cargo.toml
@@ -32,10 +32,10 @@ rustls = ["rust-tls", "webpki", "webpki-roots", "tokio-rustls"]
 nativetls = ["native-tls", "tokio-tls"]

 [dependencies]
-actix-service = "1.0.0-alpha.3"
-actix-codec = "0.2.0-alpha.3"
-actix-utils = "1.0.0-alpha.3"
-actix-rt = "1.0.0-alpha.3"
+actix-service = "1.0.0"
+actix-codec = "0.2.0"
+actix-utils = "1.0.0"
+actix-rt = "1.0.0"
 derive_more = "0.99.2"
 either = "1.5.2"
 futures = "0.3.1"
@@ -57,4 +57,4 @@ tokio-tls = { version="0.3", optional = true }

 [dev-dependencies]
 bytes = "0.5"
-actix-testing = { version="1.0.0-alpha.3" }
+actix-testing = { version="1.0.0" }
--- a/actix-tls/src/openssl.rs
+++ b/actix-tls/src/openssl.rs
@@ -3,7 +3,7 @@ use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::{Context, Poll};

-pub use open_ssl::ssl::{SslAcceptor, SslAcceptorBuilder};
+pub use open_ssl::ssl::{AlpnError, SslAcceptor, SslAcceptorBuilder};
 pub use tokio_openssl::{HandshakeError, SslStream};

 use actix_codec::{AsyncRead, AsyncWrite};
--- a/actix-utils/CHANGES.md
+++ b/actix-utils/CHANGES.md
@@ -1,5 +1,9 @@
 # Changes

+## [1.0.0] - 2019-12-11
+
+* Simplify oneshot and mpsc implementations
+
 ## [1.0.0-alpha.3] - 2019-12-07

 * Migrate to tokio 0.2
--- a/actix-utils/Cargo.toml
+++ b/actix-utils/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-utils"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix utils - various actix net related services"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,6 @@ 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 = ".."

@@ -18,9 +17,9 @@ name = "actix_utils"
 path = "src/lib.rs"

 [dependencies]
-actix-service = "1.0.0-alpha.3"
-actix-rt = "1.0.0-alpha.3"
-actix-codec = "0.2.0-alpha.3"
+actix-service = "1.0.0"
+actix-rt = "1.0.0"
+actix-codec = "0.2.0"
 bytes = "0.5.2"
 either = "1.5.2"
 futures = "0.3.1"
--- a/actix-utils/src/cell.rs
+++ b/actix-utils/src/cell.rs
@@ -2,16 +2,12 @@

 use std::cell::UnsafeCell;
 use std::fmt;
-use std::rc::{Rc, Weak};
+use std::rc::Rc;

 pub(crate) struct Cell<T> {
    pub(crate) inner: Rc<UnsafeCell<T>>,
 }

-pub(crate) struct WeakCell<T> {
-    inner: Weak<UnsafeCell<T>>,
-}
-
 impl<T> Clone for Cell<T> {
    fn clone(&self) -> Self {
        Self {
@@ -27,39 +23,26 @@ impl<T: fmt::Debug> fmt::Debug for Cell<T> {
 }

 impl<T> Cell<T> {
-    pub fn new(inner: T) -> Self {
+    pub(crate) fn new(inner: T) -> Self {
        Self {
            inner: Rc::new(UnsafeCell::new(inner)),
        }
    }

-    pub fn downgrade(&self) -> WeakCell<T> {
-        WeakCell {
-            inner: Rc::downgrade(&self.inner),
-        }
+    pub(crate) fn strong_count(&self) -> usize {
+        Rc::strong_count(&self.inner)
    }

-    pub fn get_ref(&self) -> &T {
+    pub(crate) fn get_ref(&self) -> &T {
        unsafe { &*self.inner.as_ref().get() }
    }

-    pub fn get_mut(&mut self) -> &mut T {
+    pub(crate) fn get_mut(&mut self) -> &mut T {
        unsafe { &mut *self.inner.as_ref().get() }
    }
-}

-impl<T> WeakCell<T> {
-    pub fn upgrade(&self) -> Option<Cell<T>> {
-        if let Some(inner) = self.inner.upgrade() {
-            Some(Cell { inner })
-        } else {
-            None
-        }
-    }
-}
-
-impl<T: fmt::Debug> fmt::Debug for WeakCell<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        self.inner.fmt(f)
+    #[allow(clippy::mut_from_ref)]
+    pub(crate) unsafe fn get_mut_unsafe(&self) -> &mut T {
+        &mut *self.inner.as_ref().get()
    }
 }
--- a/actix-utils/src/framed.rs
+++ b/actix-utils/src/framed.rs
@@ -1,37 +1,33 @@
 //! Framed dispatcher service and related utilities
 #![allow(type_alias_bounds)]
-use std::collections::VecDeque;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::{fmt, mem};

 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, Service};
-use futures::future::{ready, FutureExt};
-use futures::{Future, Sink, Stream};
+use futures::{Future, FutureExt, Stream};
 use log::debug;

-use crate::cell::Cell;
 use crate::mpsc;
-use crate::task::LocalWaker;

 type Request<U> = <U as Decoder>::Item;
 type Response<U> = <U as Encoder>::Item;

 /// Framed transport errors
-pub enum FramedTransportError<E, U: Encoder + Decoder> {
+pub enum DispatcherError<E, U: Encoder + Decoder> {
    Service(E),
    Encoder(<U as Encoder>::Error),
    Decoder(<U as Decoder>::Error),
 }

-impl<E, U: Encoder + Decoder> From<E> for FramedTransportError<E, U> {
+impl<E, U: Encoder + Decoder> From<E> for DispatcherError<E, U> {
    fn from(err: E) -> Self {
-        FramedTransportError::Service(err)
+        DispatcherError::Service(err)
    }
 }

-impl<E, U: Encoder + Decoder> fmt::Debug for FramedTransportError<E, U>
+impl<E, U: Encoder + Decoder> fmt::Debug for DispatcherError<E, U>
 where
    E: fmt::Debug,
    <U as Encoder>::Error: fmt::Debug,
@@ -39,20 +35,14 @@ where
 {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        match *self {
-            FramedTransportError::Service(ref e) => {
-                write!(fmt, "FramedTransportError::Service({:?})", e)
-            }
-            FramedTransportError::Encoder(ref e) => {
-                write!(fmt, "FramedTransportError::Encoder({:?})", e)
-            }
-            FramedTransportError::Decoder(ref e) => {
-                write!(fmt, "FramedTransportError::Encoder({:?})", e)
-            }
+            DispatcherError::Service(ref e) => write!(fmt, "DispatcherError::Service({:?})", e),
+            DispatcherError::Encoder(ref e) => write!(fmt, "DispatcherError::Encoder({:?})", e),
+            DispatcherError::Decoder(ref e) => write!(fmt, "DispatcherError::Decoder({:?})", e),
        }
    }
 }

-impl<E, U: Encoder + Decoder> fmt::Display for FramedTransportError<E, U>
+impl<E, U: Encoder + Decoder> fmt::Display for DispatcherError<E, U>
 where
    E: fmt::Display,
    <U as Encoder>::Error: fmt::Debug,
@@ -60,25 +50,22 @@ where
 {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        match *self {
-            FramedTransportError::Service(ref e) => write!(fmt, "{}", e),
-            FramedTransportError::Encoder(ref e) => write!(fmt, "{:?}", e),
-            FramedTransportError::Decoder(ref e) => write!(fmt, "{:?}", e),
+            DispatcherError::Service(ref e) => write!(fmt, "{}", e),
+            DispatcherError::Encoder(ref e) => write!(fmt, "{:?}", e),
+            DispatcherError::Decoder(ref e) => write!(fmt, "{:?}", e),
        }
    }
 }

-pub enum FramedMessage<T> {
-    Message(T),
+pub enum Message<T> {
+    Item(T),
    Close,
 }

-type Rx<U> = Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>;
-type Inner<S: Service, U> = Cell<FramedTransportInner<<U as Encoder>::Item, S::Error>>;
-
 /// FramedTransport - is a future that reads frames from Framed object
 /// and pass then to the service.
 #[pin_project::pin_project]
-pub struct FramedTransport<S, T, U>
+pub struct Dispatcher<S, T, U>
 where
    S: Service<Request = Request<U>, Response = Response<U>>,
    S::Error: 'static,
@@ -89,26 +76,37 @@ where
    <U as Encoder>::Error: std::fmt::Debug,
 {
    service: S,
-    state: TransportState<S, U>,
+    state: State<S, U>,
    framed: Framed<T, U>,
-    rx: Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: Cell<FramedTransportInner<<U as Encoder>::Item, S::Error>>,
+    rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
+    tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
 }

-enum TransportState<S: Service, U: Encoder + Decoder> {
+enum State<S: Service, U: Encoder + Decoder> {
    Processing,
-    Error(FramedTransportError<S::Error, U>),
-    FramedError(FramedTransportError<S::Error, U>),
+    Error(DispatcherError<S::Error, U>),
+    FramedError(DispatcherError<S::Error, U>),
    FlushAndStop,
    Stopping,
 }

-struct FramedTransportInner<I, E> {
-    buf: VecDeque<Result<I, E>>,
-    task: LocalWaker,
+impl<S: Service, U: Encoder + Decoder> State<S, U> {
+    fn take_error(&mut self) -> DispatcherError<S::Error, U> {
+        match mem::replace(self, State::Processing) {
+            State::Error(err) => err,
+            _ => panic!(),
+        }
+    }
+
+    fn take_framed_error(&mut self) -> DispatcherError<S::Error, U> {
+        match mem::replace(self, State::Processing) {
+            State::FramedError(err) => err,
+            _ => panic!(),
+        }
+    }
 }

-impl<S, T, U> FramedTransport<S, T, U>
+impl<S, T, U> Dispatcher<S, T, U>
 where
    S: Service<Request = Request<U>, Response = Response<U>>,
    S::Error: 'static,
@@ -119,25 +117,19 @@ where
    <U as Encoder>::Error: std::fmt::Debug,
 {
    pub fn new<F: IntoService<S>>(framed: Framed<T, U>, service: F) -> Self {
-        FramedTransport {
+        let (tx, rx) = mpsc::channel();
+        Dispatcher {
            framed,
-            rx: None,
+            rx,
+            tx,
            service: service.into_service(),
-            state: TransportState::Processing,
-            inner: Cell::new(FramedTransportInner {
-                buf: VecDeque::new(),
-                task: LocalWaker::new(),
-            }),
+            state: State::Processing,
        }
    }

-    /// Get Sender
-    pub fn set_receiver(
-        mut self,
-        rx: mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>,
-    ) -> Self {
-        self.rx = Some(rx);
-        self
+    /// Get sink
+    pub fn get_sink(&self) -> mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>> {
+        self.tx.clone()
    }

    /// Get reference to a service wrapped by `FramedTransport` instance.
@@ -162,9 +154,99 @@ where
    pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
        &mut self.framed
    }
+
+    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool
+    where
+        S: Service<Request = Request<U>, Response = Response<U>>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder,
+        <U as Encoder>::Item: 'static,
+        <U as Encoder>::Error: std::fmt::Debug,
+    {
+        loop {
+            match self.service.poll_ready(cx) {
+                Poll::Ready(Ok(_)) => {
+                    let item = match self.framed.next_item(cx) {
+                        Poll::Ready(Some(Ok(el))) => el,
+                        Poll::Ready(Some(Err(err))) => {
+                            self.state = State::FramedError(DispatcherError::Decoder(err));
+                            return true;
+                        }
+                        Poll::Pending => return false,
+                        Poll::Ready(None) => {
+                            self.state = State::Stopping;
+                            return true;
+                        }
+                    };
+
+                    let tx = self.tx.clone();
+                    actix_rt::spawn(self.service.call(item).map(move |item| {
+                        let _ = tx.send(item.map(Message::Item));
+                    }));
+                }
+                Poll::Pending => return false,
+                Poll::Ready(Err(err)) => {
+                    self.state = State::Error(DispatcherError::Service(err));
+                    return true;
+                }
+            }
+        }
+    }
+
+    /// write to framed object
+    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool
+    where
+        S: Service<Request = Request<U>, Response = Response<U>>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder,
+        <U as Encoder>::Item: 'static,
+        <U as Encoder>::Error: std::fmt::Debug,
+    {
+        loop {
+            while !self.framed.is_write_buf_full() {
+                match Pin::new(&mut self.rx).poll_next(cx) {
+                    Poll::Ready(Some(Ok(Message::Item(msg)))) => {
+                        if let Err(err) = self.framed.write(msg) {
+                            self.state = State::FramedError(DispatcherError::Encoder(err));
+                            return true;
+                        }
+                    }
+                    Poll::Ready(Some(Ok(Message::Close))) => {
+                        self.state = State::FlushAndStop;
+                        return true;
+                    }
+                    Poll::Ready(Some(Err(err))) => {
+                        self.state = State::Error(DispatcherError::Service(err));
+                        return true;
+                    }
+                    Poll::Ready(None) | Poll::Pending => break,
+                }
+            }
+
+            if !self.framed.is_write_buf_empty() {
+                match self.framed.flush(cx) {
+                    Poll::Pending => break,
+                    Poll::Ready(Ok(_)) => (),
+                    Poll::Ready(Err(err)) => {
+                        debug!("Error sending data: {:?}", err);
+                        self.state = State::FramedError(DispatcherError::Encoder(err));
+                        return true;
+                    }
+                }
+            } else {
+                break;
+            }
+        }
+
+        false
+    }
 }

-impl<S, T, U> Future for FramedTransport<S, T, U>
+impl<S, T, U> Future for Dispatcher<S, T, U>
 where
    S: Service<Request = Request<U>, Response = Response<U>>,
    S::Error: 'static,
@@ -175,210 +257,46 @@ where
    <U as Encoder>::Error: std::fmt::Debug,
    <U as Decoder>::Error: std::fmt::Debug,
 {
-    type Output = Result<(), FramedTransportError<S::Error, U>>;
+    type Output = Result<(), DispatcherError<S::Error, U>>;

-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.inner.get_ref().task.register(cx.waker());
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        loop {
+            let this = self.as_mut().project();

-        let this = self.project();
-        poll(
-            cx,
-            this.service,
-            this.state,
-            this.framed,
-            this.rx,
-            this.inner,
-        )
-    }
-}
-
-fn poll<S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut TransportState<S, U>,
-    framed: &mut Framed<T, U>,
-    rx: &mut Rx<U>,
-    inner: &mut Inner<S, U>,
-) -> Poll<Result<(), FramedTransportError<S::Error, U>>>
-where
-    S: Service<Request = Request<U>, Response = Response<U>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    match mem::replace(state, TransportState::Processing) {
-        TransportState::Processing => {
-            if poll_read(cx, srv, state, framed, inner)
-                || poll_write(cx, state, framed, rx, inner)
-            {
-                poll(cx, srv, state, framed, rx, inner)
-            } else {
-                Poll::Pending
-            }
-        }
-        TransportState::Error(err) => {
-            let is_empty = framed.is_write_buf_empty();
-            if is_empty || poll_write(cx, state, framed, rx, inner) {
-                Poll::Ready(Err(err))
-            } else {
-                *state = TransportState::Error(err);
-                Poll::Pending
-            }
-        }
-        TransportState::FlushAndStop => {
-            if !framed.is_write_buf_empty() {
-                match Pin::new(framed).poll_flush(cx) {
-                    Poll::Ready(Err(err)) => {
-                        debug!("Error sending data: {:?}", err);
+            return match this.state {
+                State::Processing => {
+                    if self.poll_read(cx) || self.poll_write(cx) {
+                        continue;
+                    } else {
+                        Poll::Pending
+                    }
+                }
+                State::Error(_) => {
+                    // flush write buffer
+                    if !self.framed.is_write_buf_empty() {
+                        if let Poll::Pending = self.framed.flush(cx) {
+                            return Poll::Pending;
+                        }
+                    }
+                    Poll::Ready(Err(self.state.take_error()))
+                }
+                State::FlushAndStop => {
+                    if !this.framed.is_write_buf_empty() {
+                        match this.framed.flush(cx) {
+                            Poll::Ready(Err(err)) => {
+                                debug!("Error sending data: {:?}", err);
+                                Poll::Ready(Ok(()))
+                            }
+                            Poll::Pending => Poll::Pending,
+                            Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
+                        }
+                    } else {
                        Poll::Ready(Ok(()))
                    }
-                    Poll::Pending => Poll::Pending,
-                    Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
                }
-            } else {
-                Poll::Ready(Ok(()))
-            }
-        }
-        TransportState::FramedError(err) => Poll::Ready(Err(err)),
-        TransportState::Stopping => Poll::Ready(Ok(())),
-    }
-}
-
-fn poll_read<S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut TransportState<S, U>,
-    framed: &mut Framed<T, U>,
-    inner: &mut Inner<S, U>,
-) -> bool
-where
-    S: Service<Request = Request<U>, Response = Response<U>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    loop {
-        match srv.poll_ready(cx) {
-            Poll::Ready(Ok(_)) => {
-                let item = match framed.next_item(cx) {
-                    Poll::Ready(Some(Ok(el))) => el,
-                    Poll::Ready(Some(Err(err))) => {
-                        *state =
-                            TransportState::FramedError(FramedTransportError::Decoder(err));
-                        return true;
-                    }
-                    Poll::Pending => return false,
-                    Poll::Ready(None) => {
-                        *state = TransportState::Stopping;
-                        return true;
-                    }
-                };
-
-                let mut cell = inner.clone();
-                let fut = srv.call(item).then(move |item| {
-                    let inner = cell.get_mut();
-                    inner.buf.push_back(item);
-                    inner.task.wake();
-                    ready(())
-                });
-                actix_rt::spawn(fut);
-            }
-            Poll::Pending => return false,
-            Poll::Ready(Err(err)) => {
-                *state = TransportState::Error(FramedTransportError::Service(err));
-                return true;
-            }
+                State::FramedError(_) => Poll::Ready(Err(this.state.take_framed_error())),
+                State::Stopping => Poll::Ready(Ok(())),
+            };
        }
    }
 }
-
-/// write to framed object
-fn poll_write<S, T, U>(
-    cx: &mut Context<'_>,
-    state: &mut TransportState<S, U>,
-    framed: &mut Framed<T, U>,
-    rx: &mut Rx<U>,
-    inner: &mut Inner<S, U>,
-) -> bool
-where
-    S: Service<Request = Request<U>, Response = Response<U>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    // let this = self.project();
-
-    let inner = inner.get_mut();
-    let mut rx_done = rx.is_none();
-    let mut buf_empty = inner.buf.is_empty();
-    loop {
-        while !framed.is_write_buf_full() {
-            if !buf_empty {
-                match inner.buf.pop_front().unwrap() {
-                    Ok(msg) => {
-                        if let Err(err) = framed.write(msg) {
-                            *state =
-                                TransportState::FramedError(FramedTransportError::Encoder(err));
-                            return true;
-                        }
-                        buf_empty = inner.buf.is_empty();
-                    }
-                    Err(err) => {
-                        *state = TransportState::Error(FramedTransportError::Service(err));
-                        return true;
-                    }
-                }
-            }
-
-            if !rx_done && rx.is_some() {
-                match Pin::new(rx.as_mut().unwrap()).poll_next(cx) {
-                    Poll::Ready(Some(FramedMessage::Message(msg))) => {
-                        if let Err(err) = framed.write(msg) {
-                            *state =
-                                TransportState::FramedError(FramedTransportError::Encoder(err));
-                            return true;
-                        }
-                    }
-                    Poll::Ready(Some(FramedMessage::Close)) => {
-                        *state = TransportState::FlushAndStop;
-                        return true;
-                    }
-                    Poll::Ready(None) => {
-                        rx_done = true;
-                        let _ = rx.take();
-                    }
-                    Poll::Pending => rx_done = true,
-                }
-            }
-
-            if rx_done && buf_empty {
-                break;
-            }
-        }
-
-        if !framed.is_write_buf_empty() {
-            match framed.flush(cx) {
-                Poll::Pending => break,
-                Poll::Ready(Err(err)) => {
-                    debug!("Error sending data: {:?}", err);
-                    *state = TransportState::FramedError(FramedTransportError::Encoder(err));
-                    return true;
-                }
-                Poll::Ready(Ok(_)) => (),
-            }
-        } else {
-            break;
-        }
-    }
-
-    false
-}
--- a/actix-utils/src/inflight.rs
+++ b/actix-utils/src/inflight.rs
@@ -115,7 +115,7 @@ mod tests {
    use std::time::Duration;

    use super::*;
-    use actix_service::{apply, factory_fn, Service, ServiceFactory};
+    use actix_service::{apply, fn_factory, Service, ServiceFactory};
    use futures::future::{lazy, ok, FutureExt, LocalBoxFuture};

    struct SleepService(Duration);
@@ -155,7 +155,7 @@ mod tests {
    async fn test_newtransform() {
        let wait_time = Duration::from_millis(50);

-        let srv = apply(InFlight::new(1), factory_fn(|| ok(SleepService(wait_time))));
+        let srv = apply(InFlight::new(1), fn_factory(|| ok(SleepService(wait_time))));

        let mut srv = srv.new_service(&()).await.unwrap();
        assert_eq!(lazy(|cx| srv.poll_ready(cx)).await, Poll::Ready(Ok(())));
--- a/actix-utils/src/mpsc.rs
+++ b/actix-utils/src/mpsc.rs
@@ -1,34 +1,27 @@
-//! A multi-producer, single-consumer, futures-aware, FIFO queue with back
-//! pressure, for use communicating between tasks on the same thread.
-//!
-//! These queues are the same as those in `futures::sync`, except they're not
-//! intended to be sent across threads.
-
+//! A multi-producer, single-consumer, futures-aware, FIFO queue.
 use std::any::Any;
-use std::cell::RefCell;
 use std::collections::VecDeque;
 use std::error::Error;
+use std::fmt;
 use std::pin::Pin;
-use std::rc::{Rc, Weak};
 use std::task::{Context, Poll};
-use std::{fmt, mem};

 use futures::{Sink, Stream};

+use crate::cell::Cell;
 use crate::task::LocalWaker;

 /// Creates a unbounded in-memory channel with buffered storage.
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
-    let shared = Rc::new(RefCell::new(Shared {
+    let shared = Cell::new(Shared {
+        has_receiver: true,
        buffer: VecDeque::new(),
        blocked_recv: LocalWaker::new(),
-    }));
+    });
    let sender = Sender {
-        shared: Rc::downgrade(&shared),
-    };
-    let receiver = Receiver {
-        state: State::Open(shared),
+        shared: shared.clone(),
    };
+    let receiver = Receiver { shared };
    (sender, receiver)
 }

@@ -36,6 +29,7 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
 struct Shared<T> {
    buffer: VecDeque<T>,
    blocked_recv: LocalWaker,
+    has_receiver: bool,
 }

 /// The transmission end of a channel.
@@ -43,22 +37,30 @@ struct Shared<T> {
 /// This is created by the `channel` function.
 #[derive(Debug)]
 pub struct Sender<T> {
-    shared: Weak<RefCell<Shared<T>>>,
+    shared: Cell<Shared<T>>,
 }

+impl<T> Unpin for Sender<T> {}
+
 impl<T> Sender<T> {
    /// Sends the provided message along this channel.
    pub fn send(&self, item: T) -> Result<(), SendError<T>> {
-        let shared = match self.shared.upgrade() {
-            Some(shared) => shared,
-            None => return Err(SendError(item)), // receiver was dropped
+        let shared = unsafe { self.shared.get_mut_unsafe() };
+        if !shared.has_receiver {
+            return Err(SendError(item)); // receiver was dropped
        };
-        let mut shared = shared.borrow_mut();
-
        shared.buffer.push_back(item);
        shared.blocked_recv.wake();
        Ok(())
    }
+
+    /// Closes the sender half
+    ///
+    /// This prevents any further messages from being sent on the channel while
+    /// still enabling the receiver to drain messages that are buffered.
+    pub fn close(&mut self) {
+        self.shared.get_mut().has_receiver = false;
+    }
 }

 impl<T> Clone for Sender<T> {
@@ -91,17 +93,13 @@ impl<T> Sink<T> for Sender<T> {

 impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
-        let shared = match self.shared.upgrade() {
-            Some(shared) => shared,
-            None => return,
-        };
-        // The number of existing `Weak` indicates if we are possibly the last
-        // `Sender`. If we are the last, we possibly must notify a blocked
-        // `Receiver`. `self.shared` is always one of the `Weak` to this shared
-        // data. Therefore the smallest possible Rc::weak_count(&shared) is 1.
-        if Rc::weak_count(&shared) == 1 {
+        let count = self.shared.strong_count();
+        let shared = self.shared.get_mut();
+
+        // check is last sender is about to drop
+        if shared.has_receiver && count == 2 {
            // Wake up receiver as its stream has ended
-            shared.borrow_mut().blocked_recv.wake();
+            shared.blocked_recv.wake();
        }
    }
 }
@@ -111,56 +109,32 @@ impl<T> Drop for Sender<T> {
 /// This is created by the `channel` function.
 #[derive(Debug)]
 pub struct Receiver<T> {
-    state: State<T>,
-}
-
-impl<T> Unpin for Receiver<T> {}
-
-/// Possible states of a receiver. We're either Open (can receive more messages)
-/// or we're closed with a list of messages we have left to receive.
-#[derive(Debug)]
-enum State<T> {
-    Open(Rc<RefCell<Shared<T>>>),
-    Closed(VecDeque<T>),
+    shared: Cell<Shared<T>>,
 }

 impl<T> Receiver<T> {
-    /// Closes the receiving half
-    ///
-    /// This prevents any further messages from being sent on the channel while
-    /// still enabling the receiver to drain messages that are buffered.
-    pub fn close(&mut self) {
-        let items = match self.state {
-            State::Open(ref state) => {
-                let mut state = state.borrow_mut();
-                mem::replace(&mut state.buffer, VecDeque::new())
-            }
-            State::Closed(_) => return,
-        };
-        self.state = State::Closed(items);
+    /// Create Sender
+    pub fn sender(&self) -> Sender<T> {
+        Sender {
+            shared: self.shared.clone(),
+        }
    }
 }

+impl<T> Unpin for Receiver<T> {}
+
 impl<T> Stream for Receiver<T> {
    type Item = T;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let me = match self.state {
-            State::Open(ref mut me) => me,
-            State::Closed(ref mut items) => return Poll::Ready(items.pop_front()),
-        };
-
-        if let Some(shared) = Rc::get_mut(me) {
+        if self.shared.strong_count() == 1 {
            // All senders have been dropped, so drain the buffer and end the
            // stream.
-            return Poll::Ready(shared.borrow_mut().buffer.pop_front());
-        }
-
-        let mut shared = me.borrow_mut();
-        if let Some(msg) = shared.buffer.pop_front() {
+            Poll::Ready(self.shared.get_mut().buffer.pop_front())
+        } else if let Some(msg) = self.shared.get_mut().buffer.pop_front() {
            Poll::Ready(Some(msg))
        } else {
-            shared.blocked_recv.register(cx.waker());
+            self.shared.get_mut().blocked_recv.register(cx.waker());
            Poll::Pending
        }
    }
@@ -168,7 +142,9 @@ impl<T> Stream for Receiver<T> {

 impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
-        self.close();
+        let shared = self.shared.get_mut();
+        shared.buffer.clear();
+        shared.has_receiver = false;
    }
 }

@@ -200,3 +176,44 @@ impl<T> SendError<T> {
        self.0
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::lazy;
+    use futures::{Stream, StreamExt};
+
+    #[actix_rt::test]
+    async fn test_mpsc() {
+        let (tx, mut rx) = channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.next().await.unwrap(), "test");
+
+        let tx2 = tx.clone();
+        tx2.send("test2").unwrap();
+        assert_eq!(rx.next().await.unwrap(), "test2");
+
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut rx).poll_next(cx)).await,
+            Poll::Pending
+        );
+        drop(tx2);
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut rx).poll_next(cx)).await,
+            Poll::Pending
+        );
+        drop(tx);
+        assert_eq!(rx.next().await, None);
+
+        let (tx, rx) = channel();
+        tx.send("test").unwrap();
+        drop(rx);
+        assert!(tx.send("test").is_err());
+
+        let (mut tx, _) = channel();
+        let tx2 = tx.clone();
+        tx.close();
+        assert!(tx.send("test").is_err());
+        assert!(tx2.send("test").is_err());
+    }
+}
--- a/actix-utils/src/oneshot.rs
+++ b/actix-utils/src/oneshot.rs
@@ -1,69 +1,45 @@
-//! A one-shot, futures-aware channel
-//!
-//! This channel is similar to that in `sync::oneshot` but cannot be sent across
-//! threads.
-
+//! A one-shot, futures-aware channel.
 use std::future::Future;
 use std::pin::Pin;
-use std::rc::Rc;
 use std::task::{Context, Poll};

 pub use futures::channel::oneshot::Canceled;

-use crate::cell::{Cell, WeakCell};
+use crate::cell::Cell;
 use crate::task::LocalWaker;

 /// Creates a new futures-aware, one-shot channel.
-///
-/// This function is the same as `sync::oneshot::channel` except that the
-/// returned values cannot be sent across threads.
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
    let inner = Cell::new(Inner {
        value: None,
        rx_task: LocalWaker::new(),
    });
    let tx = Sender {
-        inner: inner.downgrade(),
-    };
-    let rx = Receiver {
-        state: State::Open(inner),
+        inner: inner.clone(),
    };
+    let rx = Receiver { inner };
    (tx, rx)
 }

 /// Represents the completion half of a oneshot through which the result of a
 /// computation is signaled.
-///
-/// This is created by the `unsync::oneshot::channel` function and is equivalent
-/// in functionality to `sync::oneshot::Sender` except that it cannot be sent
-/// across threads.
 #[derive(Debug)]
 pub struct Sender<T> {
-    inner: WeakCell<Inner<T>>,
+    inner: Cell<Inner<T>>,
 }

 /// A future representing the completion of a computation happening elsewhere in
 /// memory.
-///
-/// This is created by the `unsync::oneshot::channel` function and is equivalent
-/// in functionality to `sync::oneshot::Receiver` except that it cannot be sent
-/// across threads.
 #[derive(Debug)]
 #[must_use = "futures do nothing unless polled"]
 pub struct Receiver<T> {
-    state: State<T>,
+    inner: Cell<Inner<T>>,
 }

 // The channels do not ever project Pin to the inner T
 impl<T> Unpin for Receiver<T> {}
 impl<T> Unpin for Sender<T> {}

-#[derive(Debug)]
-enum State<T> {
-    Open(Cell<Inner<T>>),
-    Closed(Option<T>),
-}
-
 #[derive(Debug)]
 struct Inner<T> {
    value: Option<T>,
@@ -78,12 +54,12 @@ impl<T> Sender<T> {
    /// represents.
    ///
    /// If the value is successfully enqueued for the remote end to receive,
-    /// then `Ok(())` is returned. If the receiving end was deallocated before
+    /// then `Ok(())` is returned. If the receiving end was dropped before
    /// this function was called, however, then `Err` is returned with the value
    /// provided.
-    pub fn send(self, val: T) -> Result<(), T> {
-        if let Some(mut inner) = self.inner.upgrade() {
-            let inner = inner.get_mut();
+    pub fn send(mut self, val: T) -> Result<(), T> {
+        if self.inner.strong_count() == 2 {
+            let inner = self.inner.get_mut();
            inner.value = Some(val);
            inner.rx_task.wake();
            Ok(())
@@ -91,47 +67,12 @@ impl<T> Sender<T> {
            Err(val)
        }
    }
-
-    /// Tests to see whether this `Sender`'s corresponding `Receiver`
-    /// has gone away.
-    ///
-    /// This function can be used to learn about when the `Receiver` (consumer)
-    /// half has gone away and nothing will be able to receive a message sent
-    /// from `send`.
-    ///
-    /// Note that this function is intended to *not* be used in the context of a
-    /// future. If you're implementing a future you probably want to call the
-    /// `poll_cancel` function which will block the current task if the
-    /// cancellation hasn't happened yet. This can be useful when working on a
-    /// non-futures related thread, though, which would otherwise panic if
-    /// `poll_cancel` were called.
-    pub fn is_canceled(&self) -> bool {
-        self.inner.upgrade().is_none()
-    }
 }

 impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
-        if let Some(inner) = self.inner.upgrade() {
-            inner.get_ref().rx_task.wake();
-        };
-    }
-}
-
-impl<T> Receiver<T> {
-    /// Gracefully close this receiver, preventing sending any future messages.
-    ///
-    /// Any `send` operation which happens after this method returns is
-    /// guaranteed to fail. Once this method is called the normal `poll` method
-    /// can be used to determine whether a message was actually sent or not. If
-    /// `Canceled` is returned from `poll` then no message was sent.
-    pub fn close(&mut self) {
-        match self.state {
-            State::Open(ref mut inner) => {
-                let value = inner.get_mut().value.take();
-                self.state = State::Closed(value);
-            }
-            State::Closed(_) => {}
+        if self.inner.strong_count() == 2 {
+            self.inner.get_ref().rx_task.wake();
        };
    }
 }
@@ -142,33 +83,48 @@ impl<T> Future for Receiver<T> {
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();

-        let inner = match this.state {
-            State::Open(ref mut inner) => inner,
-            State::Closed(ref mut item) => match item.take() {
-                Some(item) => return Poll::Ready(Ok(item)),
-                None => return Poll::Ready(Err(Canceled)),
-            },
-        };
-
        // If we've got a value, then skip the logic below as we're done.
-        if let Some(val) = inner.get_mut().value.take() {
+        if let Some(val) = this.inner.get_mut().value.take() {
            return Poll::Ready(Ok(val));
        }

-        // If we can get mutable access, then the sender has gone away. We
-        // didn't see a value above, so we're canceled. Otherwise we park
-        // our task and wait for a value to come in.
-        if Rc::get_mut(&mut inner.inner).is_some() {
+        // Check if sender is dropped and return error if it is.
+        if this.inner.strong_count() == 1 {
            Poll::Ready(Err(Canceled))
        } else {
-            inner.get_ref().rx_task.register(cx.waker());
+            this.inner.get_ref().rx_task.register(cx.waker());
            Poll::Pending
        }
    }
 }

-impl<T> Drop for Receiver<T> {
-    fn drop(&mut self) {
-        self.close();
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::lazy;
+
+    #[actix_rt::test]
+    async fn test_oneshot() {
+        let (tx, rx) = channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.await.unwrap(), "test");
+
+        let (tx, rx) = channel();
+        drop(rx);
+        assert!(tx.send("test").is_err());
+
+        let (tx, rx) = channel::<&'static str>();
+        drop(tx);
+        assert!(rx.await.is_err());
+
+        let (tx, mut rx) = channel::<&'static str>();
+        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
+        tx.send("test").unwrap();
+        assert_eq!(rx.await.unwrap(), "test");
+
+        let (tx, mut rx) = channel::<&'static str>();
+        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
+        drop(tx);
+        assert!(rx.await.is_err());
    }
 }
--- a/actix-utils/src/stream.rs
+++ b/actix-utils/src/stream.rs
@@ -3,12 +3,12 @@ use std::pin::Pin;
 use std::task::{Context, Poll};

 use actix_service::{IntoService, Service};
-use futures::Stream;
+use futures::{FutureExt, Stream};

 use crate::mpsc;

 #[pin_project::pin_project]
-pub struct StreamDispatcher<S, T>
+pub struct Dispatcher<S, T>
 where
    S: Stream,
    T: Service<Request = S::Item, Response = ()> + 'static,
@@ -20,7 +20,7 @@ where
    err_tx: mpsc::Sender<T::Error>,
 }

-impl<S, T> StreamDispatcher<S, T>
+impl<S, T> Dispatcher<S, T>
 where
    S: Stream,
    T: Service<Request = S::Item, Response = ()> + 'static,
@@ -30,7 +30,7 @@ where
        F: IntoService<T>,
    {
        let (err_tx, err_rx) = mpsc::channel();
-        StreamDispatcher {
+        Dispatcher {
            err_rx,
            err_tx,
            stream,
@@ -39,7 +39,7 @@ where
    }
 }

-impl<S, T> Future for StreamDispatcher<S, T>
+impl<S, T> Future for Dispatcher<S, T>
 where
    S: Stream,
    T: Service<Request = S::Item, Response = ()> + 'static,
@@ -54,47 +54,23 @@ where
        }

        loop {
-            match this.service.poll_ready(cx)? {
+            return match this.service.poll_ready(cx)? {
                Poll::Ready(_) => match this.stream.poll_next(cx) {
                    Poll::Ready(Some(item)) => {
-                        actix_rt::spawn(StreamDispatcherService {
-                            fut: this.service.call(item),
-                            stop: self.err_tx.clone(),
-                        });
+                        let stop = this.err_tx.clone();
+                        actix_rt::spawn(this.service.call(item).map(move |res| {
+                            if let Err(e) = res {
+                                let _ = stop.send(e);
+                            }
+                        }));
                        this = self.as_mut().project();
+                        continue;
                    }
-                    Poll::Pending => return Poll::Pending,
-                    Poll::Ready(None) => return Poll::Ready(Ok(())),
+                    Poll::Pending => Poll::Pending,
+                    Poll::Ready(None) => Poll::Ready(Ok(())),
                },
-                Poll::Pending => return Poll::Pending,
-            }
-        }
-    }
-}
-
-#[pin_project::pin_project]
-struct StreamDispatcherService<F: Future, E> {
-    #[pin]
-    fut: F,
-    stop: mpsc::Sender<E>,
-}
-
-impl<F, E> Future for StreamDispatcherService<F, E>
-where
-    F: Future<Output = Result<(), E>>,
-{
-    type Output = ();
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.project();
-
-        match this.fut.poll(cx) {
-            Poll::Ready(Ok(_)) => Poll::Ready(()),
-            Poll::Pending => Poll::Pending,
-            Poll::Ready(Err(e)) => {
-                let _ = this.stop.send(e);
-                Poll::Ready(())
-            }
+                Poll::Pending => Poll::Pending,
+            };
        }
    }
 }
--- a/actix-utils/src/timeout.rs
+++ b/actix-utils/src/timeout.rs
@@ -182,7 +182,7 @@ mod tests {
    use std::time::Duration;

    use super::*;
-    use actix_service::{apply, factory_fn, Service, ServiceFactory};
+    use actix_service::{apply, fn_factory, Service, ServiceFactory};
    use futures::future::{ok, FutureExt, LocalBoxFuture};

    struct SleepService(Duration);
@@ -229,7 +229,7 @@ mod tests {

        let timeout = apply(
            Timeout::new(resolution),
-            factory_fn(|| ok::<_, ()>(SleepService(wait_time))),
+            fn_factory(|| ok::<_, ()>(SleepService(wait_time))),
        );
        let mut srv = timeout.new_service(&()).await.unwrap();

--- a/router/CHANGES.txt
+++ b/router/CHANGES.txt
@@ -6,6 +6,8 @@

 * Update regex to 1.3

+* Use bytestring instead of string
+
 ## [0.1.5] - 2019-05-15

 * Remove debug prints
--- a/string/CHANGES.md
+++ b/string/CHANGES.md
@@ -1,5 +1,9 @@
 # Changes

+[0.1.1] - 2019-12-07
+
+* Fix hash impl
+
 [0.1.0] - 2019-12-07

 * Initial release
--- a/string/Cargo.toml
+++ b/string/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "bytestring"
-version = "0.1.0"
+version = "0.1.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "A UTF-8 encoded string with Bytes as a storage"
 keywords = ["actix"]
--- a/string/src/lib.rs
+++ b/string/src/lib.rs
@@ -45,7 +45,7 @@ impl PartialEq<str> for ByteString {

 impl hash::Hash for ByteString {
    fn hash<H: hash::Hasher>(&self, state: &mut H) {
-        self.0.hash(state);
+        (**self).hash(state);
    }
 }

@@ -144,6 +144,19 @@ impl fmt::Display for ByteString {
 #[cfg(test)]
 mod test {
    use super::*;
+    use std::collections::hash_map::DefaultHasher;
+    use std::hash::{Hash, Hasher};
+
+    #[test]
+    fn test_hash() {
+        let mut hasher1 = DefaultHasher::default();
+        "str".hash(&mut hasher1);
+
+        let mut hasher2 = DefaultHasher::default();
+        let s = ByteString::from_static("str");
+        s.hash(&mut hasher2);
+        assert_eq!(hasher1.finish(), hasher2.finish());
+    }

    #[test]
    fn test_from_string() {
Author	SHA1	Message	Date
Nikolay Kim	9d84d14ef4	update deps	2019-12-11 14:47:30 +06:00
Nikolay Kim	60bfa1bfb1	prepare actix-server release	2019-12-11 14:43:26 +06:00
Nikolay Kim	2c81c22b3e	refactor ioframe dispatcher	2019-12-11 14:36:11 +06:00
Nikolay Kim	dded482514	allow to close mpsc sender	2019-12-11 14:36:00 +06:00
Nikolay Kim	631cb86947	refactor framed and stream dispatchers	2019-12-11 12:42:07 +06:00
Nikolay Kim	2e5e69c9ba	Simplify oneshot and mpsc implementations	2019-12-11 11:28:09 +06:00
Nikolay Kim	e315cf2893	prep actix-rt release; update deps	2019-12-11 10:34:50 +06:00
Nikolay Kim	13fd615966	actix-macros release	2019-12-11 10:32:01 +06:00
Nikolay Kim	c094f84b85	prepare actix-service release	2019-12-11 10:29:34 +06:00
Nikolay Kim	25012d290a	update actix-codec dependencies	2019-12-11 10:23:01 +06:00
Nikolay Kim	32202188cc	prepare actix-codec release	2019-12-11 10:18:11 +06:00
Nikolay Kim	bf734a31dc	update docs	2019-12-10 21:34:51 +06:00
Nikolay Kim	d29e7c4ba6	Merge branch 'master' of github.com:actix/actix-net	2019-12-10 21:14:18 +06:00
Nikolay Kim	7163e2c2a2	update doc strings	2019-12-10 21:14:06 +06:00
Nikolay Kim	1d810b4561	re-export AlpnError	2019-12-10 12:15:27 +06:00
daxpedda	0913badd61	Macro improvements. (#74 ) * Macro improvements. * Fix usage in `fn main`.	2019-12-10 08:47:35 +06:00
Nikolay Kim	8b3062cd6e	Fix buffer remaining capacity calcualtion	2019-12-09 21:50:36 +06:00
Nikolay Kim	35218a4df1	add Clone impl for Apply service	2019-12-09 14:07:20 +06:00
Nikolay Kim	d47f1fb730	prepare actix-service release	2019-12-08 19:49:35 +06:00
Nikolay Kim	1ad0bbfb7f	rename fn service helpers	2019-12-08 19:05:05 +06:00
Nikolay Kim	c38a25f102	fix hash impl	2019-12-07 11:51:47 +06:00
Nikolay Kim	110457477a	update changes	2019-12-07 11:04:53 +06:00