update actix-connect changes

Future's can be pretty big (> 1500 bytes) so this probably worth doing.

I confirmed with memcpy-find that this did infact eliminate two ~1500
byte copies from the actix-web basic example.

actix-connect/CHANGES.md

@@ -1,9 +1,16 @@
 # Changes
 
+## [0.2.1] - 2019-07-17
+
+### Added
+
+* Expose Connect addrs #30
+
 ### Changed
 
 * Update `derive_more` to 0.15
 
+
 ## [0.2.0] - 2019-05-12
 
 ### Changed

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "0.2.0"
+version = "0.2.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix Connector - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]

actix-connect/src/connect.rs

@@ -1,5 +1,6 @@
-use std::collections::VecDeque;
+use std::collections::{vec_deque, VecDeque};
 use std::fmt;
+use std::iter::{FromIterator, FusedIterator};
 use std::net::SocketAddr;
 
 use either::Either;
@@ -77,6 +78,20 @@ impl<T: Address> Connect<T> {
         self
     }
 
+    /// Use addresses.
+    pub fn set_addrs<I>(mut self, addrs: I) -> Self
+    where
+        I: IntoIterator<Item = SocketAddr>,
+    {
+        let mut addrs = VecDeque::from_iter(addrs);
+        self.addr = if addrs.len() < 2 {
+            addrs.pop_front().map(Either::Left)
+        } else {
+            Some(Either::Right(addrs))
+        };
+        self
+    }
+
     /// Host name
     pub fn host(&self) -> &str {
         self.req.host()
@@ -86,6 +101,28 @@ impl<T: Address> Connect<T> {
     pub fn port(&self) -> u16 {
         self.req.port().unwrap_or(self.port)
     }
+
+    /// Preresolved addresses of the request.
+    pub fn addrs(&self) -> ConnectAddrsIter<'_> {
+        let inner = match self.addr {
+            None => Either::Left(None),
+            Some(Either::Left(addr)) => Either::Left(Some(addr)),
+            Some(Either::Right(ref addrs)) => Either::Right(addrs.iter()),
+        };
+
+        ConnectAddrsIter { inner }
+    }
+
+    /// Takes preresolved addresses of the request.
+    pub fn take_addrs(&mut self) -> ConnectTakeAddrsIter {
+        let inner = match self.addr.take() {
+            None => Either::Left(None),
+            Some(Either::Left(addr)) => Either::Left(Some(addr)),
+            Some(Either::Right(addrs)) => Either::Right(addrs.into_iter()),
+        };
+
+        ConnectTakeAddrsIter { inner }
+    }
 }
 
 impl<T: Address> From<T> for Connect<T> {
@@ -100,6 +137,70 @@ impl<T: Address> fmt::Display for Connect<T> {
     }
 }
 
