Fix actix-service 1.0.3 compatibility

* Add support for tokio tracing for actix Service.

* Address comments

* Change trace's return type to ApplyTransform

* Remove redundant type args

* Remove reference to MakeSpan from docs

Cargo.toml

@@ -10,6 +10,7 @@ members = [
   "actix-testing",
   "actix-threadpool",
   "actix-tls",
+  "actix-tracing",
   "actix-utils",
   "router",
   "string",

actix-connect/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.0.2] - 2020-01-15
+
+* Fix actix-service 1.0.3 compatibility
+
 ## [1.0.1] - 2019-12-15
 
 * Fix trust-dns-resolver compilation

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "1.0.1"
+version = "1.0.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix connect - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]
@@ -10,7 +10,6 @@ documentation = "https://docs.rs/actix-connect/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
 
 [package.metadata.docs.rs]
 features = ["openssl", "rustls", "uri"]
@@ -32,12 +31,12 @@ rustls = ["rust-tls", "tokio-rustls", "webpki"]
 uri = ["http"]
 
 [dependencies]
-actix-service = "1.0.0"
+actix-service = "1.0.3"
 actix-codec = "0.2.0"
-actix-utils = "1.0.3"
+actix-utils = "1.0.6"
 actix-rt = "1.0.0"
 derive_more = "0.99.2"
-either = "1.5.2"
+either = "1.5.3"
 futures = "0.3.1"
 http = { version = "0.2.0", optional = true }
 log = "0.4"

actix-connect/src/lib.rs

@@ -72,7 +72,7 @@ pub fn start_default_resolver() -> AsyncResolver {
 }
 
 /// Create tcp connector service
-pub fn new_connector<T: Address>(
+pub fn new_connector<T: Address + 'static>(
     resolver: AsyncResolver,
 ) -> impl Service<Request = Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError>
        + Clone {
@@ -80,7 +80,7 @@ pub fn new_connector<T: Address>(
 }
 
 /// Create tcp connector service
-pub fn new_connector_factory<T: Address>(
+pub fn new_connector_factory<T: Address + 'static>(
     resolver: AsyncResolver,
 ) -> impl ServiceFactory<
     Config = (),
@@ -93,14 +93,14 @@ pub fn new_connector_factory<T: Address>(
 }
 
 /// Create connector service with default parameters
-pub fn default_connector<T: Address>(
+pub fn default_connector<T: Address + 'static>(
 ) -> impl Service<Request = Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError>
        + Clone {
     pipeline(Resolver::default()).and_then(TcpConnector::new())
 }
 
 /// Create connector service factory with default parameters
-pub fn default_connector_factory<T: Address>() -> impl ServiceFactory<
+pub fn default_connector_factory<T: Address + 'static>() -> impl ServiceFactory<
     Config = (),
     Request = Connect<T>,
     Response = Connection<T, TcpStream>,

actix-ioframe/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [0.5.0] - 2019-12-29
+
+* Simplify state management
+
+* Allow to set custom output stream
+
+* Removed disconnect callback
+
 ## [0.4.1] - 2019-12-11
 
 * Disconnect callback accepts owned state

actix-ioframe/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-ioframe"
-version = "0.4.1"
+version = "0.5.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix framed service"
 keywords = ["network", "framework", "async", "futures"]
@@ -10,19 +10,18 @@ documentation = "https://docs.rs/actix-ioframe/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
 
 [lib]
 name = "actix_ioframe"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "1.0.0"
+actix-service = "1.0.1"
 actix-codec = "0.2.0"
-actix-utils = "1.0.1"
+actix-utils = "1.0.4"
 actix-rt = "1.0.0"
-bytes = "0.5"
-either = "1.5.2"
+bytes = "0.5.3"
+either = "1.5.3"
 futures = "0.3.1"
 pin-project = "0.4.6"
 log = "0.4"

actix-ioframe/src/connect.rs

@@ -3,17 +3,15 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
+use actix_utils::mpsc::Receiver;
 use futures::Stream;
 
-use crate::sink::Sink;
-
-pub struct Connect<Io, Codec, St = ()>
+pub struct Connect<Io, Codec>
 where
     Codec: Encoder + Decoder,
 {
     io: Io,
-    sink: Sink<<Codec as Encoder>::Item>,
-    _t: PhantomData<(St, Codec)>,
+    _t: PhantomData<Codec>,
 }
 
 impl<Io, Codec> Connect<Io, Codec>
@@ -21,36 +19,33 @@ where
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
 {
-    pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item>) -> Self {
+    pub(crate) fn new(io: Io) -> Self {
         Self {
             io,
-            sink,
             _t: PhantomData,
         }
     }
 
-    pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec> {
+    pub fn codec(
+        self,
+        codec: Codec,
+    ) -> ConnectResult<Io, (), Codec, Receiver<<Codec as Encoder>::Item>> {
         ConnectResult {
             state: (),
-            sink: self.sink,
+            out: None,
             framed: Framed::new(self.io, codec),
         }
     }
 }
 
 #[pin_project::pin_project]
-pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
+pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> {
     pub(crate) state: St,
+    pub(crate) out: Option<Out>,
     pub(crate) framed: Framed<Io, Codec>,
-    pub(crate) sink: Sink<<Codec as Encoder>::Item>,
 }
 
-impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
-    #[inline]
-    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
-        &self.sink
-    }
-
+impl<Io, St, Codec: Encoder + Decoder, Out: Unpin> ConnectResult<Io, St, Codec, Out> {
     #[inline]
     pub fn get_ref(&self) -> &Io {
         self.framed.get_ref()
@@ -61,17 +56,28 @@ impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
         self.framed.get_mut()
     }
 
+    pub fn out<U>(self, out: U) -> ConnectResult<Io, St, Codec, U>
+    where
+        U: Stream<Item = <Codec as Encoder>::Item> + Unpin,
+    {
+        ConnectResult {
+            state: self.state,
+            framed: self.framed,
+            out: Some(out),
+        }
+    }
+
     #[inline]
-    pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec> {
+    pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec, Out> {
         ConnectResult {
             state,
             framed: self.framed,
-            sink: self.sink,
+            out: self.out,
         }
     }
 }
 
-impl<Io, St, Codec> Stream for ConnectResult<Io, St, Codec>
+impl<Io, St, Codec, Out> Stream for ConnectResult<Io, St, Codec, Out>
 where
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
@@ -83,7 +89,8 @@ where
     }
 }
 
-impl<Io, St, Codec> futures::Sink<<Codec as Encoder>::Item> for ConnectResult<Io, St, Codec>
+impl<Io, St, Codec, Out> futures::Sink<<Codec as Encoder>::Item>
+    for ConnectResult<Io, St, Codec, Out>
 where
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,

actix-ioframe/src/dispatcher.rs

@@ -1,81 +1,56 @@
 //! Framed dispatcher service and related utilities
 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::{mpsc, oneshot};
-use futures::{FutureExt, Stream};
+use actix_service::Service;
+use actix_utils::mpsc;
+use futures::Stream;
 use log::debug;
 
 use crate::error::ServiceError;
-use crate::item::Item;
-use crate::sink::Sink;
 
-type Request<S, U> = Item<S, U>;
+type Request<U> = <U as Decoder>::Item;
 type Response<U> = <U as Encoder>::Item;
 
-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 Dispatcher<St, S, T, U>
+pub(crate) struct Dispatcher<S, T, U, Out>
 where
-    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
+    S: Service<Request = Request<U>, Response = Option<Response<U>>>,
     S::Error: 'static,
     S::Future: 'static,
     T: AsyncRead + AsyncWrite,
     U: Encoder + Decoder,
     <U as Encoder>::Item: 'static,
     <U as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <U as Encoder>::Item> + Unpin,
 {
     service: S,
-    sink: Sink<<U as Encoder>::Item>,
-    state: St,
-    dispatch_state: FramedState<S, U>,
+    sink: Option<Out>,
+    state: FramedState<S, U>,
     framed: Framed<T, U>,
-    rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
-    tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
-    disconnect: Option<Rc<dyn Fn(St, bool)>>,
+    rx: mpsc::Receiver<Result<<U as Encoder>::Item, S::Error>>,
 }
 
-impl<St, S, T, U> Dispatcher<St, S, T, U>
+impl<S, T, U, Out> Dispatcher<S, T, U, Out>
 where
