move helper services to separate package

actix-utils/CHANGES.md

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

actix-utils/Cargo.toml

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

actix-utils/src/cell.rs

@@ -0,0 +1,63 @@
+//! Custom cell impl
+
+#[cfg(feature = "cell")]
+use std::cell::UnsafeCell;
+#[cfg(not(feature = "cell"))]
+use std::cell::{Ref, RefCell, RefMut};
+use std::fmt;
+use std::rc::Rc;
+
+pub(crate) struct Cell<T> {
+    #[cfg(feature = "cell")]
+    inner: Rc<UnsafeCell<T>>,
+    #[cfg(not(feature = "cell"))]
+    inner: Rc<RefCell<T>>,
+}
+
+impl<T> Clone for Cell<T> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for Cell<T> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        self.inner.fmt(f)
+    }
+}
+
+#[cfg(feature = "cell")]
+impl<T> Cell<T> {
+    pub fn new(inner: T) -> Self {
+        Self {
+            inner: Rc::new(UnsafeCell::new(inner)),
+        }
+    }
+
+    pub fn borrow(&self) -> &T {
+        unsafe { &*self.inner.as_ref().get() }
+    }
+
+    pub fn borrow_mut(&self) -> &mut T {
+        unsafe { &mut *self.inner.as_ref().get() }
+    }
+}
+
+#[cfg(not(feature = "cell"))]
+impl<T> Cell<T> {
+    pub fn new(inner: T) -> Self {
+        Self {
+            inner: Rc::new(RefCell::new(inner)),
+        }
+    }
+
+    pub fn borrow(&self) -> Ref<T> {
+        self.inner.borrow()
+    }
+
+    pub fn borrow_mut(&self) -> RefMut<T> {
+        self.inner.borrow_mut()
+    }
+}

actix-utils/src/cloneable.rs