+/// Iterator over addresses in a [`Connect`](struct.Connect.html) request.
+#[derive(Clone)]
+pub struct ConnectAddrsIter<'a> {
+    inner: Either<Option<SocketAddr>, vec_deque::Iter<'a, SocketAddr>>,
+}
+
+impl Iterator for ConnectAddrsIter<'_> {
+    type Item = SocketAddr;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner {
+            Either::Left(ref mut opt) => opt.take(),
+            Either::Right(ref mut iter) => iter.next().copied(),
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        match self.inner {
+            Either::Left(Some(_)) => (1, Some(1)),
+            Either::Left(None) => (0, Some(0)),
+            Either::Right(ref iter) => iter.size_hint(),
+        }
+    }
+}
+
+impl fmt::Debug for ConnectAddrsIter<'_> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_list().entries(self.clone()).finish()
+    }
+}
+
+impl ExactSizeIterator for ConnectAddrsIter<'_> {}
+
+impl FusedIterator for ConnectAddrsIter<'_> {}
+
+/// Owned iterator over addresses in a [`Connect`](struct.Connect.html) request.
+#[derive(Debug)]
+pub struct ConnectTakeAddrsIter {
+    inner: Either<Option<SocketAddr>, vec_deque::IntoIter<SocketAddr>>,
+}
+
+impl Iterator for ConnectTakeAddrsIter {
+    type Item = SocketAddr;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner {
+            Either::Left(ref mut opt) => opt.take(),
+            Either::Right(ref mut iter) => iter.next(),
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        match self.inner {
+            Either::Left(Some(_)) => (1, Some(1)),
+            Either::Left(None) => (0, Some(0)),
+            Either::Right(ref iter) => iter.size_hint(),
+        }
+    }
+}
+
+impl ExactSizeIterator for ConnectTakeAddrsIter {}
+
+impl FusedIterator for ConnectTakeAddrsIter {}
+
 fn parse(host: &str) -> (&str, Option<u16>) {
     let mut parts_iter = host.splitn(2, ':');
     if let Some(host) = parts_iter.next() {

actix-connect/src/resolver.rs

@@ -1,4 +1,3 @@
-use std::collections::VecDeque;
 use std::marker::PhantomData;
 use std::net::SocketAddr;
 
@@ -162,23 +161,20 @@ impl<T: Address> Future for ResolverFuture<T> {
         })? {
             Async::NotReady => Ok(Async::NotReady),
             Async::Ready(ips) => {
-                let mut req = self.req.take().unwrap();
-                let mut addrs: VecDeque<_> = ips
-                    .iter()
-                    .map(|ip| SocketAddr::new(ip, req.port()))
-                    .collect();
+                let req = self.req.take().unwrap();
+
+                let port = req.port();
+                let req = req.set_addrs(ips.iter().map(|ip| SocketAddr::new(ip, port)));
+
                 trace!(
                     "DNS resolver: host {:?} resolved to {:?}",
                     req.host(),
-                    addrs
+                    req.addrs()
                 );
-                if addrs.is_empty() {
+
+                if req.addr.is_none() {
                     Err(ConnectError::NoRecords)
-                } else if addrs.len() == 1 {
-                    req.addr = Some(either::Either::Left(addrs.pop_front().unwrap()));
-                    Ok(Async::Ready(req))
                 } else {
-                    req.addr = Some(either::Either::Right(addrs));
                     Ok(Async::Ready(req))
                 }
             }

actix-ioframe/Cargo.toml

@@ -18,8 +18,8 @@ name = "actix_ioframe"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "0.4.0"
-actix-codec = "0.1.1"
+actix-service = "0.4.1"
+actix-codec = "0.1.2"
 bytes = "0.4"
 either = "1.5.2"
 futures = "0.1.25"
@@ -28,3 +28,8 @@ log = "0.4"
 
 [dev-dependencies]
 actix-rt = "0.2.2"
+actix-connect = "0.2.0"
+actix-test-server = "0.2.2"
+actix-server-config = "0.1.1"
+tokio-tcp = "0.1"
+tokio-timer = "0.2"

actix-ioframe/src/cell.rs

@@ -29,10 +29,6 @@ impl<T> Cell<T> {
         }
     }
 
-    pub fn get_ref(&self) -> &T {
-        unsafe { &*self.inner.as_ref().get() }
-    }
-
     pub fn get_mut(&mut self) -> &mut T {
         unsafe { &mut *self.inner.as_ref().get() }
     }

actix-ioframe/src/connect.rs

@@ -1,69 +1,37 @@
+use std::marker::PhantomData;
+
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use futures::unsync::mpsc;
 
-use crate::cell::Cell;
 use crate::dispatcher::FramedMessage;
 use crate::sink::Sink;
 
 pub struct Connect<Io, St = (), Codec = ()> {
     io: Io,
-    codec: Codec,
-    state: St,
-    // rx: mpsc::UnboundedReceiver<FramedMessage<<Codec as Encoder>::Item>>,
-    // sink: Sink<<Codec as Encoder>::Item>,
+    _t: PhantomData<(St, Codec)>,
 }
 
-impl<Io> Connect<Io> {
+impl<Io> Connect<Io>
+where
+    Io: AsyncRead + AsyncWrite,
+{
     pub(crate) fn new(io: Io) -> Self {
         Self {
             io,
-            codec: (),
-            state: (),
-        }
-    }
-}
-
-impl<Io, S, C> Connect<Io, S, C> {
-    pub fn codec<Codec>(self, codec: Codec) -> Connect<Io, S, Codec> {
-        Connect {
-            codec,
-            io: self.io,
-            state: self.state,
+            _t: PhantomData,
         }
     }
 
-    pub fn state<St>(self, state: St) -> Connect<Io, St, C> {
-        Connect {
-            state,
-            io: self.io,
-            codec: self.codec,
-        }
-    }
-
-    pub fn state_fn<St, F>(self, f: F) -> Connect<Io, St, C>
+    pub fn codec<Codec>(self, codec: Codec) -> ConnectResult<Io, (), Codec>
     where
-        F: FnOnce(&Connect<Io, S, C>) -> St,
+        Codec: Encoder + Decoder,
     {
-        Connect {
-            state: f(&self),
-            io: self.io,
-            codec: self.codec,
-        }
-    }
-}
-
-impl<Io, S, C> Connect<Io, S, C>
-where
-    C: Encoder + Decoder,
-    Io: AsyncRead + AsyncWrite,
-{
-    pub fn into_result(self) -> ConnectResult<Io, S, C> {
         let (tx, rx) = mpsc::unbounded();
         let sink = Sink::new(tx);
 
         ConnectResult {
-            state: Cell::new(self.state),
-            framed: Framed::new(self.io, self.codec),
+            state: (),
+            framed: Framed::new(self.io, codec),
             rx,
             sink,
         }
@@ -71,7 +39,7 @@ where
 }
 
 pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
-    pub(crate) state: Cell<St>,
+    pub(crate) state: St,
     pub(crate) framed: Framed<Io, Codec>,
     pub(crate) rx: mpsc::UnboundedReceiver<FramedMessage<<Codec as Encoder>::Item>>,
     pub(crate) sink: Sink<<Codec as Encoder>::Item>,
@@ -82,6 +50,26 @@ impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
     pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
         &self.sink
     }
+
+    #[inline]
+    pub fn get_ref(&self) -> &Io {
+        self.framed.get_ref()
+    }
+
+    #[inline]
+    pub fn get_mut(&mut self) -> &mut Io {
+        self.framed.get_mut()
+    }
+
+    #[inline]
+    pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec> {
+        ConnectResult {
+            state,
+            framed: self.framed,
+            rx: self.rx,
+            sink: self.sink,
+        }
+    }
 }
 
 impl<Io, St, Codec> futures::Stream for ConnectResult<Io, St, Codec>

actix-ioframe/src/dispatcher.rs

@@ -1,6 +1,7 @@
 //! Framed dispatcher service and related utilities
 use std::collections::VecDeque;
 use std::mem;
+use std::rc::Rc;
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, Service};
@@ -13,6 +14,7 @@ use crate::cell::Cell;
 use crate::error::ServiceError;
 use crate::item::Item;
 use crate::sink::Sink;
+use crate::state::State;
 
 type Request<S, U> = Item<S, U>;
 type Response<U> = <U as Encoder>::Item;
@@ -37,11 +39,12 @@ where
 {
     service: S,
     sink: Sink<<U as Encoder>::Item>,
-    state: Cell<St>,
-    dispatch_state: State<S, U>,
+    state: State<St>,
+    dispatch_state: FramedState<S, U>,
     framed: Framed<T, U>,
     rx: Option<mpsc::UnboundedReceiver<FramedMessage<<U as Encoder>::Item>>>,
     inner: Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
+    disconnect: Option<Rc<Fn(&mut St, bool)>>,
 }
 
 impl<St, S, T, U> FramedDispatcher<St, S, T, U>
@@ -56,18 +59,20 @@ where
 {
     pub(crate) fn new<F: IntoService<S>>(
         framed: Framed<T, U>,
-        state: Cell<St>,
+        state: State<St>,
         service: F,
         rx: mpsc::UnboundedReceiver<FramedMessage<<U as Encoder>::Item>>,
         sink: Sink<<U as Encoder>::Item>,
+        disconnect: Option<Rc<Fn(&mut St, bool)>>,
     ) -> Self {
         FramedDispatcher {
             framed,
             state,
             sink,
+            disconnect,
             rx: Some(rx),
             service: service.into_service(),
-            dispatch_state: State::Processing,
+            dispatch_state: FramedState::Processing,
             inner: Cell::new(FramedDispatcherInner {
                 buf: VecDeque::new(),
                 task: AtomicTask::new(),
@@ -76,7 +81,7 @@ where
     }
 }
 
-enum State<S: Service, U: Encoder + Decoder> {
+enum FramedState<S: Service, U: Encoder + Decoder> {
     Processing,
     Error(ServiceError<S::Error, U>),
     FramedError(ServiceError<S::Error, U>),
@@ -84,22 +89,22 @@ enum State<S: Service, U: Encoder + Decoder> {
     Stopping,
 }
 
-impl<S: Service, U: Encoder + Decoder> State<S, U> {
+impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
     fn stop(&mut self, tx: Option<oneshot::Sender<()>>) {
         match self {
-            State::FlushAndStop(ref mut vec) => {
+            FramedState::FlushAndStop(ref mut vec) => {
                 if let Some(tx) = tx {
                     vec.push(tx)
                 }
             }
-            State::Processing => {
-                *self = State::FlushAndStop(if let Some(tx) = tx {
+            FramedState::Processing => {
+                *self = FramedState::FlushAndStop(if let Some(tx) = tx {
                     vec![tx]
                 } else {
                     Vec::new()
                 })
             }
-            State::Error(_) | State::FramedError(_) | State::Stopping => {
+            FramedState::Error(_) | FramedState::FramedError(_) | FramedState::Stopping => {
                 if let Some(tx) = tx {
                     let _ = tx.send(());
                 }
@@ -123,6 +128,12 @@ where
     <U as Encoder>::Item: 'static,
     <U as Encoder>::Error: std::fmt::Debug,
 {
+    fn disconnect(&mut self, error: bool) {
+        if let Some(ref disconnect) = self.disconnect {
+            (&*disconnect)(&mut *self.state.get_mut(), error);
+        }
+    }
+
     fn poll_read(&mut self) -> bool {
         loop {
             match self.service.poll_ready() {
@@ -131,12 +142,12 @@ where
                         Ok(Async::Ready(Some(el))) => el,
                         Err(err) => {
                             self.dispatch_state =
-                                State::FramedError(ServiceError::Decoder(err));
+                                FramedState::FramedError(ServiceError::Decoder(err));
                             return true;
                         }
                         Ok(Async::NotReady) => return false,
                         Ok(Async::Ready(None)) => {
-                            self.dispatch_state = State::Stopping;
+                            self.dispatch_state = FramedState::Stopping;
                             return true;
                         }
                     };
@@ -161,7 +172,7 @@ where
                 }
                 Ok(Async::NotReady) => return false,
                 Err(err) => {
-                    self.dispatch_state = State::Error(ServiceError::Service(err));
+                    self.dispatch_state = FramedState::Error(ServiceError::Service(err));
                     return true;
                 }
             }
@@ -180,13 +191,14 @@ where
                         Ok(msg) => {
                             if let Err(err) = self.framed.force_send(msg) {
                                 self.dispatch_state =
-                                    State::FramedError(ServiceError::Encoder(err));
+                                    FramedState::FramedError(ServiceError::Encoder(err));
                                 return true;
                             }
                             buf_empty = inner.buf.is_empty();
                         }
                         Err(err) => {
-                            self.dispatch_state = State::Error(ServiceError::Service(err));
+                            self.dispatch_state =
+                                FramedState::Error(ServiceError::Service(err));
                             return true;
                         }
                     }
@@ -197,7 +209,7 @@ where
                         Ok(Async::Ready(Some(FramedMessage::Message(msg)))) => {
                             if let Err(err) = self.framed.force_send(msg) {
                                 self.dispatch_state =
-                                    State::FramedError(ServiceError::Encoder(err));
+                                    FramedState::FramedError(ServiceError::Encoder(err));
                                 return true;
                             }
                         }
@@ -231,7 +243,8 @@ where
                     Ok(Async::NotReady) => break,
                     Err(err) => {
                         debug!("Error sending data: {:?}", err);
-                        self.dispatch_state = State::FramedError(ServiceError::Encoder(err));
+                        self.dispatch_state =
+                            FramedState::FramedError(ServiceError::Encoder(err));
                         return true;
                     }
                     Ok(Async::Ready(_)) => (),
@@ -259,32 +272,33 @@ where
     type Error = ServiceError<S::Error, U>;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        match mem::replace(&mut self.dispatch_state, State::Processing) {
-            State::Processing => {
+        match mem::replace(&mut self.dispatch_state, FramedState::Processing) {
+            FramedState::Processing => {
                 if self.poll_read() || self.poll_write() {
                     self.poll()
                 } else {
                     Ok(Async::NotReady)
                 }
             }
-            State::Error(err) => {
+            FramedState::Error(err) => {
                 if self.framed.is_write_buf_empty()
                     || (self.poll_write() || self.framed.is_write_buf_empty())
                 {
+                    self.disconnect(true);
                     Err(err)
                 } else {
-                    self.dispatch_state = State::Error(err);
+                    self.dispatch_state = FramedState::Error(err);
                     Ok(Async::NotReady)
                 }
             }
-            State::FlushAndStop(mut vec) => {
+            FramedState::FlushAndStop(mut vec) => {
                 if !self.framed.is_write_buf_empty() {
                     match self.framed.poll_complete() {
                         Err(err) => {
                             debug!("Error sending data: {:?}", err);
                         }
                         Ok(Async::NotReady) => {
-                            self.dispatch_state = State::FlushAndStop(vec);
+                            self.dispatch_state = FramedState::FlushAndStop(vec);
                             return Ok(Async::NotReady);
                         }
                         Ok(Async::Ready(_)) => (),
@@ -293,10 +307,17 @@ where
                 for tx in vec.drain(..) {
                     let _ = tx.send(());
                 }
+                self.disconnect(false);
+                Ok(Async::Ready(()))
+            }
+            FramedState::FramedError(err) => {
+                self.disconnect(true);
+                Err(err)
+            }
+            FramedState::Stopping => {
+                self.disconnect(false);
                 Ok(Async::Ready(()))
             }
-            State::FramedError(err) => Err(err),
-            State::Stopping => Ok(Async::Ready(())),
         }
     }
 }

actix-ioframe/src/item.rs

@@ -1,13 +1,14 @@
+use std::cell::{Ref, RefMut};
 use std::fmt;
 use std::ops::{Deref, DerefMut};
 
 use actix_codec::{Decoder, Encoder};
 
-use crate::cell::Cell;
 use crate::sink::Sink;
+use crate::state::State;
 
-pub struct Item<S, Codec: Encoder + Decoder> {
-    state: Cell<S>,
+pub struct Item<St, Codec: Encoder + Decoder> {
+    state: State<St>,
     sink: Sink<<Codec as Encoder>::Item>,
     item: <Codec as Decoder>::Item,
 }
@@ -17,7 +18,7 @@ where
     Codec: Encoder + Decoder,
 {
     pub(crate) fn new(
-        state: Cell<St>,
+        state: State<St>,
         sink: Sink<<Codec as Encoder>::Item>,
         item: <Codec as Decoder>::Item,
     ) -> Self {
@@ -25,12 +26,12 @@ where
     }
 
     #[inline]
-    pub fn state(&self) -> &St {
+    pub fn state(&self) -> Ref<St> {
         self.state.get_ref()
     }
 
     #[inline]
-    pub fn state_mut(&mut self) -> &mut St {
+    pub fn state_mut(&mut self) -> RefMut<St> {
         self.state.get_mut()
     }
 
@@ -43,6 +44,17 @@ where
     pub fn into_inner(self) -> <Codec as Decoder>::Item {
         self.item
     }
+
+    #[inline]
+    pub fn into_parts(
+        self,
+    ) -> (
+        State<St>,
+        Sink<<Codec as Encoder>::Item>,
+        <Codec as Decoder>::Item,
+    ) {
+        (self.state, self.sink, self.item)
+    }
 }
 
 impl<St, Codec> Deref for Item<St, Codec>

actix-ioframe/src/lib.rs

@@ -5,9 +5,11 @@ mod error;
 mod item;
 mod service;
 mod sink;
+mod state;
 
 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::state::State;

actix-ioframe/src/service.rs

@@ -9,6 +9,7 @@ use crate::connect::{Connect, ConnectResult};
 use crate::dispatcher::FramedDispatcher;
 use crate::error::ServiceError;
 use crate::item::Item;
+use crate::state::State;
 
 type RequestItem<S, U> = Item<S, U>;
 type ResponseItem<U> = Option<<U as Encoder>::Item>;
@@ -32,6 +33,7 @@ impl<St, Codec> Builder<St, Codec> {
     {
         ServiceBuilder {
             connect: connect.into_service(),
+            disconnect: None,
             _t: PhantomData,
         }
     }
@@ -52,6 +54,7 @@ impl<St, Codec> Builder<St, Codec> {
     {
         NewServiceBuilder {
             connect: connect.into_new_service(),
+            disconnect: None,
             _t: PhantomData,
         }
     }
@@ -59,6 +62,7 @@ impl<St, Codec> Builder<St, Codec> {
 
 pub struct ServiceBuilder<St, C, Io, Codec> {
     connect: C,
+    disconnect: Option<Rc<Fn(&mut St, bool)>>,
     _t: PhantomData<(St, Io, Codec)>,
 }
 
@@ -72,6 +76,18 @@ where
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
 {
+    /// Callback to execute on disconnect
+    ///
+    /// Second parameter indicates error occured during disconnect.
+    pub fn disconnect<F, Out>(mut self, disconnect: F) -> Self
+    where
+        F: Fn(&mut St, bool) + 'static,
+    {
+        self.disconnect = Some(Rc::new(disconnect));
+        self
+    }
+
+    /// Provide stream items handler service and construct service factory.
     pub fn finish<F, T>(
         self,
         service: F,
@@ -89,6 +105,7 @@ where
         FramedServiceImpl {
             connect: self.connect,
             handler: Rc::new(service.into_new_service()),
+            disconnect: self.disconnect.clone(),
             _t: PhantomData,
         }
     }
@@ -96,6 +113,7 @@ where
 
 pub struct NewServiceBuilder<St, C, Io, Codec> {
     connect: C,
+    disconnect: Option<Rc<Fn(&mut St, bool)>>,
     _t: PhantomData<(St, Io, Codec)>,
 }
 
@@ -110,11 +128,22 @@ where
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
 {
-    pub fn finish<F, T>(
+    /// Callback to execute on disconnect
+    ///
+    /// Second parameter indicates error occured during disconnect.
+    pub fn disconnect<F>(mut self, disconnect: F) -> Self
+    where
+        F: Fn(&mut St, bool) + 'static,
+    {
+        self.disconnect = Some(Rc::new(disconnect));
+        self
+    }
+
+    pub fn finish<F, T, Cfg>(
         self,
         service: F,
     ) -> impl NewService<
-        Config = (),
+        Config = Cfg,
         Request = Io,
         Response = (),
         Error = ServiceError<C::Error, Codec>,
@@ -132,18 +161,20 @@ where
         FramedService {
             connect: self.connect,
             handler: Rc::new(service.into_new_service()),
+            disconnect: self.disconnect,
             _t: PhantomData,
         }
     }
 }
 
-pub(crate) struct FramedService<St, C, T, Io, Codec> {
+pub(crate) struct FramedService<St, C, T, Io, Codec, Cfg> {
     connect: C,
     handler: Rc<T>,
-    _t: PhantomData<(St, Io, Codec)>,
+    disconnect: Option<Rc<Fn(&mut St, bool)>>,
+    _t: PhantomData<(St, Io, Codec, Cfg)>,
 }
 
-impl<St, C, T, Io, Codec> NewService for FramedService<St, C, T, Io, Codec>
+impl<St, C, T, Io, Codec, Cfg> NewService for FramedService<St, C, T, Io, Codec, Cfg>
 where
     St: 'static,
     Io: AsyncRead + AsyncWrite,
@@ -161,7 +192,7 @@ where
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
 {
-    type Config = ();
+    type Config = Cfg;
     type Request = Io;
     type Response = ();
     type Error = ServiceError<C::Error, Codec>;
@@ -169,8 +200,9 @@ where
     type Service = FramedServiceImpl<St, C::Service, T, Io, Codec>;
     type Future = Box<Future<Item = Self::Service, Error = Self::InitError>>;
 
-    fn new_service(&self, _: &()) -> Self::Future {
+    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
         Box::new(
@@ -179,6 +211,7 @@ where
                 .map(move |connect| FramedServiceImpl {
                     connect,
                     handler,
+                    disconnect,
                     _t: PhantomData,
                 }),
         )
@@ -188,12 +221,12 @@ where
 pub struct FramedServiceImpl<St, C, T, Io, Codec> {
     connect: C,
     handler: Rc<T>,
+    disconnect: Option<Rc<Fn(&mut St, bool)>>,
     _t: PhantomData<(St, Io, Codec)>,
 }
 
 impl<St, C, T, Io, Codec> Service for FramedServiceImpl<St, C, T, Io, Codec>
 where
-    // St: 'static,
     Io: AsyncRead + AsyncWrite,
     C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
     C::Error: 'static,
@@ -224,6 +257,7 @@ where
                 self.connect.call(Connect::new(req)),
                 self.handler.clone(),
             ),
+            disconnect: self.disconnect.clone(),
         }
     }
 }
@@ -246,6 +280,7 @@ where
     <Codec as Encoder>::Error: std::fmt::Debug,
 {
     inner: FramedServiceImplResponseInner<St, Io, Codec, C, T>,
+    disconnect: Option<Rc<Fn(&mut St, bool)>>,
 }
 
 enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
@@ -296,7 +331,7 @@ where
                 match fut.poll()? {
                     Async::Ready(res) => {
                         self.inner = FramedServiceImplResponseInner::Handler(
-                            handler.new_service(res.state.get_ref()),
+                            handler.new_service(&res.state),
                             Some(res),
                         );
                         self.poll()
@@ -310,7 +345,12 @@ where
                         let res = res.take().unwrap();
                         self.inner =
                             FramedServiceImplResponseInner::Dispatcher(FramedDispatcher::new(
-                                res.framed, res.state, handler, res.rx, res.sink,
+                                res.framed,
+                                State::new(res.state),
+                                handler,
+                                res.rx,
+                                res.sink,
+                                self.disconnect.clone(),
                             ));
                         self.poll()
                     }

actix-ioframe/src/state.rs

@@ -0,0 +1,30 @@
+use std::cell::{Ref, RefCell, RefMut};
+use std::rc::Rc;
+
+/// Connection state
+///
+/// Connection state is an arbitrary data attached to the each incoming message.
+#[derive(Debug)]
+pub struct State<T>(Rc<RefCell<T>>);
+
+impl<T> State<T> {
+    pub(crate) fn new(st: T) -> Self {
+        State(Rc::new(RefCell::new(st)))
+    }
+
+    #[inline]
+    pub fn get_ref(&self) -> Ref<T> {
+        self.0.borrow()
+    }
+
+    #[inline]
+    pub fn get_mut(&mut self) -> RefMut<T> {
+        self.0.borrow_mut()
+    }
+}
+
+impl<T> Clone for State<T> {
+    fn clone(&self) -> Self {
+        State(self.0.clone())
+    }
+}

actix-ioframe/tests/test_server.rs

@@ -0,0 +1,60 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use actix_codec::BytesCodec;
+use actix_server_config::Io;
+use actix_service::{new_apply_fn, Service};
+use actix_test_server::TestServer;
+use futures::Future;
+use tokio_tcp::TcpStream;
+use tokio_timer::sleep;
+
+use actix_ioframe::{Builder, Connect};
+
+struct State;
+
+#[test]
+fn test_disconnect() -> std::io::Result<()> {
+    let disconnect = Arc::new(AtomicBool::new(false));
+    let disconnect1 = disconnect.clone();
+
+    let mut srv = TestServer::with(move || {
+        let disconnect1 = disconnect1.clone();
+
+        new_apply_fn(
+            Builder::new()
+                .factory(|conn: Connect<_>| Ok(conn.codec(BytesCodec).state(State)))
+                .disconnect(move |_, _| {
+                    disconnect1.store(true, Ordering::Relaxed);
+                })
+                .finish(|_t| Ok(None)),
+            |io: Io<TcpStream>, srv| srv.call(io.into_parts().0),
+        )
+    });
+
+    let mut client = Builder::new()
+        .service(|conn: Connect<_>| {
+            let conn = conn.codec(BytesCodec).state(State);
+            conn.sink().close();
+            Ok(conn)
+        })
+        .finish(|_t| Ok(None));
+
+    let conn = srv
+        .block_on(
+            actix_connect::default_connector()
+                .call(actix_connect::Connect::with(String::new(), srv.addr())),
+        )
+        .unwrap();
+
+    srv.block_on(client.call(conn.into_parts().0)).unwrap();
+    let _ = srv.block_on(
+        sleep(Duration::from_millis(100))
+            .map(|_| ())
+            .map_err(|_| ()),
+    );
+    assert!(disconnect.load(Ordering::Relaxed));
+
+    Ok(())
+}

actix-rt/CHANGES.md

@@ -1,5 +1,12 @@
 # Changes
 
+## [0.2.4] - 2019-07-17
+
+### Changed
+
+* Avoid a copy of the Future when initializing the Box. #29
+
+
 ## [0.2.3] - 2019-06-22
 
 ### Added

actix-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-rt"
-version = "0.2.3"
+version = "0.2.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -24,3 +24,4 @@ tokio-current-thread = "0.1"
 tokio-executor = "0.1.5"
 tokio-reactor = "0.1.7"
 tokio-timer = "0.2.8"
+copyless = "0.1.4"

actix-rt/src/arbiter.rs

@@ -11,6 +11,8 @@ use tokio_current_thread::spawn;
 use crate::builder::Builder;
 use crate::system::System;
 
+use copyless::BoxHelper;
+
 thread_local!(
     static ADDR: RefCell<Option<Arbiter>> = RefCell::new(None);
     static RUNNING: Cell<bool> = Cell::new(false);
@@ -141,9 +143,9 @@ impl Arbiter {
     {
         RUNNING.with(move |cell| {
             if cell.get() {
-                spawn(Box::new(future));
+                spawn(Box::alloc().init(future));
             } else {
-                Q.with(move |cell| cell.borrow_mut().push(Box::new(future)));
+                Q.with(move |cell| cell.borrow_mut().push(Box::alloc().init(future)));
             }
         });
     }

actix-service/src/transform.rs

@@ -7,16 +7,10 @@ use crate::transform_err::{TransformFromErr, TransformMapInitErr};
 use crate::{IntoNewService, NewService, Service};
 
 /// The `Transform` trait defines the interface of a Service factory. `Transform`
-/// is often implemented for middleware, defining how to manufacture a
-/// middleware Service.  A Service that is manufactured by the factory takes
+/// 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.  A Service can be a variety of types, such
-/// as (but not limited to) another middleware Service, an extractor Service,
-/// other helper Services, or the request handler endpoint Service.
-///
-/// A Service is created by the factory during server initialization.
-///
-/// `Config` is a service factory configuration type.
+/// ownership of the next Service.
 pub trait Transform<S> {
     /// Requests handled by the service.
     type Request;