-    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
+    S: Service<Request = Request<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,
+    Out: Stream<Item = <U as Encoder>::Item> + Unpin,
 {
-    pub(crate) fn new<F: IntoService<S>>(
-        framed: Framed<T, U>,
-        state: St,
-        service: F,
-        sink: Sink<<U as Encoder>::Item>,
-        rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
-        disconnect: Option<Rc<dyn Fn(St, bool)>>,
-    ) -> Self {
-        let tx = rx.sender();
-
+    pub(crate) fn new(framed: Framed<T, U>, service: S, sink: Option<Out>) -> Self {
         Dispatcher {
-            framed,
-            state,
             sink,
-            disconnect,
-            rx,
-            tx,
-            service: service.into_service(),
-            dispatch_state: FramedState::Processing,
+            service,
+            framed,
+            rx: mpsc::channel().1,
+            state: FramedState::Processing,
         }
     }
 }
@@ -84,33 +59,11 @@ enum FramedState<S: Service, U: Encoder + Decoder> {
     Processing,
     Error(ServiceError<S::Error, U>),
     FramedError(ServiceError<S::Error, U>),
-    FlushAndStop(Vec<oneshot::Sender<()>>),
+    FlushAndStop,
     Stopping,
 }
 
 impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
-    fn stop(&mut self, tx: Option<oneshot::Sender<()>>) {
-        match self {
-            FramedState::FlushAndStop(ref mut vec) => {
-                if let Some(tx) = tx {
-                    vec.push(tx)
-                }
-            }
-            FramedState::Processing => {
-                *self = FramedState::FlushAndStop(if let Some(tx) = tx {
-                    vec![tx]
-                } else {
-                    Vec::new()
-                })
-            }
-            FramedState::Error(_) | FramedState::FramedError(_) | FramedState::Stopping => {
-                if let Some(tx) = tx {
-                    let _ = tx.send(());
-                }
-            }
-        }
-    }
-
     fn take_error(&mut self) -> ServiceError<S::Error, U> {
         match std::mem::replace(self, FramedState::Processing) {
             FramedState::Error(err) => err,
@@ -126,16 +79,16 @@ impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
     }
 }
 
-impl<St, S, T, U> Dispatcher<St, S, T, U>
+impl<S, T, U, Out> Dispatcher<S, T, U, Out>
 where
-    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
+    S: Service<Request = Request<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,
+    Out: Stream<Item = <U as Encoder>::Item> + Unpin,
 {
     fn poll_read(&mut self, cx: &mut Context<'_>) -> bool {
         loop {
@@ -144,35 +97,32 @@ where
                     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));
+                            self.state = FramedState::FramedError(ServiceError::Decoder(err));
                             return true;
                         }
                         Poll::Pending => return false,
                         Poll::Ready(None) => {
                             log::trace!("Client disconnected");
-                            self.dispatch_state = FramedState::Stopping;
+                            self.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);
-                            }),
-                    );
+                    let tx = self.rx.sender();
+                    let fut = self.service.call(item);
+                    actix_rt::spawn(async move {
+                        let item = fut.await;
+                        let item = match item {
+                            Ok(Some(item)) => Ok(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));
+                    self.state = FramedState::Error(ServiceError::Service(err));
                     return true;
                 }
             }
@@ -184,27 +134,39 @@ where
         loop {
             while !self.framed.is_write_buf_full() {
                 match Pin::new(&mut self.rx).poll_next(cx) {
-                    Poll::Ready(Some(Ok(Message::Item(msg)))) => {
+                    Poll::Ready(Some(Ok(msg))) => {
                         if let Err(err) = self.framed.write(msg) {
-                            self.dispatch_state =
-                                FramedState::FramedError(ServiceError::Encoder(err));
+                            self.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;
+                        continue;
                     }
                     Poll::Ready(Some(Err(err))) => {
-                        self.dispatch_state = FramedState::Error(ServiceError::Service(err));
+                        self.state = FramedState::Error(ServiceError::Service(err));
                         return true;
                     }
-                    Poll::Ready(None) | Poll::Pending => break,
+                    Poll::Ready(None) | Poll::Pending => (),
                 }
+
+                if self.sink.is_some() {
+                    match Pin::new(self.sink.as_mut().unwrap()).poll_next(cx) {
+                        Poll::Ready(Some(msg)) => {
+                            if let Err(err) = self.framed.write(msg) {
+                                self.state =
+                                    FramedState::FramedError(ServiceError::Encoder(err));
+                                return true;
+                            }
+                            continue;
+                        }
+                        Poll::Ready(None) => {
+                            let _ = self.sink.take();
+                            self.state = FramedState::FlushAndStop;
+                            return true;
+                        }
+                        Poll::Pending => (),
+                    }
+                }
+                break;
             }
 
             if !self.framed.is_write_buf_empty() {
@@ -213,8 +175,7 @@ where
                     Poll::Ready(Ok(_)) => (),
                     Poll::Ready(Err(err)) => {
                         debug!("Error sending data: {:?}", err);
-                        self.dispatch_state =
-                            FramedState::FramedError(ServiceError::Encoder(err));
+                        self.state = FramedState::FramedError(ServiceError::Encoder(err));
                         return true;
                     }
                 }
@@ -229,14 +190,16 @@ where
         &mut self,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), ServiceError<S::Error, U>>> {
-        match self.dispatch_state {
-            FramedState::Processing => {
-                if self.poll_read(cx) || self.poll_write(cx) {
-                    self.poll(cx)
+        match self.state {
+            FramedState::Processing => loop {
+                let read = self.poll_read(cx);
+                let write = self.poll_write(cx);
+                if read || write {
+                    continue;
                 } else {
-                    Poll::Pending
+                    return Poll::Pending;
                 }
-            }
+            },
             FramedState::Error(_) => {
                 // flush write buffer
                 if !self.framed.is_write_buf_empty() {
@@ -244,12 +207,21 @@ where
                         return Poll::Pending;
                     }
                 }
-                if let Some(ref disconnect) = self.disconnect {
-                    (&*disconnect)(self.state.clone(), true);
-                }
-                Poll::Ready(Err(self.dispatch_state.take_error()))
+                Poll::Ready(Err(self.state.take_error()))
             }
-            FramedState::FlushAndStop(ref mut vec) => {
+            FramedState::FlushAndStop => {
+                // drain service responses
+                match Pin::new(&mut self.rx).poll_next(cx) {
+                    Poll::Ready(Some(Ok(msg))) => {
+                        if let Err(_) = self.framed.write(msg) {
+                            return Poll::Ready(Ok(()));
+                        }
+                    }
+                    Poll::Ready(Some(Err(_))) => return Poll::Ready(Ok(())),
+                    Poll::Ready(None) | Poll::Pending => (),
+                }
+
+                // flush io
                 if !self.framed.is_write_buf_empty() {
                     match self.framed.flush(cx) {
                         Poll::Ready(Err(err)) => {
@@ -261,26 +233,10 @@ where
                         Poll::Ready(_) => (),
                     }
                 };
-                for tx in vec.drain(..) {
-                    let _ = tx.send(());
-                }
-                if let Some(ref disconnect) = self.disconnect {
-                    (&*disconnect)(self.state.clone(), false);
-                }
-                Poll::Ready(Ok(()))
-            }
-            FramedState::FramedError(_) => {
-                if let Some(ref disconnect) = self.disconnect {
-                    (&*disconnect)(self.state.clone(), true);
-                }
-                Poll::Ready(Err(self.dispatch_state.take_framed_error()))
-            }
-            FramedState::Stopping => {
-                if let Some(ref disconnect) = self.disconnect {
-                    (&*disconnect)(self.state.clone(), false);
-                }
                 Poll::Ready(Ok(()))
             }
+            FramedState::FramedError(_) => Poll::Ready(Err(self.state.take_framed_error())),
+            FramedState::Stopping => Poll::Ready(Ok(())),
         }
     }
 }

actix-ioframe/src/item.rs

@@ -1,82 +0,0 @@
-use std::fmt;
-use std::ops::{Deref, DerefMut};
-
-use actix_codec::{Decoder, Encoder};
-
-use crate::sink::Sink;
-
-pub struct Item<St, Codec: Encoder + Decoder> {
-    state: St,
-    sink: Sink<<Codec as Encoder>::Item>,
-    item: <Codec as Decoder>::Item,
-}
-
-impl<St, Codec> Item<St, Codec>
-where
-    Codec: Encoder + Decoder,
-{
-    pub(crate) fn new(
-        state: St,
-        sink: Sink<<Codec as Encoder>::Item>,
-        item: <Codec as Decoder>::Item,
-    ) -> Self {
-        Item { state, sink, item }
-    }
-
-    #[inline]
-    pub fn state(&self) -> &St {
-        &self.state
-    }
-
-    #[inline]
-    pub fn state_mut(&mut self) -> &mut St {
-        &mut self.state
-    }
-
-    #[inline]
-    pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
-        &self.sink
-    }
-
-    #[inline]
-    pub fn into_inner(self) -> <Codec as Decoder>::Item {
-        self.item
-    }
-
-    #[inline]
-    pub fn into_parts(self) -> (St, Sink<<Codec as Encoder>::Item>, <Codec as Decoder>::Item) {
-        (self.state, self.sink, self.item)
-    }
-}
-
-impl<St, Codec> Deref for Item<St, Codec>
-where
-    Codec: Encoder + Decoder,
-{
-    type Target = <Codec as Decoder>::Item;
-
-    #[inline]
-    fn deref(&self) -> &<Codec as Decoder>::Item {
-        &self.item
-    }
-}
-
-impl<St, Codec> DerefMut for Item<St, Codec>
-where
-    Codec: Encoder + Decoder,
-{
-    #[inline]
-    fn deref_mut(&mut self) -> &mut <Codec as Decoder>::Item {
-        &mut self.item
-    }
-}
-
-impl<St, Codec> fmt::Debug for Item<St, Codec>
-where
-    Codec: Encoder + Decoder,
-    <Codec as Decoder>::Item: fmt::Debug,
-{
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_tuple("Item").field(&self.item).finish()
-    }
-}

actix-ioframe/src/lib.rs

@@ -1,15 +1,11 @@
-#![deny(rust_2018_idioms, warnings)]
+// #![deny(rust_2018_idioms, warnings)]
 #![allow(clippy::type_complexity, clippy::too_many_arguments)]
 
 mod connect;
 mod dispatcher;
 mod error;
-mod item;
 mod service;
-mod sink;
 
 pub use self::connect::{Connect, ConnectResult};
 pub use self::error::ServiceError;
-pub use self::item::Item;
-pub use self::service::{Builder, NewServiceBuilder, ServiceBuilder};
-pub use self::sink::Sink;
+pub use self::service::{Builder, FactoryBuilder};

actix-ioframe/src/service.rs

@@ -4,108 +4,57 @@ use std::pin::Pin;
 use std::rc::Rc;
 use std::task::{Context, Poll};
 
-use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder};
+use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
-use actix_utils::mpsc;
 use either::Either;
-use futures::future::{FutureExt, LocalBoxFuture};
+use futures::{ready, Stream};
 use pin_project::project;
 
 use crate::connect::{Connect, ConnectResult};
-use crate::dispatcher::{Dispatcher, Message};
+use crate::dispatcher::Dispatcher;
 use crate::error::ServiceError;
-use crate::item::Item;
-use crate::sink::Sink;
 
-type RequestItem<S, U> = Item<S, U>;
+type RequestItem<U> = <U as Decoder>::Item;
 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.
-pub struct Builder<St, Codec>(PhantomData<(St, Codec)>);
-
-impl<St: Clone, Codec> Default for Builder<St, Codec> {
-    fn default() -> Builder<St, Codec> {
-        Builder::new()
-    }
-}
-
-impl<St: Clone, Codec> Builder<St, Codec> {
-    pub fn new() -> Builder<St, Codec> {
-        Builder(PhantomData)
-    }
-
-    /// Construct framed handler service with specified connect service
-    pub fn service<Io, C, F>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec>
-    where
-        F: IntoService<C>,
-        Io: AsyncRead + AsyncWrite,
-        C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
-        Codec: Decoder + Encoder,
-    {
-        ServiceBuilder {
-            connect: connect.into_service(),
-            disconnect: None,
-            _t: PhantomData,
-        }
-    }
-
-    /// Construct framed handler new service with specified connect service
-    pub fn factory<Io, C, F>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec>
-    where
-        F: IntoServiceFactory<C>,
-        Io: AsyncRead + AsyncWrite,
-        C: ServiceFactory<
-            Config = (),
-            Request = Connect<Io, Codec>,
-            Response = ConnectResult<Io, St, Codec>,
-        >,
-        C::Error: 'static,
-        C::Future: 'static,
-        Codec: Decoder + Encoder,
-    {
-        NewServiceBuilder {
-            connect: connect.into_factory(),
-            disconnect: None,
-            _t: PhantomData,
-        }
-    }
-}
-
-pub struct ServiceBuilder<St, C, Io, Codec> {
+pub struct Builder<St, C, Io, Codec, Out> {
     connect: C,
-    disconnect: Option<Rc<dyn Fn(St, bool)>>,
-    _t: PhantomData<(St, Io, Codec)>,
+    _t: PhantomData<(St, Io, Codec, Out)>,
 }
 
-impl<St, C, Io, Codec> ServiceBuilder<St, C, Io, Codec>
+impl<St, C, Io, Codec, Out> Builder<St, C, Io, Codec, Out>
 where
-    St: Clone,
-    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
     Io: AsyncRead + AsyncWrite,
     Codec: Decoder + Encoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
-    /// Callback to execute on disconnect
-    ///
-    /// Second parameter indicates error occured during disconnect.
-    pub fn disconnect<F, Out>(mut self, disconnect: F) -> Self
+    /// Construct framed handler service with specified connect service
+    pub fn new<F>(connect: F) -> Builder<St, C, Io, Codec, Out>
     where
-        F: Fn(St, bool) + 'static,
+        F: IntoService<C>,
+        Io: AsyncRead + AsyncWrite,
+        C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
+        Codec: Decoder + Encoder,
+        Out: Stream<Item = <Codec as Encoder>::Item>,
     {
-        self.disconnect = Some(Rc::new(disconnect));
-        self
+        Builder {
+            connect: connect.into_service(),
+            _t: PhantomData,
+        }
     }
 
     /// 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 build<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec, Out>
     where
         F: IntoServiceFactory<T>,
         T: ServiceFactory<
             Config = St,
-            Request = RequestItem<St, Codec>,
+            Request = RequestItem<Codec>,
             Response = ResponseItem<Codec>,
             Error = C::Error,
             InitError = C::Error,
@@ -114,211 +63,261 @@ where
         FramedServiceImpl {
             connect: self.connect,
             handler: Rc::new(service.into_factory()),
-            disconnect: self.disconnect.clone(),
             _t: PhantomData,
         }
     }
 }
 