@@ -0,0 +1,51 @@
+use std::marker::PhantomData;
+use std::rc::Rc;
+
+use actix_service::Service;
+use futures::Poll;
+
+use super::cell::Cell;
+
+/// Service that allows to turn non-clone service to a service with `Clone` impl
+pub struct CloneableService<T: 'static> {
+    service: Cell<T>,
+    _t: PhantomData<Rc<()>>,
+}
+
+impl<T: 'static> CloneableService<T> {
+    pub fn new<Request>(service: T) -> Self
+    where
+        T: Service<Request>,
+    {
+        Self {
+            service: Cell::new(service),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T: 'static> Clone for CloneableService<T> {
+    fn clone(&self) -> Self {
+        Self {
+            service: self.service.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T: 'static, Request> Service<Request> for CloneableService<T>
+where
+    T: Service<Request>,
+{
+    type Response = T::Response;
+    type Error = T::Error;
+    type Future = T::Future;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        self.service.borrow_mut().poll_ready()
+    }
+
+    fn call(&mut self, req: Request) -> Self::Future {
+        self.service.borrow_mut().call(req)
+    }
+}

actix-utils/src/counter.rs

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

actix-utils/src/either.rs

@@ -0,0 +1,90 @@
+//! Contains `Either` service and related types and functions.
+use actix_service::{NewService, Service};
+use futures::{future, try_ready, Async, Future, Poll};
+
+/// Combine two different service types into a single type.
+///
+/// Both services must be of the same request, response, and error types.
+/// `EitherService` is useful for handling conditional branching in service
+/// middleware to different inner service types.
+pub enum EitherService<A, B> {
+    A(A),
+    B(B),
+}
+
+impl<A, B, Request> Service<Request> for EitherService<A, B>
+where
+    A: Service<Request>,
+    B: Service<Request, Response = A::Response, Error = A::Error>,
+{
+    type Response = A::Response;
+    type Error = A::Error;
+    type Future = future::Either<A::Future, B::Future>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        match self {
+            EitherService::A(ref mut inner) => inner.poll_ready(),
+            EitherService::B(ref mut inner) => inner.poll_ready(),
+        }
+    }
+
+    fn call(&mut self, req: Request) -> Self::Future {
+        match self {
+            EitherService::A(ref mut inner) => future::Either::A(inner.call(req)),
+            EitherService::B(ref mut inner) => future::Either::B(inner.call(req)),
+        }
+    }
+}
+
+/// Combine two different new service types into a single type.
+pub enum Either<A, B> {
+    A(A),
+    B(B),
+}
+
+impl<A, B, Request> NewService<Request> for Either<A, B>
+where
+    A: NewService<Request>,
+    B: NewService<Request, Response = A::Response, Error = A::Error, InitError = A::InitError>,
+{
+    type Response = A::Response;
+    type Error = A::Error;
+    type InitError = A::InitError;
+    type Service = EitherService<A::Service, B::Service>;
+    type Future = EitherNewService<A, B, Request>;
+
+    fn new_service(&self) -> Self::Future {
+        match self {
+            Either::A(ref inner) => EitherNewService::A(inner.new_service()),
+            Either::B(ref inner) => EitherNewService::B(inner.new_service()),
+        }
+    }
+}
+
+#[doc(hidden)]
+pub enum EitherNewService<A: NewService<R>, B: NewService<R>, R> {
+    A(A::Future),
+    B(B::Future),
+}
+
+impl<A, B, Request> Future for EitherNewService<A, B, Request>
+where
+    A: NewService<Request>,
+    B: NewService<Request, Response = A::Response, Error = A::Error, InitError = A::InitError>,
+{
+    type Item = EitherService<A::Service, B::Service>;
+    type Error = A::InitError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self {
+            EitherNewService::A(ref mut fut) => {
+                let service = try_ready!(fut.poll());
+                Ok(Async::Ready(EitherService::A(service)))
+            }
+            EitherNewService::B(ref mut fut) => {
+                let service = try_ready!(fut.poll());
+                Ok(Async::Ready(EitherService::B(service)))
+            }
+        }
+    }
+}

actix-utils/src/framed.rs

@@ -0,0 +1,478 @@
+//! Framed dispatcher service and related utilities
+use std::marker::PhantomData;
+use std::mem;
+
+use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
+use actix_rt::Arbiter;
+use actix_service::{IntoNewService, IntoService, NewService, Service};
+use futures::future::{ok, FutureResult};
+use futures::unsync::mpsc;
+use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
+
+type Request<U> = <U as Decoder>::Item;
+type Response<U> = <U as Encoder>::Item;
+
+pub struct FramedNewService<S, T, U> {
+    factory: S,
+    _t: PhantomData<(T, U)>,
+}
+
+impl<S, T, U> FramedNewService<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: NewService<Request<U>, Response = Response<U>>,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Future: 'static,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Error: 'static,
+    <U as Encoder>::Item: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    pub fn new<F1: IntoNewService<S, Request<U>>>(factory: F1) -> Self {
+        Self {
+            factory: factory.into_new_service(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<S, T, U> Clone for FramedNewService<S, T, U>
+where
+    S: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            factory: self.factory.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<S, T, U> NewService<Framed<T, U>> for FramedNewService<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: NewService<Request<U>, Response = Response<U>> + Clone,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Future: 'static,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Error: 'static,
+    <U as Encoder>::Item: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    type Response = FramedTransport<S::Service, T, U>;
+    type Error = S::InitError;
+    type InitError = S::InitError;
+    type Service = FramedService<S, T, U>;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        ok(FramedService {
+            factory: self.factory.clone(),
+            _t: PhantomData,
+        })
+    }
+}
+
+pub struct FramedService<S, T, U> {
+    factory: S,
+    _t: PhantomData<(T, U)>,
+}
+
+impl<S, T, U> Clone for FramedService<S, T, U>
+where
+    S: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            factory: self.factory.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<S, T, U> Service<Framed<T, U>> for FramedService<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: NewService<Request<U>, Response = Response<U>>,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Future: 'static,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Error: 'static,
+    <U as Encoder>::Item: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    type Response = FramedTransport<S::Service, T, U>;
+    type Error = S::InitError;
+    type Future = FramedServiceResponseFuture<S, T, U>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: Framed<T, U>) -> Self::Future {
+        FramedServiceResponseFuture {
+            fut: self.factory.new_service(),
+
+            framed: Some(req),
+        }
+    }
+}
+
+#[doc(hidden)]
+pub struct FramedServiceResponseFuture<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: NewService<Request<U>, Response = Response<U>>,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Future: 'static,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Error: 'static,
+    <U as Encoder>::Item: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    fut: S::Future,
+    framed: Option<Framed<T, U>>,
+}
+
+impl<S, T, U> Future for FramedServiceResponseFuture<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: NewService<Request<U>, Response = Response<U>>,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Future: 'static,
+    <<S as NewService<Request<U>>>::Service as Service<Request<U>>>::Error: 'static,
+    <U as Encoder>::Item: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    type Item = FramedTransport<S::Service, T, U>;
+    type Error = S::InitError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self.fut.poll()? {
+            Async::NotReady => Ok(Async::NotReady),
+            Async::Ready(service) => Ok(Async::Ready(FramedTransport::new(
+                self.framed.take().unwrap(),
+                service,
+            ))),
+        }
+    }
+}
+
+/// Framed transport errors
+pub enum FramedTransportError<E, U: Encoder + Decoder> {
+    Service(E),
+    Encoder(<U as Encoder>::Error),
+    Decoder(<U as Decoder>::Error),
+}
+
+impl<E, U: Encoder + Decoder> From<E> for FramedTransportError<E, U> {
+    fn from(err: E) -> Self {
+        FramedTransportError::Service(err)
+    }
+}
+
+/// FramedTransport - is a future that reads frames from Framed object
+/// and pass then to the service.
+pub struct FramedTransport<S, T, U>
+where
+    S: Service<Request<U>, Response = Response<U>>,
+    T: AsyncRead + AsyncWrite,
+    U: Encoder + Decoder,
+{
+    service: S,
+    state: TransportState<S, U>,
+    framed: Framed<T, U>,
+    request: Option<Request<U>>,
+    response: Option<Response<U>>,
+    write_rx: mpsc::Receiver<Result<Response<U>, S::Error>>,
+    write_tx: mpsc::Sender<Result<Response<U>, S::Error>>,
+    flushed: bool,
+}
+
+enum TransportState<S: Service<Request<U>>, U: Encoder + Decoder> {
+    Processing,
+    Error(FramedTransportError<S::Error, U>),
+    EncoderError(FramedTransportError<S::Error, U>),
+    Stopping,
+}
+
+impl<S, T, U> FramedTransport<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: Service<Request<U>, Response = Response<U>>,
+    S::Future: 'static,
+    S::Error: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    pub fn new<F: IntoService<S, Request<U>>>(framed: Framed<T, U>, service: F) -> Self {
+        let (write_tx, write_rx) = mpsc::channel(16);
+        FramedTransport {
+            framed,
+            write_rx,
+            write_tx,
+            service: service.into_service(),
+            state: TransportState::Processing,
+            request: None,
+            response: None,
+            flushed: true,
+        }
+    }
+
+    /// Get reference to a service wrapped by `FramedTransport` instance.
+    pub fn get_ref(&self) -> &S {
+        &self.service
+    }
+
+    /// Get mutable reference to a service wrapped by `FramedTransport`
+    /// instance.
+    pub fn get_mut(&mut self) -> &mut S {
+        &mut self.service
+    }
+
+    /// Get reference to a framed instance wrapped by `FramedTransport`
+    /// instance.
+    pub fn get_framed(&self) -> &Framed<T, U> {
+        &self.framed
+    }
+
+    /// Get mutable reference to a framed instance wrapped by `FramedTransport`
+    /// instance.
+    pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
+        &mut self.framed
+    }
+}
+
+impl<S, T, U> FramedTransport<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: Service<Request<U>, Response = Response<U>>,
+    S::Future: 'static,
+    S::Error: 'static,
+    <U as Encoder>::Item: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    fn poll_service(&mut self) -> bool {
+        match self.service.poll_ready() {
+            Ok(Async::Ready(_)) => {
+                if let Some(item) = self.request.take() {
+                    let sender = self.write_tx.clone();
+                    Arbiter::spawn(
+                        self.service
+                            .call(item)
+                            .then(|item| sender.send(item).map(|_| ()).map_err(|_| ())),
+                    );
+                }
+
+                loop {
+                    let item = match self.framed.poll() {
+                        Ok(Async::Ready(Some(el))) => el,
+                        Err(err) => {
+                            self.state =
+                                TransportState::Error(FramedTransportError::Decoder(err));
+                            return true;
+                        }
+                        Ok(Async::NotReady) => return false,
+                        Ok(Async::Ready(None)) => {
+                            self.state = TransportState::Stopping;
+                            return true;
+                        }
+                    };
+
+                    match self.service.poll_ready() {
+                        Ok(Async::Ready(_)) => {
+                            let sender = self.write_tx.clone();
+                            Arbiter::spawn(
+                                self.service
+                                    .call(item)
+                                    .then(|item| sender.send(item).map(|_| ()).map_err(|_| ())),
+                            );
+                        }
+                        Ok(Async::NotReady) => {
+                            self.request = Some(item);
+                            return false;
+                        }
+                        Err(err) => {
+                            self.state =
+                                TransportState::Error(FramedTransportError::Service(err));
+                            return true;
+                        }
+                    }
+                }
+            }
+            Ok(Async::NotReady) => false,
+            Err(err) => {
+                self.state = TransportState::Error(FramedTransportError::Service(err));
+                true
+            }
+        }
+    }
+
+    /// write to sink
+    fn poll_response(&mut self) -> bool {
+        let mut item = self.response.take();
+        loop {
+            item = if let Some(msg) = item {
+                self.flushed = false;
+                match self.framed.start_send(msg) {
+                    Ok(AsyncSink::Ready) => None,
+                    Ok(AsyncSink::NotReady(item)) => Some(item),
+                    Err(err) => {
+                        self.state =
+                            TransportState::EncoderError(FramedTransportError::Encoder(err));
+                        return true;
+                    }
+                }
+            } else {
+                None
+            };
+
+            // flush sink
+            if !self.flushed {
+                match self.framed.poll_complete() {
+                    Ok(Async::Ready(_)) => {
+                        self.flushed = true;
+                    }
+                    Ok(Async::NotReady) => break,
+                    Err(err) => {
+                        self.state =
+                            TransportState::EncoderError(FramedTransportError::Encoder(err));
+                        return true;
+                    }
+                }
+            }
+
+            // check channel
+            if self.flushed {
+                if item.is_none() {
+                    match self.write_rx.poll() {
+                        Ok(Async::Ready(Some(msg))) => match msg {
+                            Ok(msg) => item = Some(msg),
+                            Err(err) => {
+                                self.state =
+                                    TransportState::Error(FramedTransportError::Service(err));
+                                return true;
+                            }
+                        },
+                        Ok(Async::NotReady) => break,
+                        Err(_) => panic!("Bug in gw code"),
+                        Ok(Async::Ready(None)) => panic!("Bug in gw code"),
+                    }
+                } else {
+                    continue;
+                }
+            } else {
+                self.response = item;
+                break;
+            }
+        }
+
+        false
+    }
+}
+
+impl<S, T, U> Future for FramedTransport<S, T, U>
+where
+    T: AsyncRead + AsyncWrite,
+    U: Decoder + Encoder,
+    S: Service<Request<U>, Response = Response<U>>,
+    S::Future: 'static,
+    S::Error: 'static,
+    <U as Encoder>::Item: 'static,
+    <U as Encoder>::Error: 'static,
+{
+    type Item = ();
+    type Error = FramedTransportError<S::Error, U>;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match mem::replace(&mut self.state, TransportState::Processing) {
+            TransportState::Processing => {
+                if self.poll_service() || self.poll_response() {
+                    self.poll()
+                } else {
+                    Ok(Async::NotReady)
+                }
+            }
+            TransportState::Error(err) => {
+                if self.poll_response() || self.flushed {
+                    Err(err)
+                } else {
+                    self.state = TransportState::Error(err);
+                    Ok(Async::NotReady)
+                }
+            }
+            TransportState::EncoderError(err) => Err(err),
+            TransportState::Stopping => Ok(Async::Ready(())),
+        }
+    }
+}
+
+pub struct IntoFramed<T, U, F>
+where
+    T: AsyncRead + AsyncWrite,
+    F: Fn() -> U + Send + Clone + 'static,
+    U: Encoder + Decoder,
+{
+    factory: F,
+    _t: PhantomData<(T,)>,
+}
+
+impl<T, U, F> IntoFramed<T, U, F>
+where
+    T: AsyncRead + AsyncWrite,
+    F: Fn() -> U + Send + Clone + 'static,
+    U: Encoder + Decoder,
+{
+    pub fn new(factory: F) -> Self {
+        IntoFramed {
+            factory,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<T, U, F> NewService<T> for IntoFramed<T, U, F>
+where
+    T: AsyncRead + AsyncWrite,
+    F: Fn() -> U + Send + Clone + 'static,
+    U: Encoder + Decoder,
+{
+    type Response = Framed<T, U>;
+    type Error = ();
+    type InitError = ();
+    type Service = IntoFramedService<T, U, F>;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        ok(IntoFramedService {
+            factory: self.factory.clone(),
+            _t: PhantomData,
+        })
+    }
+}
+
+pub struct IntoFramedService<T, U, F>
+where
+    T: AsyncRead + AsyncWrite,
+    F: Fn() -> U + Send + Clone + 'static,
+    U: Encoder + Decoder,
+{
+    factory: F,
+    _t: PhantomData<(T,)>,
+}
+
+impl<T, U, F> Service<T> for IntoFramedService<T, U, F>
+where
+    T: AsyncRead + AsyncWrite,
+    F: Fn() -> U + Send + Clone + 'static,
+    U: Encoder + Decoder,
+{
+    type Response = Framed<T, U>;
+    type Error = ();
+    type Future = FutureResult<Self::Response, Self::Error>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: T) -> Self::Future {
+        ok(Framed::new(req, (self.factory)()))
+    }
+}

actix-utils/src/inflight.rs

@@ -0,0 +1,137 @@
+use actix_service::{IntoNewService, IntoService, NewService, Service};
+use futures::{try_ready, Async, Future, Poll};
+
+use super::counter::{Counter, CounterGuard};
+
+/// InFlight - new service for service that can limit number of in-flight
+/// async requests.
+///
+/// Default number of in-flight requests is 15
+pub struct InFlight<T> {
+    factory: T,
+    max_inflight: usize,
+}
+
+impl<T> InFlight<T> {
+    pub fn new<F, Request>(factory: F) -> Self
+    where
+        T: NewService<Request>,
+        F: IntoNewService<T, Request>,
+    {
+        Self {
+            factory: factory.into_new_service(),
+            max_inflight: 15,
+        }
+    }
+
+    /// Set max number of in-flight requests.
+    ///
+    /// By default max in-flight requests is 15.
+    pub fn max_inflight(mut self, max: usize) -> Self {
+        self.max_inflight = max;
+        self
+    }
+}
+
+impl<T, Request> NewService<Request> for InFlight<T>
+where
+    T: NewService<Request>,
+{
+    type Response = T::Response;
+    type Error = T::Error;
+    type InitError = T::InitError;
+    type Service = InFlightService<T::Service>;
+    type Future = InFlightResponseFuture<T, Request>;
+
+    fn new_service(&self) -> Self::Future {
+        InFlightResponseFuture {
+            fut: self.factory.new_service(),
+            max_inflight: self.max_inflight,
+        }
+    }
+}
+
+pub struct InFlightResponseFuture<T: NewService<Request>, Request> {
+    fut: T::Future,
+    max_inflight: usize,
+}
+
+impl<T: NewService<Request>, Request> Future for InFlightResponseFuture<T, Request> {
+    type Item = InFlightService<T::Service>;
+    type Error = T::InitError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        Ok(Async::Ready(InFlightService::with_max_inflight(
+            self.max_inflight,
+            try_ready!(self.fut.poll()),
+        )))
+    }
+}
+
+pub struct InFlightService<T> {
+    service: T,
+    count: Counter,
+}
+
+impl<T> InFlightService<T> {
+    pub fn new<F, Request>(service: F) -> Self
+    where
+        T: Service<Request>,
+        F: IntoService<T, Request>,
+    {
+        Self {
+            service: service.into_service(),
+            count: Counter::new(15),
+        }
+    }
+
+    pub fn with_max_inflight<F, Request>(max: usize, service: F) -> Self
+    where
+        T: Service<Request>,
+        F: IntoService<T, Request>,
+    {
+        Self {
+            service: service.into_service(),
+            count: Counter::new(max),
+        }
+    }
+}
+
+impl<T, Request> Service<Request> for InFlightService<T>
+where
+    T: Service<Request>,
+{
+    type Response = T::Response;
+    type Error = T::Error;
+    type Future = InFlightServiceResponse<T, Request>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        let res = self.service.poll_ready();
+        if res.is_ok() && !self.count.available() {
+            return Ok(Async::NotReady);
+        }
+        res
+    }
+
+    fn call(&mut self, req: Request) -> Self::Future {
+        InFlightServiceResponse {
+            fut: self.service.call(req),
+            _guard: self.count.get(),
+        }
+    }
+}
+
+#[doc(hidden)]
+pub struct InFlightServiceResponse<T: Service<Request>, Request> {
+    fut: T::Future,
+    _guard: CounterGuard,
+}
+
+impl<T: Service<Request>, Request> Future for InFlightServiceResponse<T, Request> {
+    type Item = T::Response;
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.fut.poll()
+    }
+}

