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-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-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,8 +1,8 @@
 # Changes
 
-## [0.3.0] - 2019-12-31
+## [0.2.4] - 2019-12-31
 
-* Support named parameters for `ResourceDef::resource_path()` in form of `((&k, &v), ...)`
+* Add `ResourceDef::resource_path_named()` path generation method
 
 ## [0.2.3] - 2019-12-25
 

router/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-router"
-version = "0.3.0"
+version = "0.2.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Path router"
 keywords = ["actix"]
@@ -18,11 +18,10 @@ path = "src/lib.rs"
 default = ["http"]
 
 [dependencies]
-bytestring = "0.1.2"
-either = "1.5.3"
 regex = "1.3.1"
-log = "0.4.8"
 serde = "1.0.104"
+bytestring = "0.1.2"
+log = "0.4.8"
 http = { version="0.2.0", optional=true }
 
 [dev-dependencies]

router/src/lib.rs

@@ -4,8 +4,6 @@ mod path;
 mod resource;
 mod router;
 
-use either::Either;
-
 pub use self::de::PathDeserializer;
 pub use self::path::Path;
 pub use self::resource::ResourceDef;
@@ -37,127 +35,6 @@ impl ResourcePath for bytestring::ByteString {
     }
 }
 
-pub trait ResourceElements {
-    fn elements<F, R>(self, for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>;
-}
-
-impl<'a, T: AsRef<str>> ResourceElements for &'a [T] {
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self {
-            if let Some(res) = for_each(Either::Left(t.as_ref())) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-impl<'a, U, I> ResourceElements for &'a U
-where
-    &'a U: IntoIterator<Item = I>,
-    I: AsRef<str>,
-{
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self.into_iter() {
-            if let Some(res) = for_each(Either::Left(t.as_ref())) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-impl<I> ResourceElements for Vec<I>
-where
-    I: AsRef<str>,
-{
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self.iter() {
-            if let Some(res) = for_each(Either::Left(t.as_ref())) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-impl<'a, K, V, S> ResourceElements for std::collections::HashMap<K, V, S>
-where
-    K: AsRef<str>,
-    V: AsRef<str>,
-    S: std::hash::BuildHasher,
-{
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self.iter() {
-            if let Some(res) = for_each(Either::Right((t.0.as_ref(), t.1.as_ref()))) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-#[rustfmt::skip]
-mod _m {
-use super::*;
-// macro_rules! elements_tuple ({ $(($n:tt, $T:ident)),+} => {
-//     impl<$($T: AsRef<str>,)+> ResourceElements for ($($T,)+) {
-//         fn elements<F_, R_>(self, mut for_each: F_) -> Option<R_>
-//         where
-//             F_: FnMut(Either<&str, (&str, &str)>) -> Option<R_>,
-//         {
-//             $(
-//                 if let Some(res) = for_each(Either::Left(self.$n.as_ref())) {
-//                     return Some(res)
-//                 }
-//             )+
-//                 None
-//         }
-//     }
-// });
-
-macro_rules! elements_2tuple ({ $(($n:tt, $V:ident)),+} => {
-    impl<'a, $($V: AsRef<str>,)+> ResourceElements for ($((&'a str, $V),)+) {
-        fn elements<F_, R_>(self, mut for_each: F_) -> Option<R_>
-        where
-            F_: FnMut(Either<&str, (&str, &str)>) -> Option<R_>,
-        {
-            $(
-                if let Some(res) = for_each(Either::Right((self.$n.0, self.$n.1.as_ref()))) {
-                    return Some(res)
-                }
-            )+
-                None
-        }
-    }
-});
-
-elements_2tuple!((0, A));
-elements_2tuple!((0, A), (1, B));
-elements_2tuple!((0, A), (1, B), (2, C));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G), (7, H));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G), (7, H), (8, I));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G), (7, H), (8, I), (9, J));
-}
-
 /// Helper trait for type that could be converted to path pattern
 pub trait IntoPattern {
     /// Signle patter

router/src/resource.rs

@@ -5,7 +5,7 @@ use std::hash::{Hash, Hasher};
 use regex::{escape, Regex, RegexSet};
 
 use crate::path::{Path, PathItem};
-use crate::{IntoPattern, Resource, ResourceElements, ResourcePath};
+use crate::{IntoPattern, Resource, ResourcePath};
 
 const MAX_DYNAMIC_SEGMENTS: usize = 16;
 
@@ -464,66 +464,61 @@ impl ResourceDef {
     }
 
     /// Build resource path from elements. Returns `true` on success.
-    pub fn resource_path<U: ResourceElements>(&self, path: &mut String, data: U) -> bool {
+    pub fn resource_path<U, I>(&self, path: &mut String, elements: &mut U) -> bool
+    where
+        U: Iterator<Item = I>,
+        I: AsRef<str>,
+    {
         match self.tp {
             PatternType::Prefix(ref p) => path.push_str(p),
             PatternType::Static(ref p) => path.push_str(p),
             PatternType::Dynamic(..) => {
-                let mut iter = self.elements.iter();
-                let mut map = HashMap::new();
-
-                let result = data.elements(|item| match item {
-                    either::Either::Left(val) => loop {
-                        if let Some(el) = iter.next() {
-                            match *el {
-                                PatternElement::Str(ref s) => path.push_str(s),
-                                PatternElement::Var(_) => {
-                                    path.push_str(val.as_ref());
-                                    return None;
-                                }
-                            }
-                        } else {
-                            return Some(false);
-                        }
-                    },
-                    either::Either::Right((name, val)) => {
-                        map.insert(name.to_string(), val.to_string());
-                        None
-                    }
-                });
-
-                if result.is_some() {
-                    return true;
-                } else {
-                    if map.is_empty() {
-                        // push static sections
-                        loop {
-                            if let Some(el) = iter.next() {
-                                match *el {
-                                    PatternElement::Str(ref s) => path.push_str(s),
-                                    PatternElement::Var(_) => {
-                                        return false;
-                                    }
-                                }
+                for el in &self.elements {
+                    match *el {
+                        PatternElement::Str(ref s) => path.push_str(s),
+                        PatternElement::Var(_) => {
+                            if let Some(val) = elements.next() {
+                                path.push_str(val.as_ref())
                             } else {
-                                break;
-                            }
-                        }
-                    } else {
-                        for el in iter {
-                            match *el {
-                                PatternElement::Str(ref s) => path.push_str(s),
-                                PatternElement::Var(ref name) => {
-                                    if let Some(val) = map.get(name) {
-                                        path.push_str(val);
-                                        continue;
-                                    }
-                                    return false;
-                                }
+                                return false;
+                            }
+                        }
+                    }
+                }
+            }
+            PatternType::DynamicSet(..) => {
+                return false;
+            }
+        }
+        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;
                             }
                         }
                     }
-                    return true;
                 }
             }
             PatternType::DynamicSet(..) => {
@@ -673,7 +668,6 @@ pub(crate) fn insert_slash(path: &str) -> String {
 mod tests {
     use super::*;
     use http::Uri;
-    use std::collections::HashMap;
     use std::convert::TryFrom;
 
     #[test]
@@ -905,66 +899,45 @@ mod tests {
     fn test_resource_path() {
         let mut s = String::new();
         let resource = ResourceDef::new("/user/{item1}/test");
-        assert!(resource.resource_path(&mut s, &["user1"]));
-        assert_eq!(s, "/user/user1/test");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "user1"),)));
+        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, &["item", "item2"]));
-        assert_eq!(s, "/user/item/item2/test");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "item"), ("item2", "item2"))));
+        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, &["item", "item2"]));
-        assert_eq!(s, "/user/item/item2");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "item"), ("item2", "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, &["item", "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, &["item"]));
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).into_iter()));
 
         let mut s = String::new();
-        assert!(resource.resource_path(&mut s, &["item", "item2"]));
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).into_iter()));
         assert_eq!(s, "/user/item/item2/");
-        assert!(!resource.resource_path(&mut s, &["item"]));
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).into_iter()));
 
         let mut s = String::new();
-        assert!(resource.resource_path(&mut s, vec!["item", "item2"]));
-        assert_eq!(s, "/user/item/item2/");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, &vec!["item", "item2"]));
-        assert_eq!(s, "/user/item/item2/");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, &vec!["item", "item2"][..]));
+        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");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, map));
-        assert_eq!(s, "/user/item/item2/");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "item"), ("item2", "item2"))));
+        assert!(resource.resource_path_named(&mut s, &map));
         assert_eq!(s, "/user/item/item2/");
     }
 }

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]