-pub struct NewServiceBuilder<St, C, Io, Codec> {
+/// Service builder - structure that follows the builder pattern
+/// for building instances for framed services.
+pub struct FactoryBuilder<St, C, Io, Codec, Out> {
     connect: C,
-    disconnect: Option<Rc<dyn Fn(St, bool)>>,
-    _t: PhantomData<(St, Io, Codec)>,
+    _t: PhantomData<(St, Io, Codec, Out)>,
 }
 
-impl<St, C, Io, Codec> NewServiceBuilder<St, C, Io, Codec>
+impl<St, C, Io, Codec, Out> FactoryBuilder<St, C, Io, Codec, Out>
 where
-    St: Clone,
     Io: AsyncRead + AsyncWrite,
     C: ServiceFactory<
         Config = (),
         Request = Connect<Io, Codec>,
-        Response = ConnectResult<Io, St, Codec>,
+        Response = ConnectResult<Io, St, Codec, Out>,
     >,
-    C::Error: 'static,
-    C::Future: 'static,
     Codec: Decoder + Encoder,
-    <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
-    /// Callback to execute on disconnect
-    ///
-    /// Second parameter indicates error occured during disconnect.
-    pub fn disconnect<F>(mut self, disconnect: F) -> Self
+    /// Construct framed handler new service with specified connect service
+    pub fn new<F>(connect: F) -> FactoryBuilder<St, C, Io, Codec, Out>
     where
-        F: Fn(St, bool) + 'static,
+        F: IntoServiceFactory<C>,
+        Io: AsyncRead + AsyncWrite,
+        C: ServiceFactory<
+            Config = (),
+            Request = Connect<Io, Codec>,
+            Response = ConnectResult<Io, St, Codec, Out>,
+        >,
+        Codec: Decoder + Encoder,
+        Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
     {
-        self.disconnect = Some(Rc::new(disconnect));
-        self
+        FactoryBuilder {
+            connect: connect.into_factory(),
+            _t: PhantomData,
+        }
     }
 
-    pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Cfg>
+    pub fn build<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Out, Cfg>
     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<Codec>,
+            Response = ResponseItem<Codec>,
+            Error = C::Error,
+            InitError = C::Error,
+        >,
     {
         FramedService {
             connect: self.connect,
             handler: Rc::new(service.into_factory()),
-            disconnect: self.disconnect,
             _t: PhantomData,
         }
     }
 }
 
-pub struct FramedService<St, C, T, Io, Codec, Cfg> {
+pub struct FramedService<St, C, T, Io, Codec, Out, Cfg> {
     connect: C,
     handler: Rc<T>,
-    disconnect: Option<Rc<dyn Fn(St, bool)>>,
-    _t: PhantomData<(St, Io, Codec, Cfg)>,
+    _t: PhantomData<(St, Io, Codec, Out, Cfg)>,
 }
 
-impl<St, C, T, Io, Codec, Cfg> ServiceFactory for FramedService<St, C, T, Io, Codec, Cfg>
+impl<St, C, T, Io, Codec, Out, Cfg> ServiceFactory
+    for FramedService<St, C, T, Io, Codec, Out, Cfg>
 where
-    St: Clone + 'static,
     Io: AsyncRead + AsyncWrite,
     C: ServiceFactory<
         Config = (),
         Request = Connect<Io, Codec>,
-        Response = ConnectResult<Io, St, Codec>,
+        Response = ConnectResult<Io, St, Codec, Out>,
     >,
-    C::Error: 'static,
-    C::Future: 'static,
     T: ServiceFactory<
-            Config = St,
-            Request = RequestItem<St, Codec>,
-            Response = ResponseItem<Codec>,
-            Error = C::Error,
-            InitError = C::Error,
-        > + 'static,
+        Config = St,
+        Request = RequestItem<Codec>,
+        Response = ResponseItem<Codec>,
+        Error = C::Error,
+        InitError = C::Error,
+    >,
+    <T::Service as Service>::Error: 'static,
     <T::Service as Service>::Future: 'static,
     Codec: Decoder + Encoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
     type Config = Cfg;
     type Request = Io;
     type Response = ();
     type Error = ServiceError<C::Error, Codec>;
     type InitError = C::InitError;
-    type Service = FramedServiceImpl<St, C::Service, T, Io, Codec>;
-    type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
+    type Service = FramedServiceImpl<St, C::Service, T, Io, Codec, Out>;
+    type Future = FramedServiceResponse<St, C, T, Io, Codec, Out>;
 
     fn new_service(&self, _: Cfg) -> Self::Future {
-        let handler = self.handler.clone();
-        let disconnect = self.disconnect.clone();
-
         // create connect service and then create service impl
-        self.connect
-            .new_service(())
-            .map(move |result| {
-                result.map(move |connect| FramedServiceImpl {
-                    connect,
-                    handler,
-                    disconnect,
-                    _t: PhantomData,
-                })
-            })
-            .boxed_local()
+        FramedServiceResponse {
+            fut: self.connect.new_service(()),
+            handler: self.handler.clone(),
+        }
     }
 }
 
-pub struct FramedServiceImpl<St, C, T, Io, Codec> {
-    connect: C,
-    handler: Rc<T>,
-    disconnect: Option<Rc<dyn Fn(St, bool)>>,
-    _t: PhantomData<(St, Io, Codec)>,
-}
-
-impl<St, C, T, Io, Codec> Service for FramedServiceImpl<St, C, T, Io, Codec>
+#[pin_project::pin_project]
+pub struct FramedServiceResponse<St, C, T, Io, Codec, Out>
 where
-    St: Clone,
     Io: AsyncRead + AsyncWrite,
-    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: ServiceFactory<
+        Config = (),
+        Request = Connect<Io, Codec>,
+        Response = ConnectResult<Io, St, Codec, Out>,
+    >,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<Codec>,
         Response = ResponseItem<Codec>,
         Error = C::Error,
         InitError = C::Error,
     >,
+    <T::Service as Service>::Error: 'static,
     <T::Service as Service>::Future: 'static,
     Codec: Decoder + Encoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