actix-utils/src/keepalive.rs

@@ -0,0 +1,121 @@
+use std::marker::PhantomData;
+use std::time::{Duration, Instant};
+
+use actix_service::{NewService, Service};
+use futures::future::{ok, FutureResult};
+use futures::{Async, Future, Poll};
+use tokio_timer::Delay;
+
+use super::time::{LowResTime, LowResTimeService};
+use super::Never;
+
+pub struct KeepAlive<R, E, F> {
+    f: F,
+    ka: Duration,
+    time: LowResTime,
+    _t: PhantomData<(R, E)>,
+}
+
+impl<R, E, F> KeepAlive<R, E, F>
+where
+    F: Fn() -> E + Clone,
+{
+    pub fn new(ka: Duration, time: LowResTime, f: F) -> Self {
+        KeepAlive {
+            f,
+            ka,
+            time,
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<R, E, F> Clone for KeepAlive<R, E, F>
+where
+    F: Clone,
+{
+    fn clone(&self) -> Self {
+        KeepAlive {
+            f: self.f.clone(),
+            ka: self.ka,
+            time: self.time.clone(),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<R, E, F> NewService<R> for KeepAlive<R, E, F>
+where
+    F: Fn() -> E + Clone,
+{
+    type Response = R;
+    type Error = E;
+    type InitError = Never;
+    type Service = KeepAliveService<R, E, F>;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        ok(KeepAliveService::new(
+            self.ka,
+            self.time.timer(),
+            self.f.clone(),
+        ))
+    }
+}
+
+pub struct KeepAliveService<R, E, F> {
+    f: F,
+    ka: Duration,
+    time: LowResTimeService,
+    delay: Delay,
+    expire: Instant,
+    _t: PhantomData<(R, E)>,
+}
+
+impl<R, E, F> KeepAliveService<R, E, F>
+where
+    F: Fn() -> E,
+{
+    pub fn new(ka: Duration, time: LowResTimeService, f: F) -> Self {
+        let expire = time.now() + ka;
+        KeepAliveService {
+            f,
+            ka,
+            time,
+            expire,
+            delay: Delay::new(expire),
+            _t: PhantomData,
+        }
+    }
+}
+
+impl<R, E, F> Service<R> for KeepAliveService<R, E, F>
+where
+    F: Fn() -> E,
+{
+    type Response = R;
+    type Error = E;
+    type Future = FutureResult<R, E>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        match self.delay.poll() {
+            Ok(Async::Ready(_)) => {
+                let now = self.time.now();
+                if self.expire <= now {
+                    Err((self.f)())
+                } else {
+                    self.delay.reset(self.expire);
+                    let _ = self.delay.poll();
+                    Ok(Async::Ready(()))
+                }
+            }
+            Ok(Async::NotReady) => Ok(Async::Ready(())),
+            Err(_) => panic!(),
+        }
+    }
+
+    fn call(&mut self, req: R) -> Self::Future {
+        self.expire = self.time.now() + self.ka;
+        ok(req)
+    }
+}

actix-utils/src/lib.rs

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

actix-utils/src/stream.rs

@@ -0,0 +1,172 @@
+use std::marker::PhantomData;
+
+use actix_rt::spawn;
+use actix_service::{IntoService, NewService, Service};
+use futures::unsync::mpsc;
+use futures::{future, Async, Future, Poll, Stream};
+
+pub struct StreamDispatcher<S: Stream, T> {
+    stream: S,
+    service: T,
+    item: Option<Result<S::Item, S::Error>>,
+    stop_rx: mpsc::UnboundedReceiver<()>,
+    stop_tx: mpsc::UnboundedSender<()>,
+}
+
+impl<S, T> StreamDispatcher<S, T>
+where
+    S: Stream,
+    T: Service<Result<S::Item, S::Error>, Response = (), Error = ()>,
+    T::Future: 'static,
+{
+    pub fn new<F>(stream: S, service: F) -> Self
+    where
+        F: IntoService<T, Result<S::Item, S::Error>>,
+    {
+        let (stop_tx, stop_rx) = mpsc::unbounded();
+        StreamDispatcher {
+            stream,
+            item: None,
+            service: service.into_service(),
+            stop_rx,
+            stop_tx,
+        }
+    }
+}
+
+impl<S, T> Future for StreamDispatcher<S, T>
+where
+    S: Stream,
+    T: Service<Result<S::Item, S::Error>, Response = (), Error = ()>,
+    T::Future: 'static,
+{
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        if let Ok(Async::Ready(Some(_))) = self.stop_rx.poll() {
+            return Ok(Async::Ready(()));
+        }
+
+        let mut item = self.item.take();
+        loop {
+            if item.is_some() {
+                match self.service.poll_ready()? {
+                    Async::Ready(_) => spawn(StreamDispatcherService {
+                        fut: self.service.call(item.take().unwrap()),
+                        stop: self.stop_tx.clone(),
+                    }),
+                    Async::NotReady => {
+                        self.item = item;
+                        return Ok(Async::NotReady);
+                    }
+                }
+            }
+            match self.stream.poll() {
+                Ok(Async::Ready(Some(el))) => item = Some(Ok(el)),
+                Err(err) => item = Some(Err(err)),
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Ready(None)) => return Ok(Async::Ready(())),
+            }
+        }
+    }
+}
+
+struct StreamDispatcherService<F: Future> {
+    fut: F,
+    stop: mpsc::UnboundedSender<()>,
+}
+
+impl<F: Future> Future for StreamDispatcherService<F> {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self.fut.poll() {
+            Ok(Async::Ready(_)) => Ok(Async::Ready(())),
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Err(_) => {
+                let _ = self.stop.unbounded_send(());
+                Ok(Async::Ready(()))
+            }
+        }
+    }
+}
+
+/// `NewService` that implements, read one item from the stream.
+pub struct TakeItem<T> {
+    _t: PhantomData<T>,
+}
+
+impl<T> TakeItem<T> {
+    /// Create new `TakeRequest` instance.
+    pub fn new() -> Self {
+        TakeItem { _t: PhantomData }
+    }
+}
+
+impl<T> Default for TakeItem<T> {
+    fn default() -> Self {
+        TakeItem { _t: PhantomData }
+    }
+}
+
+impl<T> Clone for TakeItem<T> {
+    fn clone(&self) -> TakeItem<T> {
+        TakeItem { _t: PhantomData }
+    }
+}
+
+impl<T: Stream> NewService<T> for TakeItem<T> {
+    type Response = (Option<T::Item>, T);
+    type Error = T::Error;
+    type InitError = ();
+    type Service = TakeItemService<T>;
+    type Future = future::FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        future::ok(TakeItemService { _t: PhantomData })
+    }
+}
+
+/// `NewService` that implements, read one request from framed object feature.
+pub struct TakeItemService<T> {
+    _t: PhantomData<T>,
+}
+
+impl<T> Clone for TakeItemService<T> {
+    fn clone(&self) -> TakeItemService<T> {
+        TakeItemService { _t: PhantomData }
+    }
+}
+
+impl<T: Stream> Service<T> for TakeItemService<T> {
+    type Response = (Option<T::Item>, T);
+    type Error = T::Error;
+    type Future = TakeItemServiceResponse<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: T) -> Self::Future {
+        TakeItemServiceResponse { stream: Some(req) }
+    }
+}
+
+#[doc(hidden)]
+pub struct TakeItemServiceResponse<T: Stream> {
+    stream: Option<T>,
+}
+
+impl<T: Stream> Future for TakeItemServiceResponse<T> {
+    type Item = (Option<T::Item>, T);
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        match self.stream.as_mut().expect("Use after finish").poll()? {
+            Async::Ready(item) => Ok(Async::Ready((item, self.stream.take().unwrap()))),
+            Async::NotReady => Ok(Async::NotReady),
+        }
+    }
+}