+{
+    #[pin]
+    fut: C::Future,
+    handler: Rc<T>,
+}
+
+impl<St, C, T, Io, Codec, Out> Future for FramedServiceResponse<St, C, T, Io, Codec, Out>
+where
+    Io: AsyncRead + AsyncWrite,
+    C: ServiceFactory<
+        Config = (),
+        Request = Connect<Io, Codec>,
+        Response = ConnectResult<Io, St, Codec, Out>,
+    >,
+    T: ServiceFactory<
+        Config = St,
+        Request = RequestItem<Codec>,
+        Response = ResponseItem<Codec>,
+        Error = C::Error,
+        InitError = C::Error,
+    >,
+    <T::Service as Service>::Error: 'static,
+    <T::Service as Service>::Future: 'static,
+    Codec: Decoder + Encoder,
+    <Codec as Encoder>::Item: 'static,
+    <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
+{
+    type Output = Result<FramedServiceImpl<St, C::Service, T, Io, Codec, Out>, C::InitError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        let connect = ready!(this.fut.poll(cx))?;
+
+        Poll::Ready(Ok(FramedServiceImpl {
+            connect,
+            handler: this.handler.clone(),
+            _t: PhantomData,
+        }))
+    }
+}
+
+pub struct FramedServiceImpl<St, C, T, Io, Codec, Out> {
+    connect: C,
+    handler: Rc<T>,
+    _t: PhantomData<(St, Io, Codec, Out)>,
+}
+
+impl<St, C, T, Io, Codec, Out> Service for FramedServiceImpl<St, C, T, Io, Codec, Out>
+where
+    Io: AsyncRead + AsyncWrite,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
+    T: ServiceFactory<
+        Config = St,
+        Request = RequestItem<Codec>,
+        Response = ResponseItem<Codec>,
+        Error = C::Error,
+        InitError = C::Error,
+    >,
+    <T::Service as Service>::Error: 'static,
+    <T::Service as Service>::Future: 'static,
+    Codec: Decoder + Encoder,
+    <Codec as Encoder>::Item: 'static,
+    <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
     type Request = Io;
     type Response = ();
     type Error = ServiceError<C::Error, Codec>;
-    type Future = FramedServiceImplResponse<St, Io, Codec, C, T>;
+    type Future = FramedServiceImplResponse<St, Io, Codec, Out, 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(Rc::new(move |msg| {
-            let _ = tx.send(Ok(msg));
-        }));
         FramedServiceImplResponse {
             inner: FramedServiceImplResponseInner::Connect(
-                self.connect.call(Connect::new(req, sink.clone())),
+                self.connect.call(Connect::new(req)),
                 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, Out, C, T>
 where
-    St: Clone,
-    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<Codec>,
         Response = ResponseItem<Codec>,
         Error = C::Error,
         InitError = C::Error,
     >,
+    <T::Service as Service>::Error: 'static,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
     #[pin]
-    inner: FramedServiceImplResponseInner<St, Io, Codec, C, T>,
+    inner: FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>,
 }
 
-impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
+impl<St, Io, Codec, Out, C, T> Future for FramedServiceImplResponse<St, Io, Codec, Out, C, T>
 where
-    St: Clone,
-    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<Codec>,
         Response = ResponseItem<Codec>,
         Error = C::Error,
         InitError = C::Error,
     >,
+    <T::Service as Service>::Error: 'static,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
     type Output = Result<(), ServiceError<C::Error, Codec>>;
 
@@ -338,94 +337,71 @@ where
 }
 
 #[pin_project::pin_project]
-enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
+enum FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
 where
-    St: Clone,
-    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<Codec>,
         Response = ResponseItem<Codec>,
         Error = C::Error,
         InitError = C::Error,
     >,