actix-utils/src/time.rs

@@ -0,0 +1,102 @@
+use std::time::{Duration, Instant};
+
+use actix_rt::spawn;
+use actix_service::{NewService, Service};
+use futures::future::{ok, FutureResult};
+use futures::{Async, Future, Poll};
+use tokio_timer::sleep;
+
+use super::cell::Cell;
+use super::Never;
+
+#[derive(Clone, Debug)]
+pub struct LowResTime(Cell<Inner>);
+
+#[derive(Debug)]
+struct Inner {
+    resolution: Duration,
+    current: Option<Instant>,
+}
+
+impl Inner {
+    fn new(resolution: Duration) -> Self {
+        Inner {
+            resolution,
+            current: None,
+        }
+    }
+}
+
+impl LowResTime {
+    pub fn with(resolution: Duration) -> LowResTime {
+        LowResTime(Cell::new(Inner::new(resolution)))
+    }
+
+    pub fn timer(&self) -> LowResTimeService {
+        LowResTimeService(self.0.clone())
+    }
+}
+
+impl Default for LowResTime {
+    fn default() -> Self {
+        LowResTime(Cell::new(Inner::new(Duration::from_secs(1))))
+    }
+}
+
+impl NewService<()> for LowResTime {
+    type Response = Instant;
+    type Error = Never;
+    type InitError = Never;
+    type Service = LowResTimeService;
+    type Future = FutureResult<Self::Service, Self::InitError>;
+
+    fn new_service(&self) -> Self::Future {
+        ok(self.timer())
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct LowResTimeService(Cell<Inner>);
+
+impl LowResTimeService {
+    pub fn with(resolution: Duration) -> LowResTimeService {
+        LowResTimeService(Cell::new(Inner::new(resolution)))
+    }
+
+    /// Get current time. This function has to be called from
+    /// future's poll method, otherwise it panics.
+    pub fn now(&self) -> Instant {
+        let cur = self.0.borrow().current;
+        if let Some(cur) = cur {
+            cur
+        } else {
+            let now = Instant::now();
+            let inner = self.0.clone();
+            let interval = {
+                let mut b = inner.borrow_mut();
+                b.current = Some(now);
+                b.resolution
+            };
+
+            spawn(sleep(interval).map_err(|_| panic!()).and_then(move |_| {
+                inner.borrow_mut().current.take();
+                Ok(())
+            }));
+            now
+        }
+    }
+}
+
+impl Service<()> for LowResTimeService {
+    type Response = Instant;
+    type Error = Never;
+    type Future = FutureResult<Self::Response, Self::Error>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, _: ()) -> Self::Future {
+        ok(self.now())
+    }
+}

actix-utils/src/timeout.rs

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