+    <T::Service as Service>::Error: 'static,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
-    Connect(
-        #[pin] C::Future,
-        Rc<T>,
-        Option<Rc<dyn Fn(St, bool)>>,
-        Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
-    ),
-    Handler(
-        #[pin] T::Future,
-        Option<ConnectResult<Io, St, Codec>>,
-        Option<Rc<dyn Fn(St, bool)>>,
-        Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
-    ),
-    Dispatcher(Dispatcher<St, T::Service, Io, Codec>),
+    Connect(#[pin] C::Future, Rc<T>),
+    Handler(#[pin] T::Future, Option<Framed<Io, Codec>>, Option<Out>),
+    Dispatcher(Dispatcher<T::Service, Io, Codec, Out>),
 }
 
-impl<St, Io, Codec, C, T> FramedServiceImplResponseInner<St, Io, Codec, C, T>
+impl<St, Io, Codec, Out, C, T> FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
 where
-    St: Clone,
-    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec, Out>>,
     T: ServiceFactory<
         Config = St,
-        Request = RequestItem<St, Codec>,
+        Request = RequestItem<Codec>,
         Response = ResponseItem<Codec>,
         Error = C::Error,
         InitError = C::Error,
     >,
+    <T::Service as Service>::Error: 'static,
     <T::Service as Service>::Future: 'static,
     Io: AsyncRead + AsyncWrite,
     Codec: Encoder + Decoder,
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
+    Out: Stream<Item = <Codec as Encoder>::Item> + Unpin,
 {
     #[project]
     fn poll(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Either<
-        FramedServiceImplResponseInner<St, Io, Codec, C, T>,
+        FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>,
         Poll<Result<(), ServiceError<C::Error, Codec>>>,
     > {
         #[project]
         match self.project() {
-            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, rx) => {
+            FramedServiceImplResponseInner::Connect(fut, handler) => match fut.poll(cx) {
+                Poll::Ready(Ok(res)) => Either::Left(FramedServiceImplResponseInner::Handler(
+                    handler.new_service(res.state),
+                    Some(res.framed),
+                    res.out,
+                )),
+                Poll::Pending => Either::Right(Poll::Pending),
+                Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
+            },
+            FramedServiceImplResponseInner::Handler(fut, framed, out) => {
                 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(),
-                            ),
+                            Dispatcher::new(framed.take().unwrap(), handler, out.take()),
                         ))
                     }
                     Poll::Pending => Either::Right(Poll::Pending),

actix-ioframe/src/sink.rs

@@ -1,45 +0,0 @@
-use std::fmt;
-use std::rc::Rc;
-
-use actix_utils::oneshot;
-use futures::future::{Future, FutureExt};
-
-use crate::dispatcher::Message;
-
-pub struct Sink<T>(Rc<dyn Fn(Message<T>)>);
-
-impl<T> Clone for Sink<T> {
-    fn clone(&self) -> Self {
-        Sink(self.0.clone())
-    }
-}
-
-impl<T> Sink<T> {
-    pub(crate) fn new(tx: Rc<dyn Fn(Message<T>)>) -> Self {
-        Sink(tx)
-    }
-
-    /// Close connection
-    pub fn close(&self) {
-        (self.0)(Message::Close);
-    }
-
-    /// Close connection
-    pub fn wait_close(&self) -> impl Future<Output = ()> {
-        let (tx, rx) = oneshot::channel();
-        (self.0)(Message::WaitClose(tx));
-
-        rx.map(|_| ())
-    }
-
-    /// Send item
-    pub fn send(&self, item: T) {
-        (self.0)(Message::Item(item));
-    }
-}
-
-impl<T> fmt::Debug for Sink<T> {
-    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        fmt.debug_struct("Sink").finish()
-    }
-}

actix-ioframe/tests/test_server.rs

@@ -1,43 +1,49 @@
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
-use std::time::Duration;
+use std::cell::Cell;
+use std::rc::Rc;
 
 use actix_codec::BytesCodec;
-use actix_rt::time::delay_for;
-use actix_service::{fn_service, Service};
+use actix_service::{fn_factory_with_config, fn_service, IntoService, Service};
 use actix_testing::TestServer;
+use actix_utils::mpsc;
+use bytes::{Bytes, BytesMut};
 use futures::future::ok;
 
-use actix_ioframe::{Builder, Connect};
+use actix_ioframe::{Builder, Connect, FactoryBuilder};
 
 #[derive(Clone)]
-struct State;
+struct State(Option<mpsc::Sender<Bytes>>);
 
 #[actix_rt::test]
-async fn test_disconnect() -> std::io::Result<()> {
-    let disconnect = Arc::new(AtomicBool::new(false));
-    let disconnect1 = disconnect.clone();
+async fn test_basic() {
+    let client_item = Rc::new(Cell::new(false));
 
     let srv = TestServer::with(move || {
-        let disconnect1 = disconnect1.clone();
-
-        Builder::new()
-            .factory(fn_service(|conn: Connect<_, _>| {
-                ok(conn.codec(BytesCodec).state(State))
-            }))
-            .disconnect(move |_, _| {
-                disconnect1.store(true, Ordering::Relaxed);
-            })
-            .finish(fn_service(|_t| ok(None)))
+        FactoryBuilder::new(fn_service(|conn: Connect<_, _>| {
+            ok(conn.codec(BytesCodec).state(State(None)))
+        }))
+        // echo
+        .build(fn_service(|t: BytesMut| ok(Some(t.freeze()))))
     });
 
-    let mut client = Builder::new()
-        .service(|conn: Connect<_, _>| {
-            let conn = conn.codec(BytesCodec).state(State);
-            conn.sink().close();
-            ok(conn)
+    let item = client_item.clone();
+    let mut client = Builder::new(fn_service(move |conn: Connect<_, _>| {
+        async move {
+            let (tx, rx) = mpsc::channel();
+            let _ = tx.send(Bytes::from_static(b"Hello"));
+            Ok(conn.codec(BytesCodec).out(rx).state(State(Some(tx))))
+        }
+    }))
+    .build(fn_factory_with_config(move |mut cfg: State| {
+        let item = item.clone();
+        ok((move |t: BytesMut| {
+            assert_eq!(t.freeze(), Bytes::from_static(b"Hello"));
+            item.set(true);
+            // drop Sender, which will close connection
+            cfg.0.take();
+            ok::<_, ()>(None)
         })
-        .finish(fn_service(|_t| ok(None)));
+        .into_service())
+    }));
 
     let conn = actix_connect::default_connector()
         .call(actix_connect::Connect::with(String::new(), srv.addr()))
@@ -45,8 +51,5 @@ async fn test_disconnect() -> std::io::Result<()> {
         .unwrap();
 
     client.call(conn.into_parts().0).await.unwrap();
-    let _ = delay_for(Duration::from_millis(100)).await;
-    assert!(disconnect.load(Ordering::Relaxed));
-
-    Ok(())
+    assert!(client_item.get());
 }

actix-rt/src/arbiter.rs

@@ -42,8 +42,8 @@ impl fmt::Debug for ArbiterCommand {
 
 #[derive(Debug)]
 /// Arbiters provide an asynchronous execution environment for actors, functions
-/// and futures. When an Arbiter is created, they spawn a new OS thread, and
-/// host an event loop. Some Arbiter functions execute on the current thread.
+/// and futures. When an Arbiter is created, it spawns a new OS thread, and
+/// hosts an event loop. Some Arbiter functions execute on the current thread.
 pub struct Arbiter {
     sender: UnboundedSender<ArbiterCommand>,
     thread_handle: Option<thread::JoinHandle<()>>,

actix-server/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [1.0.1] - 2019-12-29
+
+### Changed
+
+* Rename `.start()` method to `.run()`
+
 ## [1.0.0] - 2019-12-11
 
 ### Changed

actix-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-server"
-version = "1.0.0"
+version = "1.0.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix server - General purpose tcp server"
 keywords = ["network", "framework", "async", "futures"]
@@ -21,10 +21,10 @@ path = "src/lib.rs"
 default = []
 
 [dependencies]
-actix-service = "1.0.0"
+actix-service = "1.0.1"
 actix-rt = "1.0.0"
 actix-codec = "0.2.0"
-actix-utils = "1.0.0"
+actix-utils = "1.0.4"
 
 log = "0.4"
 num_cpus = "1.11"
@@ -38,5 +38,5 @@ mio-uds = { version = "0.6.7" }
 
 [dev-dependencies]
 bytes = "0.5"
-env_logger = "0.6"
+env_logger = "0.7"
 actix-testing = "1.0.0"

actix-server/src/builder.rs

@@ -31,7 +31,7 @@ pub struct ServerBuilder {
     backlog: i32,
     workers: Vec<(usize, WorkerClient)>,
     services: Vec<Box<dyn InternalServiceFactory>>,
-    sockets: Vec<(Token, StdListener)>,
+    sockets: Vec<(Token, String, StdListener)>,
     accept: AcceptLoop,
     exit: bool,
     shutdown_timeout: Duration,
@@ -146,8 +146,8 @@ impl ServerBuilder {
             let mut srv = ConfiguredService::new(apply);
             for (name, lst) in cfg.services {
                 let token = self.token.next();
-                srv.stream(token, name, lst.local_addr()?);
-                self.sockets.push((token, StdListener::Tcp(lst)));
+                srv.stream(token, name.clone(), lst.local_addr()?);
+                self.sockets.push((token, name, StdListener::Tcp(lst)));
             }
             self.services.push(Box::new(srv));
         }
@@ -172,7 +172,8 @@ impl ServerBuilder {
                 factory.clone(),
                 lst.local_addr()?,
             ));
-            self.sockets.push((token, StdListener::Tcp(lst)));
+            self.sockets
+                .push((token, name.as_ref().to_string(), StdListener::Tcp(lst)));
         }
         Ok(self)
     }
@@ -222,7 +223,8 @@ impl ServerBuilder {
             factory.clone(),
             addr,
         ));
-        self.sockets.push((token, StdListener::Uds(lst)));
+        self.sockets
+            .push((token, name.as_ref().to_string(), StdListener::Uds(lst)));
         Ok(self)
     }
 
@@ -243,36 +245,18 @@ impl ServerBuilder {
             factory,
             lst.local_addr()?,
         ));
-        self.sockets.push((token, StdListener::Tcp(lst)));
+        self.sockets
+            .push((token, name.as_ref().to_string(), StdListener::Tcp(lst)));
         Ok(self)
     }
 
-    /// Spawn new thread and start listening for incoming connections.
-    ///
-    /// This method spawns new thread and starts new actix system. Other than
-    /// that it is similar to `start()` method. This method blocks.
-    ///
-    /// This methods panics if no socket addresses get bound.
-    ///
-    /// ```rust,ignore
-    /// use actix_web::*;
-    ///
-    /// fn main() -> std::io::Result<()> {
-    ///     Server::new().
-    ///         .service(
-    ///            HttpServer::new(|| App::new().service(web::service("/").to(|| HttpResponse::Ok())))
-    ///                .bind("127.0.0.1:0")
-    ///         .run()
-    /// }
-    /// ```
-    pub fn run(self) -> io::Result<()> {
-        let sys = System::new("http-server");
-        self.start();
-        sys.run()
+    #[doc(hidden)]
+    pub fn start(self) -> Server {
+        self.run()
     }
 
     /// Starts processing incoming connections and return server controller.
-    pub fn start(mut self) -> Server {
+    pub fn run(mut self) -> Server {
         if self.sockets.is_empty() {
             panic!("Server should have at least one bound socket");
         } else {
@@ -288,10 +272,15 @@ impl ServerBuilder {
 
             // start accept thread
             for sock in &self.sockets {
-                info!("Starting server on {}", sock.1);
+                info!("Starting \"{}\" service on {}", sock.1, sock.2);
             }
-            self.accept
-                .start(mem::replace(&mut self.sockets, Vec::new()), workers);
+            self.accept.start(
+                mem::replace(&mut self.sockets, Vec::new())
+                    .into_iter()
+                    .map(|t| (t.0, t.2))
+                    .collect(),
+                workers,
+            );
 
             // handle signals
             if !self.no_signals {

actix-service/CHANGES.md

@@ -1,5 +1,18 @@
 # Changes
 
+## [1.0.3] - 2020-01-15
+
+### Fixed
+
+* Fixed unsoundness in `AndThenService` impl
+
+## [1.0.2] - 2020-01-08
+
+### Added
+
+* Add `into_service` helper function
+
+
 ## [1.0.1] - 2019-12-22
 
 ### Changed

actix-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-service"
-version = "1.0.1"
+version = "1.0.3"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix service"
 keywords = ["network", "framework", "async", "futures"]
@@ -10,7 +10,6 @@ documentation = "https://docs.rs/actix-service/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
 
 [badges]
 travis-ci = { repository = "actix/actix-service", branch = "master" }

actix-service/src/lib.rs

@@ -57,7 +57,7 @@ pub use self::transform::{apply, Transform};
 ///
 ///      fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { ... }
 ///
-///      fn call(&mut self) -> Self::Future { ... }
+///      fn call(&mut self, req: Self::Request) -> Self::Future { ... }
 /// }
 /// ```
 ///
@@ -82,7 +82,7 @@ pub trait Service {
 
     /// Returns `Ready` when the service is able to process requests.
     ///
-    /// If the service is at capacity, then `NotReady` is returned and the task
+    /// If the service is at capacity, then `Pending` is returned and the task
     /// is notified when the service becomes ready again. This function is
     /// expected to be called while on a task.
     ///
@@ -351,6 +351,15 @@ where
     }
 }
 
+/// Convert object of type `T` to a service `S`
+pub fn into_service<T, S>(tp: T) -> S
+where
+    S: Service,
+    T: IntoService<S>,
+{
+    tp.into_service()
+}
+
 pub mod dev {
     pub use crate::and_then::{AndThenService, AndThenServiceFactory};
     pub use crate::and_then_apply_fn::{AndThenApplyFn, AndThenApplyFnFactory};

actix-service/src/pipeline.rs

@@ -50,7 +50,7 @@ impl<T: Service> Pipeline<T> {
     where
         Self: Sized,
         F: IntoService<U>,
-        U: Service<Request = T::Response, Error = T::Error>,
+        U: Service<Request = T::Response, Error = T::Error> + 'static,
     {
         Pipeline {
             service: AndThenService::new(self.service, service.into_service()),
@@ -69,7 +69,7 @@ impl<T: Service> Pipeline<T> {
     where
         Self: Sized,
         I: IntoService<U>,
-        U: Service,
+        U: Service + 'static,
         F: FnMut(T::Response, &mut U) -> Fut,
         Fut: Future<Output = Result<Res, Err>>,
         Err: From<T::Error> + From<U::Error>,
@@ -88,7 +88,7 @@ impl<T: Service> Pipeline<T> {
     where
         Self: Sized,
         F: IntoService<U>,
-        U: Service<Request = Result<T::Response, T::Error>, Error = T::Error>,
+        U: Service<Request = Result<T::Response, T::Error>, Error = T::Error> + 'static,
     {
         Pipeline {
             service: ThenService::new(self.service, service.into_service()),
@@ -179,6 +179,7 @@ impl<T: ServiceFactory> PipelineFactory<T> {
             Error = T::Error,
             InitError = T::InitError,
         >,
+        U::Service: 'static,
     {
         PipelineFactory {
             factory: AndThenServiceFactory::new(self.factory, factory.into_factory()),
@@ -199,6 +200,7 @@ impl<T: ServiceFactory> PipelineFactory<T> {
         T::Config: Clone,
         I: IntoServiceFactory<U>,
         U: ServiceFactory<Config = T::Config, InitError = T::InitError>,
+        U::Service: 'static,
         F: FnMut(T::Response, &mut U::Service) -> Fut + Clone,
         Fut: Future<Output = Result<Res, Err>>,
         Err: From<T::Error> + From<U::Error>,
@@ -225,6 +227,7 @@ impl<T: ServiceFactory> PipelineFactory<T> {
             Error = T::Error,
             InitError = T::InitError,
         >,
+        U::Service: 'static,
     {
         PipelineFactory {
             factory: ThenServiceFactory::new(self.factory, factory.into_factory()),

actix-tracing/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "actix-tracing"
+version = "0.1.0"
+authors = ["Rajasekharan Vengalil <avranju@gmail.com>"]
+description = "Support for tokio tracing with Actix services"
+keywords = ["network", "framework", "tracing"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-tracing/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+edition = "2018"
+workspace = ".."
+
+[lib]
+name = "actix_tracing"
+path = "src/lib.rs"
+
+[dependencies]
+actix-service = "1.0"
+futures-util = "0.3"
+tracing = "0.1"
+tracing-futures = "0.2"
+
+[dev_dependencies]
+actix-rt = "1.0"
+slab = "0.4"

actix-tracing/src/lib.rs

@@ -0,0 +1,261 @@
+//! Actix tracing - support for tokio tracing with Actix services.
+#![deny(rust_2018_idioms, warnings)]
+
+use std::marker::PhantomData;
+use std::task::{Context, Poll};
+
+use actix_service::{
+    apply, dev::ApplyTransform, IntoServiceFactory, Service, ServiceFactory, Transform,
+};
+use futures_util::future::{ok, Either, Ready};
+use tracing_futures::{Instrument, Instrumented};
+
+/// A `Service` implementation that automatically enters/exits tracing spans
+/// for the wrapped inner service.
+#[derive(Clone)]
+pub struct TracingService<S, F> {
+    inner: S,
+    make_span: F,
+}
+
+impl<S, F> TracingService<S, F> {
+    pub fn new(inner: S, make_span: F) -> Self {
+        TracingService { inner, make_span }
+    }
+}
+
+impl<S, F> Service for TracingService<S, F>
+where
+    S: Service,
+    F: Fn(&S::Request) -> Option<tracing::Span>,
+{
+    type Request = S::Request;
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = Either<S::Future, Instrumented<S::Future>>;
+
+    fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(ctx)
+    }
+
+    fn call(&mut self, req: Self::Request) -> Self::Future {
+        let span = (self.make_span)(&req);
+        let _enter = span.as_ref().map(|s| s.enter());
+
+        let fut = self.inner.call(req);
+
+        // make a child span to track the future's execution
+        if let Some(span) = span
+            .clone()
+            .map(|span| tracing::span!(parent: &span, tracing::Level::INFO, "future"))
+        {
+            Either::Right(fut.instrument(span))
+        } else {
+            Either::Left(fut)
+        }
+    }
+}
+
+/// A `Transform` implementation that wraps services with a [`TracingService`].
+///
+/// [`TracingService`]: struct.TracingService.html
+pub struct TracingTransform<S, U, F> {
+    make_span: F,
+    _p: PhantomData<fn(S, U)>,
+}
+
+impl<S, U, F> TracingTransform<S, U, F> {
+    pub fn new(make_span: F) -> Self {
+        TracingTransform {
+            make_span,
+            _p: PhantomData,
+        }
+    }
+}
+
+impl<S, U, F> Transform<S> for TracingTransform<S, U, F>
+where
+    S: Service,
+    U: ServiceFactory<
+        Request = S::Request,
+        Response = S::Response,
+        Error = S::Error,
+        Service = S,
+    >,
+    F: Fn(&S::Request) -> Option<tracing::Span> + Clone,
+{
+    type Request = S::Request;
+    type Response = S::Response;
+    type Error = S::Error;
+    type Transform = TracingService<S, F>;
+    type InitError = U::InitError;
+    type Future = Ready<Result<Self::Transform, Self::InitError>>;
+
+    fn new_transform(&self, service: S) -> Self::Future {
+        ok(TracingService::new(service, self.make_span.clone()))
+    }
+}
+
+/// Wraps the provided service factory with a transform that automatically
+/// enters/exits the given span.
+///
+/// The span to be entered/exited can be provided via a closure. The closure
+/// is passed in a reference to the request being handled by the service.
+///
+/// For example:
+/// ```rust,ignore
+/// let traced_service = trace(
+///     web_service,
+///     |req: &Request| Some(span!(Level::INFO, "request", req.id))
+/// );
+/// ```
+pub fn trace<S, U, F>(
+    service_factory: U,
+    make_span: F,
+) -> ApplyTransform<TracingTransform<S::Service, S, F>, S>
+where
+    S: ServiceFactory,
+    F: Fn(&S::Request) -> Option<tracing::Span> + Clone,
+    U: IntoServiceFactory<S>,
+{
+    apply(
+        TracingTransform::new(make_span),
+        service_factory.into_factory(),
+    )
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    use std::cell::RefCell;
+    use std::collections::{BTreeMap, BTreeSet};
+    use std::sync::{Arc, RwLock};
+
+    use actix_service::{fn_factory, fn_service};
+    use slab::Slab;
+    use tracing::{span, Event, Level, Metadata, Subscriber};
+
+    thread_local! {
+        static SPAN: RefCell<Vec<span::Id>> = RefCell::new(Vec::new());
+    }
+
+    #[derive(Default)]
+    struct Stats {
+        entered_spans: BTreeSet<u64>,
+        exited_spans: BTreeSet<u64>,
+        events_count: BTreeMap<u64, usize>,
+    }
+
+    #[derive(Default)]
+    struct Inner {
+        spans: Slab<&'static Metadata<'static>>,
+        stats: Stats,
+    }
+
+    #[derive(Clone, Default)]
+    struct TestSubscriber {
+        inner: Arc<RwLock<Inner>>,
+    }
+
+    impl Subscriber for TestSubscriber {
+        fn enabled(&self, _metadata: &Metadata<'_>) -> bool {
+            true
+        }
+
+        fn new_span(&self, span: &span::Attributes<'_>) -> span::Id {
+            let id = self.inner.write().unwrap().spans.insert(span.metadata());
+            span::Id::from_u64(id as u64 + 1)
+        }
+
+        fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {}
+
+        fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {}
+
+        fn event(&self, event: &Event<'_>) {
+            let id = event
+                .parent()
+                .cloned()
+                .or_else(|| SPAN.with(|current_span| current_span.borrow().last().cloned()))
+                .unwrap();
+
+            *self
+                .inner
+                .write()
+                .unwrap()
+                .stats
+                .events_count
+                .entry(id.into_u64())
+                .or_insert(0) += 1;
+        }
+
+        fn enter(&self, span: &span::Id) {
+            self.inner
+                .write()
+                .unwrap()
+                .stats
+                .entered_spans
+                .insert(span.into_u64());
+
+            SPAN.with(|current_span| {
+                current_span.borrow_mut().push(span.clone());
+            });
+        }
+
+        fn exit(&self, span: &span::Id) {
+            self.inner
+                .write()
+                .unwrap()
+                .stats
+                .exited_spans
+                .insert(span.into_u64());
+
+            // we are guaranteed that on any given thread, spans are exited in reverse order
+            SPAN.with(|current_span| {
+                let leaving = current_span
+                    .borrow_mut()
+                    .pop()
+                    .expect("told to exit span when not in span");
+                assert_eq!(
+                    &leaving, span,
+                    "told to exit span that was not most recently entered"
+                );
+            });
+        }
+    }
+
+    #[actix_rt::test]
+    async fn service_call() {
+        let service_factory = fn_factory(|| {
+            ok::<_, ()>(fn_service(|req: &'static str| {
+                tracing::event!(Level::TRACE, "It's happening - {}!", req);
+                ok::<_, ()>(())
+            }))
+        });
+
+        let subscriber = TestSubscriber::default();
+        let _guard = tracing::subscriber::set_default(subscriber.clone());
+
+        let span_svc = span!(Level::TRACE, "span_svc");
+        let trace_service_factory = trace(service_factory, |_: &&str| Some(span_svc.clone()));
+        let mut service = trace_service_factory.new_service(()).await.unwrap();
+        service.call("boo").await.unwrap();
+
+        let id = span_svc.id().unwrap().into_u64();
+        assert!(subscriber
+            .inner
+            .read()
+            .unwrap()
+            .stats
+            .entered_spans
+            .contains(&id));
+        assert!(subscriber
+            .inner
+            .read()
+            .unwrap()
+            .stats
+            .exited_spans
+            .contains(&id));
+        assert_eq!(subscriber.inner.read().unwrap().stats.events_count[&id], 1);
+    }
+}

actix-utils/CHANGES.md

@@ -1,5 +1,15 @@
 # Changes
 
+## [1.0.6] - 2020-01-08
+
+* Add `Clone` impl for `condition::Waiter`
+
+## [1.0.5] - 2020-01-08
+
+* Add `Condition` type.
+
+* Add `Pool` of one-shot's.
+
 ## [1.0.4] - 2019-12-20
 
 * Add methods to check `LocalWaker` registration state.

actix-utils/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-utils"
-version = "1.0.4"
+version = "1.0.6"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix utils - various actix net related services"
 keywords = ["network", "framework", "async", "futures"]
@@ -16,11 +16,13 @@ name = "actix_utils"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "1.0.0"
+actix-service = "1.0.1"
 actix-rt = "1.0.0"
 actix-codec = "0.2.0"
+bitflags = "1.2"
 bytes = "0.5.3"
 either = "1.5.3"
 futures = "0.3.1"
 pin-project = "0.4.6"
 log = "0.4"
+slab = "0.4"

actix-utils/src/condition.rs

@@ -0,0 +1,127 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use slab::Slab;
+
+use crate::cell::Cell;
+use crate::task::LocalWaker;
+
+/// Condition allows to notify multiple receivers at the same time
+pub struct Condition(Cell<Inner>);
+
+struct Inner {
+    data: Slab<Option<LocalWaker>>,
+}
+
+impl Default for Condition {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Condition {
+    pub fn new() -> Condition {
+        Condition(Cell::new(Inner { data: Slab::new() }))
+    }
+
+    /// Get condition waiter
+    pub fn wait(&mut self) -> Waiter {
+        let token = self.0.get_mut().data.insert(None);
+        Waiter {
+            token,
+            inner: self.0.clone(),
+        }
+    }
+
+    /// Notify all waiters
+    pub fn notify(&self) {
+        let inner = self.0.get_ref();
+        for item in inner.data.iter() {
+            if let Some(waker) = item.1 {
+                waker.wake();
+            }
+        }
+    }
+}
+
+impl Drop for Condition {
+    fn drop(&mut self) {
+        self.notify()
+    }
+}
+
+#[must_use = "Waiter do nothing unless polled"]
+pub struct Waiter {
+    token: usize,
+    inner: Cell<Inner>,
+}
+
+impl Clone for Waiter {
+    fn clone(&self) -> Self {
+        let token = unsafe { self.inner.get_mut_unsafe() }.data.insert(None);
+        Waiter {
+            token,
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl Future for Waiter {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+
+        let inner = unsafe { this.inner.get_mut().data.get_unchecked_mut(this.token) };
+        if inner.is_none() {
+            let waker = LocalWaker::default();
+            waker.register(cx.waker());
+            *inner = Some(waker);
+            Poll::Pending
+        } else if inner.as_mut().unwrap().register(cx.waker()) {
+            Poll::Pending
+        } else {
+            Poll::Ready(())
+        }
+    }
+}
+
+impl Drop for Waiter {
+    fn drop(&mut self) {
+        self.inner.get_mut().data.remove(self.token);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::lazy;
+
+    #[actix_rt::test]
+    async fn test_condition() {
+        let mut cond = Condition::new();
+        let mut waiter = cond.wait();
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut waiter).poll(cx)).await,
+            Poll::Pending
+        );
+        cond.notify();
+        assert_eq!(waiter.await, ());
+
+        let mut waiter = cond.wait();
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut waiter).poll(cx)).await,
+            Poll::Pending
+        );
+        let mut waiter2 = waiter.clone();
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut waiter2).poll(cx)).await,
+            Poll::Pending
+        );
+
+        drop(cond);
+        assert_eq!(waiter.await, ());
+        assert_eq!(waiter2.await, ());
+    }
+}

actix-utils/src/lib.rs

@@ -3,6 +3,7 @@
 #![allow(clippy::type_complexity)]
 
 mod cell;
+pub mod condition;
 pub mod counter;
 pub mod either;
 pub mod framed;

actix-utils/src/oneshot.rs

@@ -4,6 +4,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 pub use futures::channel::oneshot::Canceled;
+use slab::Slab;
 
 use crate::cell::Cell;
 use crate::task::LocalWaker;
@@ -21,6 +22,11 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     (tx, rx)
 }
 
+/// Creates a new futures-aware, pool of one-shot's.
+pub fn pool<T>() -> Pool<T> {
+    Pool(Cell::new(Slab::new()))
+}
+
 /// Represents the completion half of a oneshot through which the result of a
 /// computation is signaled.
 #[derive(Debug)]
@@ -77,9 +83,7 @@ impl<T> Sender<T> {
 
 impl<T> Drop for Sender<T> {
     fn drop(&mut self) {
-        if self.inner.strong_count() == 2 {
-            self.inner.get_ref().rx_task.wake();
-        };
+        self.inner.get_ref().rx_task.wake();
     }
 }
 
@@ -104,6 +108,148 @@ impl<T> Future for Receiver<T> {
     }
 }
 
+/// Futures-aware, pool of one-shot's.
+pub struct Pool<T>(Cell<Slab<PoolInner<T>>>);
+
+bitflags::bitflags! {
+    pub struct Flags: u8 {
+        const SENDER = 0b0000_0001;
+        const RECEIVER = 0b0000_0010;
+    }
+}
+
+#[derive(Debug)]
+struct PoolInner<T> {
+    flags: Flags,
+    value: Option<T>,
+    waker: LocalWaker,
+}
+
+impl<T> Pool<T> {
+    pub fn channel(&mut self) -> (PSender<T>, PReceiver<T>) {
+        let token = self.0.get_mut().insert(PoolInner {
+            flags: Flags::all(),
+            value: None,
+            waker: LocalWaker::default(),
+        });
+
+        (
+            PSender {
+                token,
+                inner: self.0.clone(),
+            },
+            PReceiver {
+                token,
+                inner: self.0.clone(),
+            },
+        )
+    }
+}
+
+impl<T> Clone for Pool<T> {
+    fn clone(&self) -> Self {
+        Pool(self.0.clone())
+    }
+}
+
+/// Represents the completion half of a oneshot through which the result of a
+/// computation is signaled.
+#[derive(Debug)]
+pub struct PSender<T> {
+    token: usize,
+    inner: Cell<Slab<PoolInner<T>>>,
+}
+
+/// A future representing the completion of a computation happening elsewhere in
+/// memory.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless polled"]
+pub struct PReceiver<T> {
+    token: usize,
+    inner: Cell<Slab<PoolInner<T>>>,
+}
+
+// The oneshots do not ever project Pin to the inner T
+impl<T> Unpin for PReceiver<T> {}
+impl<T> Unpin for PSender<T> {}
+
+impl<T> PSender<T> {
+    /// Completes this oneshot with a successful result.
+    ///
+    /// This function will consume `self` and indicate to the other end, the
+    /// `Receiver`, that the error provided is the result of the computation this
+    /// represents.
+    ///
+    /// If the value is successfully enqueued for the remote end to receive,
+    /// 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(mut self, val: T) -> Result<(), T> {
+        let inner = unsafe { self.inner.get_mut().get_unchecked_mut(self.token) };
+
+        if inner.flags.contains(Flags::RECEIVER) {
+            inner.value = Some(val);
+            inner.waker.wake();
+            Ok(())
+        } else {
+            Err(val)
+        }
+    }
+
+    /// Tests to see whether this `Sender`'s corresponding `Receiver`
+    /// has gone away.
+    pub fn is_canceled(&self) -> bool {
+        !unsafe { self.inner.get_ref().get_unchecked(self.token) }
+            .flags
+            .contains(Flags::RECEIVER)
+    }
+}
+
+impl<T> Drop for PSender<T> {
+    fn drop(&mut self) {
+        let inner = unsafe { self.inner.get_mut().get_unchecked_mut(self.token) };
+        if inner.flags.contains(Flags::RECEIVER) {
+            inner.waker.wake();
+            inner.flags.remove(Flags::SENDER);
+        } else {
+            self.inner.get_mut().remove(self.token);
+        }
+    }
+}
+
+impl<T> Drop for PReceiver<T> {
+    fn drop(&mut self) {
+        let inner = unsafe { self.inner.get_mut().get_unchecked_mut(self.token) };
+        if inner.flags.contains(Flags::SENDER) {
+            inner.flags.remove(Flags::RECEIVER);
+        } else {
+            self.inner.get_mut().remove(self.token);
+        }
+    }
+}
+
+impl<T> Future for PReceiver<T> {
+    type Output = Result<T, Canceled>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+        let inner = unsafe { this.inner.get_mut().get_unchecked_mut(this.token) };
+
+        // If we've got a value, then skip the logic below as we're done.
+        if let Some(val) = inner.value.take() {
+            return Poll::Ready(Ok(val));
+        }
+
+        // Check if sender is dropped and return error if it is.
+        if !inner.flags.contains(Flags::SENDER) {
+            Poll::Ready(Err(Canceled))
+        } else {
+            inner.waker.register(cx.waker());
+            Poll::Pending
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -135,4 +281,31 @@ mod tests {
         drop(tx);
         assert!(rx.await.is_err());
     }
+
+    #[actix_rt::test]
+    async fn test_pool() {
+        let (tx, rx) = pool().channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.await.unwrap(), "test");
+
+        let (tx, rx) = pool().channel();
+        assert!(!tx.is_canceled());
+        drop(rx);
+        assert!(tx.is_canceled());
+        assert!(tx.send("test").is_err());
+
+        let (tx, rx) = pool::<&'static str>().channel();
+        drop(tx);
+        assert!(rx.await.is_err());
+
+        let (tx, mut rx) = pool::<&'static str>().channel();
+        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) = pool::<&'static str>().channel();
+        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
+        drop(tx);
+        assert!(rx.await.is_err());
+    }
 }

router/CHANGES.txt

@@ -1,5 +1,17 @@
 # Changes
 
+## [0.2.4] - 2019-12-31
+
+* Add `ResourceDef::resource_path_named()` path generation method
+
+## [0.2.3] - 2019-12-25
+
+* Add impl `IntoPattern` for `&String`
+
+## [0.2.2] - 2019-12-25
+
+* Use `IntoPattern` for `RouterBuilder::path()`
+
 ## [0.2.1] - 2019-12-25
 
 * Add `IntoPattern` trait

router/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-router"
-version = "0.2.1"
+version = "0.2.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Path router"
 keywords = ["actix"]

router/src/lib.rs

@@ -53,6 +53,16 @@ impl IntoPattern for String {
     }
 }
 
+impl<'a> IntoPattern for &'a String {
+    fn is_single(&self) -> bool {
+        true
+    }
+
+    fn patterns(&self) -> Vec<String> {
+        vec![self.as_str().to_string()]
+    }
+}
+
 impl<'a> IntoPattern for &'a str {
     fn is_single(&self) -> bool {
         true

router/src/path.rs

@@ -1,5 +1,4 @@
 use std::ops::Index;
-use std::rc::Rc;
 
 use serde::de;
 
@@ -19,7 +18,7 @@ pub(crate) enum PathItem {
 pub struct Path<T> {
     path: T,
     pub(crate) skip: u16,
-    pub(crate) segments: Vec<(Rc<String>, PathItem)>,
+    pub(crate) segments: Vec<(&'static str, PathItem)>,
 }
 
 impl<T: Default> Default for Path<T> {
@@ -96,7 +95,7 @@ impl<T: ResourcePath> Path<T> {
         self.skip += n;
     }
 
-    pub(crate) fn add(&mut self, name: Rc<String>, value: PathItem) {
+    pub(crate) fn add(&mut self, name: &'static str, value: PathItem) {
         match value {
             PathItem::Static(s) => self.segments.push((name, PathItem::Static(s))),
             PathItem::Segment(begin, end) => self
@@ -106,9 +105,8 @@ impl<T: ResourcePath> Path<T> {
     }
 
     #[doc(hidden)]
-    pub fn add_static(&mut self, name: &str, value: &'static str) {
-        self.segments
-            .push((Rc::new(name.to_string()), PathItem::Static(value)));
+    pub fn add_static(&mut self, name: &'static str, value: &'static str) {
+        self.segments.push((name, PathItem::Static(value)));
     }
 
     #[inline]
@@ -126,7 +124,7 @@ impl<T: ResourcePath> Path<T> {
     /// Get matched parameter by name without type conversion
     pub fn get(&self, key: &str) -> Option<&str> {
         for item in self.segments.iter() {
-            if key == item.0.as_str() {
+            if key == item.0 {
                 return match item.1 {
                     PathItem::Static(ref s) => Some(&s),
                     PathItem::Segment(s, e) => {

router/src/resource.rs

@@ -1,6 +1,6 @@
 use std::cmp::min;
+use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
-use std::rc::Rc;
 
 use regex::{escape, Regex, RegexSet};
 
@@ -31,8 +31,8 @@ enum PatternElement {
 enum PatternType {
     Static(String),
     Prefix(String),
-    Dynamic(Regex, Vec<Rc<String>>, usize),
-    DynamicSet(RegexSet, Vec<(Regex, Vec<Rc<String>>, usize)>),
+    Dynamic(Regex, Vec<&'static str>, usize),
+    DynamicSet(RegexSet, Vec<(Regex, Vec<&'static str>, usize)>),
 }
 
 impl ResourceDef {
@@ -58,7 +58,9 @@ impl ResourceDef {
                 // actix creates one router per thread
                 let names: Vec<_> = re
                     .capture_names()
-                    .filter_map(|name| name.map(|name| Rc::new(name.to_owned())))
+                    .filter_map(|name| {
+                        name.map(|name| Box::leak(Box::new(name.to_owned())).as_str())
+                    })
                     .collect();
                 data.push((re, names, len));
                 re_set.push(pattern);
@@ -117,7 +119,9 @@ impl ResourceDef {
             // actix creates one router per thread
             let names = re
                 .capture_names()
-                .filter_map(|name| name.map(|name| Rc::new(name.to_owned())))
+                .filter_map(|name| {
+                    name.map(|name| Box::leak(Box::new(name.to_owned())).as_str())
+                })
                 .collect();
             PatternType::Dynamic(re, names, len)
         } else if for_prefix {
@@ -489,6 +493,41 @@ impl ResourceDef {
         true
     }
 
+    /// Build resource path from elements. Returns `true` on success.
+    pub fn resource_path_named<K, V, S>(
+        &self,
+        path: &mut String,
+        elements: &HashMap<K, V, S>,
+    ) -> bool
+    where
+        K: std::borrow::Borrow<str> + Eq + Hash,
+        V: AsRef<str>,
+        S: std::hash::BuildHasher,
+    {
+        match self.tp {
+            PatternType::Prefix(ref p) => path.push_str(p),
+            PatternType::Static(ref p) => path.push_str(p),
+            PatternType::Dynamic(..) => {
+                for el in &self.elements {
+                    match *el {
+                        PatternElement::Str(ref s) => path.push_str(s),
+                        PatternElement::Var(ref name) => {
+                            if let Some(val) = elements.get(name) {
+                                path.push_str(val.as_ref())
+                            } else {
+                                return false;
+                            }
+                        }
+                    }
+                }
+            }
+            PatternType::DynamicSet(..) => {
+                return false;
+            }
+        }
+        true
+    }
+
     fn parse_param(pattern: &str) -> (PatternElement, String, &str, bool) {
         const DEFAULT_PATTERN: &str = "[^/]+";
         const DEFAULT_PATTERN_TAIL: &str = ".*";
@@ -855,4 +894,50 @@ mod tests {
         assert_eq!(&path["name"], "test2");
         assert_eq!(&path[0], "test2");
     }
+
+    #[test]
+    fn test_resource_path() {
+        let mut s = String::new();
+        let resource = ResourceDef::new("/user/{item1}/test");
+        assert!(resource.resource_path(&mut s, &mut (&["user1"]).into_iter()));
+        assert_eq!(s, "/user/user1/test");
+
+        let mut s = String::new();
+        let resource = ResourceDef::new("/user/{item1}/{item2}/test");
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).into_iter()));
+        assert_eq!(s, "/user/item/item2/test");
+
+        let mut s = String::new();
+        let resource = ResourceDef::new("/user/{item1}/{item2}");
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).into_iter()));
+        assert_eq!(s, "/user/item/item2");
+
+        let mut s = String::new();
+        let resource = ResourceDef::new("/user/{item1}/{item2}/");
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).into_iter()));
+        assert_eq!(s, "/user/item/item2/");
+
+        let mut s = String::new();
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).into_iter()));
+
+        let mut s = String::new();
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).into_iter()));
+        assert_eq!(s, "/user/item/item2/");
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).into_iter()));
+
+        let mut s = String::new();
+        assert!(resource.resource_path(&mut s, &mut vec!["item", "item2"].into_iter()));
+        assert_eq!(s, "/user/item/item2/");
+
+        let mut map = HashMap::new();
+        map.insert("item1", "item");
+
+        let mut s = String::new();
+        assert!(!resource.resource_path_named(&mut s, &map));
+
+        let mut s = String::new();
+        map.insert("item2", "item2");
+        assert!(resource.resource_path_named(&mut s, &map));
+        assert_eq!(s, "/user/item/item2/");
+    }
 }

router/src/router.rs

@@ -1,4 +1,4 @@
-use crate::{Resource, ResourceDef, ResourcePath};
+use crate::{IntoPattern, Resource, ResourceDef, ResourcePath};
 
 #[derive(Debug, Copy, Clone, PartialEq)]
 pub struct ResourceId(pub u16);
@@ -70,7 +70,11 @@ pub struct RouterBuilder<T, U = ()> {
 
 impl<T, U> RouterBuilder<T, U> {
     /// Register resource for specified path.
-    pub fn path(&mut self, path: &str, resource: T) -> &mut (ResourceDef, T, Option<U>) {
+    pub fn path<P: IntoPattern>(
+        &mut self,
+        path: P,
+        resource: T,
+    ) -> &mut (ResourceDef, T, Option<U>) {
         self.resources
             .push((ResourceDef::new(path), resource, None));
         self.resources.last_mut().unwrap()

string/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [0.1.4] - 2020-01-14
+
+* Fix `AsRef<str>` impl
+
+## [0.1.3] - 2020-01-13
+
+* Add `PartialEq<T: AsRef<str>>`, `AsRef<[u8]>` impls
+
 ## [0.1.2] - 2019-12-22
 
 * Fix `new()` method

string/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bytestring"
-version = "0.1.2"
+version = "0.1.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "A UTF-8 encoded string with Bytes as a storage"
 keywords = ["actix"]

string/src/lib.rs

@@ -7,7 +7,7 @@ use bytes::Bytes;
 /// A utf-8 encoded string with [`Bytes`] as a storage.
 ///
 /// [`Bytes`]: https://docs.rs/bytes/0.5.3/bytes/struct.Bytes.html
-#[derive(Clone, Eq, PartialEq, Ord, PartialOrd, Default)]
+#[derive(Clone, Eq, Ord, PartialOrd, Default)]
 pub struct ByteString(Bytes);
 
 impl ByteString {
@@ -43,6 +43,24 @@ impl PartialEq<str> for ByteString {
     }
 }
 
+impl<T: AsRef<str>> PartialEq<T> for ByteString {
+    fn eq(&self, other: &T) -> bool {
+        &self[..] == other.as_ref()
+    }
+}
+
+impl AsRef<[u8]> for ByteString {
+    fn as_ref(&self) -> &[u8] {
+        self.0.as_ref()
+    }
+}
+
+impl AsRef<str> for ByteString {
+    fn as_ref(&self) -> &str {
+        &*self
+    }
+}
+
 impl hash::Hash for ByteString {
     fn hash<H: hash::Hasher>(&self, state: &mut H) {
         (**self).hash(state);
@@ -147,6 +165,14 @@ mod test {
     use std::collections::hash_map::DefaultHasher;
     use std::hash::{Hash, Hasher};
 
+    #[test]
+    fn test_partial_eq() {
+        let s: ByteString = ByteString::from_static("test");
+        assert_eq!(s, "test");
+        assert_eq!(s, *"test");
+        assert_eq!(s, "test".to_string());
+    }
+
     #[test]
     fn test_new() {
         let _: ByteString = ByteString::new();
@@ -167,6 +193,8 @@ mod test {
     fn test_from_string() {
         let s: ByteString = "hello".to_string().into();
         assert_eq!(&s, "hello");
+        let t: &str = s.as_ref();
+        assert_eq!(t, "hello");
     }
 
     #[test]