rename .run to .start()

.github/workflows/main.yml

@@ -2,6 +2,9 @@ name: CI
 
 on: [push, pull_request]
 
+env:
+  VCPKGRS_DYNAMIC: 1
+
 jobs:
   build_and_test:
     strategy:
@@ -18,10 +21,12 @@ jobs:
         include:
         - toolchain: x86_64-pc-windows-msvc
           os: windows-latest
+          arch: x64
         - toolchain: x86_64-pc-windows-gnu
           os: windows-latest
         - toolchain: i686-pc-windows-msvc
           os: windows-latest
+          arch: x86
         - toolchain: x86_64-apple-darwin
           os: macOS-latest
 
@@ -37,6 +42,13 @@ jobs:
           toolchain: ${{ matrix.version }}-${{ matrix.toolchain }}
           default: true
 
+
+      - name: Install OpenSSL
+        if: matrix.toolchain == 'x86_64-pc-windows-msvc' || matrix.toolchain == 'i686-pc-windows-msvc'
+        run: |
+          vcpkg integrate install
+          vcpkg install openssl:${{ matrix.arch }}-windows
+
       - name: check nightly
         if: matrix.version == 'nightly'
         uses: actions-rs/cargo@v1
@@ -52,6 +64,14 @@ jobs:
           args: --all --bins --examples --tests
 
       - name: tests
+        if: matrix.toolchain != 'x86_64-pc-windows-gnu'
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --all --all-features -- --nocapture
+
+      - name: tests on x86_64-pc-windows-gnu
+        if: matrix.toolchain == 'x86_64-pc-windows-gnu'
         uses: actions-rs/cargo@v1
         with:
           command: test

actix-codec/CHANGES.md

@@ -1,5 +1,7 @@
 # Changes
 
+* Use `.advance()` intead of `.split_to()`
+
 ## [0.2.0] - 2019-12-10
 
 * Use specific futures dependencies

actix-codec/src/framed.rs

@@ -2,7 +2,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::{fmt, io};
 
-use bytes::BytesMut;
+use bytes::{Buf, BytesMut};
 use futures_core::{ready, Stream};
 use futures_sink::Sink;
 
@@ -288,7 +288,7 @@ impl<T, U> Framed<T, U> {
             }
 
             // remove written data
-            let _ = self.write_buf.split_to(n);
+            self.write_buf.advance(n);
         }
 
         // Try flushing the underlying IO

actix-connect/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [1.0.1] - 2019-12-15
+
+* Fix trust-dns-resolver compilation
+
+## [1.0.0] - 2019-12-11
+
+* Release
+
 ## [1.0.0-alpha.3] - 2019-12-07
 
 ### Changed

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "1.0.0-alpha.3"
+version = "1.0.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix connect - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,6 @@ repository = "https://github.com/actix/actix-net.git"
 documentation = "https://docs.rs/actix-connect/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
-exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
 workspace = ".."
 
@@ -33,16 +32,17 @@ rustls = ["rust-tls", "tokio-rustls", "webpki"]
 uri = ["http"]
 
 [dependencies]
-actix-service = "1.0.0-alpha.3"
+actix-service = "1.0.0"
 actix-codec = "0.2.0"
-actix-utils = "1.0.0-alpha.3"
-actix-rt = "1.0.0-alpha.3"
+actix-utils = "1.0.3"
+actix-rt = "1.0.0"
 derive_more = "0.99.2"
 either = "1.5.2"
 futures = "0.3.1"
 http = { version = "0.2.0", optional = true }
 log = "0.4"
-trust-dns-resolver = { version="0.18.0-alpha.2", default-features = false, features=["tokio"] }
+trust-dns-proto = "=0.18.0-alpha.2"
+trust-dns-resolver = "=0.18.0-alpha.2"
 
 # openssl
 open-ssl = { version="0.10", package = "openssl", optional = true }
@@ -54,5 +54,5 @@ tokio-rustls = { version = "0.12.0", optional = true }
 webpki = { version = "0.21", optional = true }
 
 [dev-dependencies]
-bytes = "0.5.2"
-actix-testing = { version="1.0.0-alpha.2" }
+bytes = "0.5.3"
+actix-testing = { version="1.0.0" }

actix-ioframe/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [0.4.1] - 2019-12-11
+
+* Disconnect callback accepts owned state
+
+## [0.4.0] - 2019-12-11
+
+* Remove `E` param
+
 ## [0.3.0-alpha.3] - 2019-12-07
 
 * Migrate to tokio 0.2

actix-ioframe/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-ioframe"
-version = "0.3.0-alpha.3"
+version = "0.4.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix framed service"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,6 @@ repository = "https://github.com/actix/actix-net.git"
 documentation = "https://docs.rs/actix-ioframe/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
-exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
 workspace = ".."
 
@@ -18,10 +17,10 @@ name = "actix_ioframe"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "1.0.0-alpha.3"
+actix-service = "1.0.0"
 actix-codec = "0.2.0"
-actix-utils = "1.0.0-alpha.2"
-actix-rt = "1.0.0-alpha.2"
+actix-utils = "1.0.1"
+actix-rt = "1.0.0"
 bytes = "0.5"
 either = "1.5.2"
 futures = "0.3.1"
@@ -29,5 +28,5 @@ pin-project = "0.4.6"
 log = "0.4"
 
 [dev-dependencies]
-actix-connect = "1.0.0-alpha.2"
-actix-testing = "1.0.0-alpha.2"
+actix-connect = "1.0.0"
+actix-testing = "1.0.0"

actix-ioframe/src/cell.rs

@@ -1,39 +0,0 @@
-//! Custom cell impl
-
-use std::cell::UnsafeCell;
-use std::fmt;
-use std::rc::Rc;
-
-pub(crate) struct Cell<T> {
-    inner: Rc<UnsafeCell<T>>,
-}
-
-impl<T> Clone for Cell<T> {
-    fn clone(&self) -> Self {
-        Self {
-            inner: self.inner.clone(),
-        }
-    }
-}
-
-impl<T: fmt::Debug> fmt::Debug for Cell<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        self.inner.fmt(f)
-    }
-}
-
-impl<T> Cell<T> {
-    pub fn new(inner: T) -> Self {
-        Self {
-            inner: Rc::new(UnsafeCell::new(inner)),
-        }
-    }
-
-    pub(crate) unsafe fn get_ref(&mut self) -> &T {
-        &*self.inner.as_ref().get()
-    }
-
-    pub(crate) unsafe fn get_mut(&mut self) -> &mut T {
-        &mut *self.inner.as_ref().get()
-    }
-}

actix-ioframe/src/connect.rs

@@ -3,40 +3,37 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
-use actix_utils::mpsc;
 use futures::Stream;
 
-use crate::dispatcher::FramedMessage;
 use crate::sink::Sink;
 
-pub struct Connect<Io, St = (), Codec = ()> {
+pub struct Connect<Io, Codec, St = ()>
+where
+    Codec: Encoder + Decoder,
+{
     io: Io,
+    sink: Sink<<Codec as Encoder>::Item>,
     _t: PhantomData<(St, Codec)>,
 }
 
-impl<Io> Connect<Io>
+impl<Io, Codec> Connect<Io, Codec>
 where
     Io: AsyncRead + AsyncWrite,
+    Codec: Encoder + Decoder,
 {
-    pub(crate) fn new(io: Io) -> Self {
+    pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item>) -> Self {
         Self {
             io,
+            sink,
             _t: PhantomData,
         }
     }
 
-    pub fn codec<Codec>(self, codec: Codec) -> ConnectResult<Io, (), Codec>
-    where
-        Codec: Encoder + Decoder,
-    {
-        let (tx, rx) = mpsc::channel();
-        let sink = Sink::new(tx);
-
+    pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec> {
         ConnectResult {
             state: (),
+            sink: self.sink,
             framed: Framed::new(self.io, codec),
-            rx,
-            sink,
         }
     }
 }
@@ -45,7 +42,6 @@ where
 pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
     pub(crate) state: St,
     pub(crate) framed: Framed<Io, Codec>,
-    pub(crate) rx: mpsc::Receiver<FramedMessage<<Codec as Encoder>::Item>>,
     pub(crate) sink: Sink<<Codec as Encoder>::Item>,
 }
 
@@ -70,7 +66,6 @@ impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
         ConnectResult {
             state,
             framed: self.framed,
-            rx: self.rx,
             sink: self.sink,
         }
     }

actix-ioframe/src/dispatcher.rs

@@ -1,19 +1,14 @@
 //! Framed dispatcher service and related utilities
-use std::collections::VecDeque;
-use std::mem;
 use std::pin::Pin;
 use std::rc::Rc;
 use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, Service};
-use actix_utils::task::LocalWaker;
 use actix_utils::{mpsc, oneshot};
-use futures::future::ready;
-use futures::{FutureExt, Sink as FutureSink, Stream};
+use futures::{FutureExt, Stream};
 use log::debug;
 
-use crate::cell::Cell;
 use crate::error::ServiceError;
 use crate::item::Item;
 use crate::sink::Sink;
@@ -21,16 +16,16 @@ use crate::sink::Sink;
 type Request<S, U> = Item<S, U>;
 type Response<U> = <U as Encoder>::Item;
 
-pub(crate) enum FramedMessage<T> {
-    Message(T),
-    Close,
+pub(crate) enum Message<T> {
+    Item(T),
     WaitClose(oneshot::Sender<()>),
+    Close,
 }
 
 /// FramedTransport - is a future that reads frames from Framed object
 /// and pass then to the service.
 #[pin_project::pin_project]
-pub(crate) struct FramedDispatcher<St, S, T, U>
+pub(crate) struct Dispatcher<St, S, T, U>
 where
     St: Clone,
     S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
@@ -46,12 +41,12 @@ where
     state: St,
     dispatch_state: FramedState<S, U>,
     framed: Framed<T, U>,
-    rx: Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
-    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
+    rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
+    tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
+    disconnect: Option<Rc<dyn Fn(St, bool)>>,
 }
 
-impl<St, S, T, U> FramedDispatcher<St, S, T, U>
+impl<St, S, T, U> Dispatcher<St, S, T, U>
 where
     St: Clone,
     S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
@@ -66,22 +61,21 @@ where
         framed: Framed<T, U>,
         state: St,
         service: F,
-        rx: mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>,
         sink: Sink<<U as Encoder>::Item>,
-        disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
+        rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
+        disconnect: Option<Rc<dyn Fn(St, bool)>>,
     ) -> Self {
-        FramedDispatcher {
+        let tx = rx.sender();
+
+        Dispatcher {
             framed,
             state,
             sink,
             disconnect,
-            rx: Some(rx),
+            rx,
+            tx,
             service: service.into_service(),
             dispatch_state: FramedState::Processing,
-            inner: Cell::new(FramedDispatcherInner {
-                buf: VecDeque::new(),
-                task: LocalWaker::new(),
-            }),
         }
     }
 }
@@ -116,14 +110,23 @@ impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
             }
         }
     }
+
+    fn take_error(&mut self) -> ServiceError<S::Error, U> {
+        match std::mem::replace(self, FramedState::Processing) {
+            FramedState::Error(err) => err,
+            _ => panic!(),
+        }
+    }
+
+    fn take_framed_error(&mut self) -> ServiceError<S::Error, U> {
+        match std::mem::replace(self, FramedState::Processing) {
+            FramedState::FramedError(err) => err,
+            _ => panic!(),
+        }
+    }
 }
 
-struct FramedDispatcherInner<I, E> {
-    buf: VecDeque<Result<I, E>>,
-    task: LocalWaker,
-}
-
-impl<St, S, T, U> FramedDispatcher<St, S, T, U>
+impl<St, S, T, U> Dispatcher<St, S, T, U>
 where
     St: Clone,
     S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
@@ -134,263 +137,150 @@ where
     <U as Encoder>::Item: 'static,
     <U as Encoder>::Error: std::fmt::Debug,
 {
+    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool {
+        loop {
+            match self.service.poll_ready(cx) {
+                Poll::Ready(Ok(_)) => {
+                    let item = match self.framed.next_item(cx) {
+                        Poll::Ready(Some(Ok(el))) => el,
+                        Poll::Ready(Some(Err(err))) => {
+                            self.dispatch_state =
+                                FramedState::FramedError(ServiceError::Decoder(err));
+                            return true;
+                        }
+                        Poll::Pending => return false,
+                        Poll::Ready(None) => {
+                            log::trace!("Client disconnected");
+                            self.dispatch_state = FramedState::Stopping;
+                            return true;
+                        }
+                    };
+
+                    let tx = self.tx.clone();
+                    actix_rt::spawn(
+                        self.service
+                            .call(Item::new(self.state.clone(), self.sink.clone(), item))
+                            .map(move |item| {
+                                let item = match item {
+                                    Ok(Some(item)) => Ok(Message::Item(item)),
+                                    Ok(None) => return,
+                                    Err(err) => Err(err),
+                                };
+                                let _ = tx.send(item);
+                            }),
+                    );
+                }
+                Poll::Pending => return false,
+                Poll::Ready(Err(err)) => {
+                    self.dispatch_state = FramedState::Error(ServiceError::Service(err));
+                    return true;
+                }
+            }
+        }
+    }
+
+    /// write to framed object
+    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool {
+        loop {
+            while !self.framed.is_write_buf_full() {
+                match Pin::new(&mut self.rx).poll_next(cx) {
+                    Poll::Ready(Some(Ok(Message::Item(msg)))) => {
+                        if let Err(err) = self.framed.write(msg) {
+                            self.dispatch_state =
+                                FramedState::FramedError(ServiceError::Encoder(err));
+                            return true;
+                        }
+                    }
+                    Poll::Ready(Some(Ok(Message::Close))) => {
+                        self.dispatch_state.stop(None);
+                        return true;
+                    }
+                    Poll::Ready(Some(Ok(Message::WaitClose(tx)))) => {
+                        self.dispatch_state.stop(Some(tx));
+                        return true;
+                    }
+                    Poll::Ready(Some(Err(err))) => {
+                        self.dispatch_state = FramedState::Error(ServiceError::Service(err));
+                        return true;
+                    }
+                    Poll::Ready(None) | Poll::Pending => break,
+                }
+            }
+
+            if !self.framed.is_write_buf_empty() {
+                match self.framed.flush(cx) {
+                    Poll::Pending => break,
+                    Poll::Ready(Ok(_)) => (),
+                    Poll::Ready(Err(err)) => {
+                        debug!("Error sending data: {:?}", err);
+                        self.dispatch_state =
+                            FramedState::FramedError(ServiceError::Encoder(err));
+                        return true;
+                    }
+                }
+            } else {
+                break;
+            }
+        }
+        false
+    }
+
     pub(crate) fn poll(
         &mut self,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), ServiceError<S::Error, U>>> {
-        let this = self;
-        unsafe { this.inner.get_ref().task.register(cx.waker()) };
-
-        poll(
-            cx,
-            &mut this.service,
-            &mut this.state,
-            &mut this.sink,
-            &mut this.framed,
-            &mut this.dispatch_state,
-            &mut this.rx,
-            &mut this.inner,
-            &mut this.disconnect,
-        )
-    }
-}
-
-fn poll<St, S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut St,
-    sink: &mut Sink<<U as Encoder>::Item>,
-    framed: &mut Framed<T, U>,
-    dispatch_state: &mut FramedState<S, U>,
-    rx: &mut Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: &mut Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
-    disconnect: &mut Option<Rc<dyn Fn(&mut St, bool)>>,
-) -> Poll<Result<(), ServiceError<S::Error, U>>>
-where
-    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    match mem::replace(dispatch_state, FramedState::Processing) {
-        FramedState::Processing => {
-            if poll_read(cx, srv, state, sink, framed, dispatch_state, inner)
-                || poll_write(cx, framed, dispatch_state, rx, inner)
-            {
-                poll(
-                    cx,
-                    srv,
-                    state,
-                    sink,
-                    framed,
-                    dispatch_state,
-                    rx,
-                    inner,
-                    disconnect,
-                )
-            } else {
-                Poll::Pending
-            }
-        }
-        FramedState::Error(err) => {
-            if framed.is_write_buf_empty()
-                || (poll_write(cx, framed, dispatch_state, rx, inner)
-                    || framed.is_write_buf_empty())
-            {
-                if let Some(ref disconnect) = disconnect {
-                    (&*disconnect)(&mut *state, true);
+        match self.dispatch_state {
+            FramedState::Processing => {
+                if self.poll_read(cx) || self.poll_write(cx) {
+                    self.poll(cx)
+                } else {
+                    Poll::Pending
                 }
-                Poll::Ready(Err(err))
-            } else {
-                *dispatch_state = FramedState::Error(err);
-                Poll::Pending
             }
-        }
-        FramedState::FlushAndStop(mut vec) => {
-            if !framed.is_write_buf_empty() {
-                match Pin::new(framed).poll_flush(cx) {
-                    Poll::Ready(Err(err)) => {
-                        debug!("Error sending data: {:?}", err);
-                    }
-                    Poll::Pending => {
-                        *dispatch_state = FramedState::FlushAndStop(vec);
+            FramedState::Error(_) => {
+                // flush write buffer
+                if !self.framed.is_write_buf_empty() {
+                    if let Poll::Pending = self.framed.flush(cx) {
                         return Poll::Pending;
                     }
-                    Poll::Ready(_) => (),
                 }
-            };
-            for tx in vec.drain(..) {
-                let _ = tx.send(());
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(self.state.clone(), true);
+                }
+                Poll::Ready(Err(self.dispatch_state.take_error()))
             }
-            if let Some(ref disconnect) = disconnect {
-                (&*disconnect)(&mut *state, false);
-            }
-            Poll::Ready(Ok(()))
-        }
-        FramedState::FramedError(err) => {
-            if let Some(ref disconnect) = disconnect {
-                (&*disconnect)(&mut *state, true);
-            }
-            Poll::Ready(Err(err))
-        }
-        FramedState::Stopping => {
-            if let Some(ref disconnect) = disconnect {
-                (&*disconnect)(&mut *state, false);
-            }
-            Poll::Ready(Ok(()))
-        }
-    }
-}
-
-fn poll_read<St, S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut St,
-    sink: &mut Sink<<U as Encoder>::Item>,
-    framed: &mut Framed<T, U>,
-    dispatch_state: &mut FramedState<S, U>,
-    inner: &mut Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
-) -> bool
-where
-    St: Clone,
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    loop {
-        match srv.poll_ready(cx) {
-            Poll::Ready(Ok(_)) => {
-                let item = match framed.next_item(cx) {
-                    Poll::Ready(Some(Ok(el))) => el,
-                    Poll::Ready(Some(Err(err))) => {
-                        *dispatch_state = FramedState::FramedError(ServiceError::Decoder(err));
-                        return true;
-                    }
-                    Poll::Pending => return false,
-                    Poll::Ready(None) => {
-                        log::trace!("Client disconnected");
-                        *dispatch_state = FramedState::Stopping;
-                        return true;
+            FramedState::FlushAndStop(ref mut vec) => {
+                if !self.framed.is_write_buf_empty() {
+                    match self.framed.flush(cx) {
+                        Poll::Ready(Err(err)) => {
+                            debug!("Error sending data: {:?}", err);
+                        }
+                        Poll::Pending => {
+                            return Poll::Pending;
+                        }
+                        Poll::Ready(_) => (),
                     }
                 };
-
-                let mut cell = inner.clone();
-                actix_rt::spawn(srv.call(Item::new(state.clone(), sink.clone(), item)).then(
-                    move |item| {
-                        let item = match item {
-                            Ok(Some(item)) => Ok(item),
-                            Ok(None) => return ready(()),
-                            Err(err) => Err(err),
-                        };
-                        unsafe {
-                            let inner = cell.get_mut();
-                            inner.buf.push_back(item);
-                            inner.task.wake();
-                        }
-                        ready(())
-                    },
-                ));
+                for tx in vec.drain(..) {
+                    let _ = tx.send(());
+                }
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(self.state.clone(), false);
+                }
+                Poll::Ready(Ok(()))
             }
-            Poll::Pending => return false,
-            Poll::Ready(Err(err)) => {
-                *dispatch_state = FramedState::Error(ServiceError::Service(err));
-                return true;
+            FramedState::FramedError(_) => {
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(self.state.clone(), true);
+                }
+                Poll::Ready(Err(self.dispatch_state.take_framed_error()))
+            }
+            FramedState::Stopping => {
+                if let Some(ref disconnect) = self.disconnect {
+                    (&*disconnect)(self.state.clone(), false);
+                }
+                Poll::Ready(Ok(()))
             }
         }
     }
 }
-
-/// write to framed object
-fn poll_write<St, S, T, U>(
-    cx: &mut Context<'_>,
-    framed: &mut Framed<T, U>,
-    dispatch_state: &mut FramedState<S, U>,
-    rx: &mut Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: &mut Cell<FramedDispatcherInner<<U as Encoder>::Item, S::Error>>,
-) -> bool
-where
-    S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    let inner = unsafe { inner.get_mut() };
-    let mut rx_done = rx.is_none();
-    let mut buf_empty = inner.buf.is_empty();
-    loop {
-        while !framed.is_write_buf_full() {
-            if !buf_empty {
-                match inner.buf.pop_front().unwrap() {
-                    Ok(msg) => {
-                        if let Err(err) = framed.write(msg) {
-                            *dispatch_state =
-                                FramedState::FramedError(ServiceError::Encoder(err));
-                            return true;
-                        }
-                        buf_empty = inner.buf.is_empty();
-                    }
-                    Err(err) => {
-                        *dispatch_state = FramedState::Error(ServiceError::Service(err));
-                        return true;
-                    }
-                }
-            }
-
-            if !rx_done && rx.is_some() {
-                match Pin::new(rx.as_mut().unwrap()).poll_next(cx) {
-                    Poll::Ready(Some(FramedMessage::Message(msg))) => {
-                        if let Err(err) = framed.write(msg) {
-                            *dispatch_state =
-                                FramedState::FramedError(ServiceError::Encoder(err));
-                            return true;
-                        }
-                    }
-                    Poll::Ready(Some(FramedMessage::Close)) => {
-                        dispatch_state.stop(None);
-                        return true;
-                    }
-                    Poll::Ready(Some(FramedMessage::WaitClose(tx))) => {
-                        dispatch_state.stop(Some(tx));
-                        return true;
-                    }
-                    Poll::Ready(None) => {
-                        rx_done = true;
-                        let _ = rx.take();
-                    }
-                    Poll::Pending => rx_done = true,
-                }
-            }
-
-            if rx_done && buf_empty {
-                break;
-            }
-        }
-
-        if !framed.is_write_buf_empty() {
-            match framed.flush(cx) {
-                Poll::Pending => break,
-                Poll::Ready(Err(err)) => {
-                    debug!("Error sending data: {:?}", err);
-                    *dispatch_state = FramedState::FramedError(ServiceError::Encoder(err));
-                    return true;
-                }
-                Poll::Ready(_) => (),
-            }
-        } else {
-            break;
-        }
-    }
-
-    false
-}

actix-ioframe/src/item.rs

@@ -77,6 +77,6 @@ where
     <Codec as Decoder>::Item: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_tuple("FramedItem").field(&self.item).finish()
+        f.debug_tuple("Item").field(&self.item).finish()
     }
 }

actix-ioframe/src/lib.rs

@@ -1,7 +1,6 @@
 #![deny(rust_2018_idioms, warnings)]
 #![allow(clippy::type_complexity, clippy::too_many_arguments)]
 
-mod cell;
 mod connect;
 mod dispatcher;
 mod error;

actix-ioframe/src/service.rs

@@ -6,17 +6,20 @@ use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder};
 use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
+use actix_utils::mpsc;
 use either::Either;
 use futures::future::{FutureExt, LocalBoxFuture};
 use pin_project::project;
 
 use crate::connect::{Connect, ConnectResult};
-use crate::dispatcher::FramedDispatcher;
+use crate::dispatcher::{Dispatcher, Message};
 use crate::error::ServiceError;
 use crate::item::Item;
+use crate::sink::Sink;
 
 type RequestItem<S, U> = Item<S, U>;
 type ResponseItem<U> = Option<<U as Encoder>::Item>;
+type ServiceResult<U, E> = Result<Message<<U as Encoder>::Item>, E>;
 
 /// Service builder - structure that follows the builder pattern
 /// for building instances for framed services.
@@ -38,7 +41,7 @@ impl<St: Clone, Codec> Builder<St, Codec> {
     where
         F: IntoService<C>,
         Io: AsyncRead + AsyncWrite,
-        C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
+        C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
         Codec: Decoder + Encoder,
     {
         ServiceBuilder {
@@ -55,7 +58,7 @@ impl<St: Clone, Codec> Builder<St, Codec> {
         Io: AsyncRead + AsyncWrite,
         C: ServiceFactory<
             Config = (),
-            Request = Connect<Io>,
+            Request = Connect<Io, Codec>,
             Response = ConnectResult<Io, St, Codec>,
         >,
         C::Error: 'static,
@@ -72,15 +75,14 @@ impl<St: Clone, Codec> Builder<St, Codec> {
 
 pub struct ServiceBuilder<St, C, Io, Codec> {
     connect: C,
-    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
+    disconnect: Option<Rc<dyn Fn(St, bool)>>,
     _t: PhantomData<(St, Io, Codec)>,
 }
 
 impl<St, C, Io, Codec> ServiceBuilder<St, C, Io, Codec>
 where
     St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
-    C::Error: 'static,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
     Io: AsyncRead + AsyncWrite,
     Codec: Decoder + Encoder,
     <Codec as Encoder>::Item: 'static,
@@ -91,7 +93,7 @@ where
     /// Second parameter indicates error occured during disconnect.
     pub fn disconnect<F, Out>(mut self, disconnect: F) -> Self
     where
-        F: Fn(&mut St, bool) + 'static,
+        F: Fn(St, bool) + 'static,
     {
         self.disconnect = Some(Rc::new(disconnect));
         self
@@ -102,12 +104,12 @@ where
     where
         F: IntoServiceFactory<T>,
         T: ServiceFactory<
-                Config = St,
-                Request = RequestItem<St, Codec>,
-                Response = ResponseItem<Codec>,
-                Error = C::Error,
-                InitError = C::Error,
-            > + 'static,
+            Config = St,
+            Request = RequestItem<St, Codec>,
+            Response = ResponseItem<Codec>,
+            Error = C::Error,
+            InitError = C::Error,
+        >,
     {
         FramedServiceImpl {
             connect: self.connect,
@@ -120,7 +122,7 @@ where
 
 pub struct NewServiceBuilder<St, C, Io, Codec> {
     connect: C,
-    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
+    disconnect: Option<Rc<dyn Fn(St, bool)>>,
     _t: PhantomData<(St, Io, Codec)>,
 }
 
@@ -130,7 +132,7 @@ where
     Io: AsyncRead + AsyncWrite,
     C: ServiceFactory<
         Config = (),
-        Request = Connect<Io>,
+        Request = Connect<Io, Codec>,
         Response = ConnectResult<Io, St, Codec>,
     >,
     C::Error: 'static,
@@ -144,7 +146,7 @@ where
     /// Second parameter indicates error occured during disconnect.
     pub fn disconnect<F>(mut self, disconnect: F) -> Self
     where
-        F: Fn(&mut St, bool) + 'static,
+        F: Fn(St, bool) + 'static,
     {
         self.disconnect = Some(Rc::new(disconnect));
         self
@@ -173,7 +175,7 @@ where
 pub struct FramedService<St, C, T, Io, Codec, Cfg> {
     connect: C,
     handler: Rc<T>,
-    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
+    disconnect: Option<Rc<dyn Fn(St, bool)>>,
     _t: PhantomData<(St, Io, Codec, Cfg)>,
 }
 
@@ -183,7 +185,7 @@ where
     Io: AsyncRead + AsyncWrite,
     C: ServiceFactory<
         Config = (),
-        Request = Connect<Io>,
+        Request = Connect<Io, Codec>,
         Response = ConnectResult<Io, St, Codec>,
     >,
     C::Error: 'static,
@@ -230,7 +232,7 @@ where
 pub struct FramedServiceImpl<St, C, T, Io, Codec> {
     connect: C,
     handler: Rc<T>,
-    disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
+    disconnect: Option<Rc<dyn Fn(St, bool)>>,
     _t: PhantomData<(St, Io, Codec)>,
 }
 
@@ -238,7 +240,7 @@ impl<St, C, T, Io, Codec> Service for FramedServiceImpl<St, C, T, Io, Codec>
 where
     St: Clone,
     Io: AsyncRead + AsyncWrite,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
     C::Error: 'static,
     T: ServiceFactory<
         Config = St,
@@ -262,11 +264,16 @@ where
     }
 
     fn call(&mut self, req: Io) -> Self::Future {
+        let (tx, rx) = mpsc::channel();
+        let sink = Sink::new(Rc::new(move |msg| {
+            let _ = tx.send(Ok(msg));
+        }));
         FramedServiceImplResponse {
             inner: FramedServiceImplResponseInner::Connect(
-                self.connect.call(Connect::new(req)),
+                self.connect.call(Connect::new(req, sink.clone())),
                 self.handler.clone(),
                 self.disconnect.clone(),
+                Some(rx),
             ),
         }
     }
@@ -276,7 +283,7 @@ where
 pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
 where
     St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
     C::Error: 'static,
     T: ServiceFactory<
         Config = St,
@@ -298,7 +305,7 @@ where
 impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
 where
     St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
     C::Error: 'static,
     T: ServiceFactory<
         Config = St,
@@ -334,7 +341,7 @@ where
 enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
 where
     St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
     C::Error: 'static,
     T: ServiceFactory<
         Config = St,
@@ -349,19 +356,25 @@ where
     <Codec as Encoder>::Item: 'static,
     <Codec as Encoder>::Error: std::fmt::Debug,
 {
-    Connect(#[pin] C::Future, Rc<T>, Option<Rc<dyn Fn(&mut St, bool)>>),
+    Connect(
+        #[pin] C::Future,
+        Rc<T>,
+        Option<Rc<dyn Fn(St, bool)>>,
+        Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
+    ),
     Handler(
         #[pin] T::Future,
         Option<ConnectResult<Io, St, Codec>>,
-        Option<Rc<dyn Fn(&mut St, bool)>>,
+        Option<Rc<dyn Fn(St, bool)>>,
+        Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
     ),
-    Dispatcher(#[pin] FramedDispatcher<St, T::Service, Io, Codec>),
+    Dispatcher(Dispatcher<St, T::Service, Io, Codec>),
 }
 
 impl<St, Io, Codec, C, T> FramedServiceImplResponseInner<St, Io, Codec, C, T>
 where
     St: Clone,
-    C: Service<Request = Connect<Io>, Response = ConnectResult<Io, St, Codec>>,
+    C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
     C::Error: 'static,
     T: ServiceFactory<
         Config = St,
@@ -386,37 +399,39 @@ where
     > {
         #[project]
         match self.project() {
-            FramedServiceImplResponseInner::Connect(fut, handler, disconnect) => {
+            FramedServiceImplResponseInner::Connect(fut, handler, disconnect, rx) => {
                 match fut.poll(cx) {
                     Poll::Ready(Ok(res)) => {
                         Either::Left(FramedServiceImplResponseInner::Handler(
                             handler.new_service(res.state.clone()),
                             Some(res),
                             disconnect.take(),
+                            rx.take(),
                         ))
                     }
                     Poll::Pending => Either::Right(Poll::Pending),
                     Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
                 }
             }
-            FramedServiceImplResponseInner::Handler(fut, res, disconnect) => match fut.poll(cx)
-            {
-                Poll::Ready(Ok(handler)) => {
-                    let res = res.take().unwrap();
-                    Either::Left(FramedServiceImplResponseInner::Dispatcher(
-                        FramedDispatcher::new(
-                            res.framed,
-                            res.state,
-                            handler,
-                            res.rx,
-                            res.sink,
-                            disconnect.take(),
-                        ),
-                    ))
+            FramedServiceImplResponseInner::Handler(fut, res, disconnect, rx) => {
+                match fut.poll(cx) {
+                    Poll::Ready(Ok(handler)) => {
+                        let res = res.take().unwrap();
+                        Either::Left(FramedServiceImplResponseInner::Dispatcher(
+                            Dispatcher::new(
+                                res.framed,
+                                res.state,
+                                handler,
+                                res.sink,
+                                rx.take().unwrap(),
+                                disconnect.take(),
+                            ),
+                        ))
+                    }
+                    Poll::Pending => Either::Right(Poll::Pending),
+                    Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
                 }
-                Poll::Pending => Either::Right(Poll::Pending),
-                Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
-            },
+            }
             FramedServiceImplResponseInner::Dispatcher(ref mut fut) => {
                 Either::Right(fut.poll(cx))
             }

actix-ioframe/src/sink.rs

@@ -1,11 +1,12 @@
 use std::fmt;
+use std::rc::Rc;
 
-use actix_utils::{mpsc, oneshot};
+use actix_utils::oneshot;
 use futures::future::{Future, FutureExt};
 
-use crate::dispatcher::FramedMessage;
+use crate::dispatcher::Message;
 
-pub struct Sink<T>(mpsc::Sender<FramedMessage<T>>);
+pub struct Sink<T>(Rc<dyn Fn(Message<T>)>);
 
 impl<T> Clone for Sink<T> {
     fn clone(&self) -> Self {
@@ -14,26 +15,26 @@ impl<T> Clone for Sink<T> {
 }
 
 impl<T> Sink<T> {
-    pub(crate) fn new(tx: mpsc::Sender<FramedMessage<T>>) -> Self {
+    pub(crate) fn new(tx: Rc<dyn Fn(Message<T>)>) -> Self {
         Sink(tx)
     }
 
     /// Close connection
     pub fn close(&self) {
-        let _ = self.0.send(FramedMessage::Close);
+        (self.0)(Message::Close);
     }
 
     /// Close connection
     pub fn wait_close(&self) -> impl Future<Output = ()> {
         let (tx, rx) = oneshot::channel();
-        let _ = self.0.send(FramedMessage::WaitClose(tx));
+        (self.0)(Message::WaitClose(tx));
 
         rx.map(|_| ())
     }
 
     /// Send item
     pub fn send(&self, item: T) {
-        let _ = self.0.send(FramedMessage::Message(item));
+        (self.0)(Message::Item(item));
     }
 }
 

actix-ioframe/tests/test_server.rs

@@ -22,7 +22,7 @@ async fn test_disconnect() -> std::io::Result<()> {
         let disconnect1 = disconnect1.clone();
 
         Builder::new()
-            .factory(fn_service(|conn: Connect<_>| {
+            .factory(fn_service(|conn: Connect<_, _>| {
                 ok(conn.codec(BytesCodec).state(State))
             }))
             .disconnect(move |_, _| {
@@ -32,7 +32,7 @@ async fn test_disconnect() -> std::io::Result<()> {
     });
 
     let mut client = Builder::new()
-        .service(|conn: Connect<_>| {
+        .service(|conn: Connect<_, _>| {
             let conn = conn.codec(BytesCodec).state(State);
             conn.sink().close();
             ok(conn)

actix-macros/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-macros"
-version = "0.1.0-alpha.1"
+version = "0.1.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime macros"
 repository = "https://github.com/actix/actix-net"
@@ -18,4 +18,4 @@ quote = "^1"
 syn = { version = "^1", features = ["full"] }
 
 [dev-dependencies]
-actix-rt = { version = "1.0.0-alpha.1" }
+actix-rt = { version = "1.0.0" }

actix-macros/LICENSE-APACHE

@@ -0,0 +1 @@
+../LICENSE-APACHE

actix-macros/LICENSE-MIT

@@ -0,0 +1 @@
+../LICENSE-MIT

actix-macros/src/lib.rs

@@ -71,7 +71,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
     if input.sig.asyncness.is_none() {
         return syn::Error::new_spanned(
             input.sig.fn_token,
-            format!("only async fn is supported, {:?}", input.sig.ident),
+            format!("only async fn is supported, {}", input.sig.ident),
         )
         .to_compile_error()
         .into();

actix-rt/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.0.0] - 2019-12-11
+
+* Update dependencies
+
 ## [1.0.0-alpha.3] - 2019-12-07
 
 ### Fixed

actix-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-rt"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,17 +9,15 @@ repository = "https://github.com/actix/actix-net.git"
 documentation = "https://docs.rs/actix-rt/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
-exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
-workspace = ".."
 
 [lib]
 name = "actix_rt"
 path = "src/lib.rs"
 
 [dependencies]
-actix-macros = "0.1.0-alpha.1"
+actix-macros = "0.1.0"
 actix-threadpool = "0.3"
 futures = "0.3.1"
 copyless = "0.1.4"
-tokio = { version = "0.2.4", default-features=false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }
+tokio = { version = "0.2.6", default-features=false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }

actix-server/CHANGES.md

@@ -1,5 +1,18 @@
 # Changes
 
+## [1.0.1] - 2019-12-29
+
+### Changed
+
+* Rename `.start()` method to `.run()`
+
+## [1.0.0] - 2019-12-11
+
+### Changed
+
+* Use actix-net releases
+
+
 ## [1.0.0-alpha.4] - 2019-12-08
 
 ### Changed

actix-server/Cargo.toml

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

actix-server/src/builder.rs

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

actix-server/src/config.rs

@@ -153,7 +153,7 @@ impl InternalServiceFactory for ConfiguredService {
             }
             return Ok(res);
         }
-            .boxed_local()
+        .boxed_local()
     }
 }
 
@@ -280,6 +280,6 @@ where
                 }
             };
         }
-            .boxed_local()
+        .boxed_local()
     }
 }

actix-server/src/worker.rs

@@ -217,7 +217,7 @@ impl Worker {
                     wrk.await
                 });
             }
-                .boxed(),
+            .boxed(),
         );
 
         WorkerClient::new(idx, tx1, tx2, avail)

actix-service/CHANGES.md

@@ -1,5 +1,12 @@
 # Changes
 
+## [1.0.1] - 2019-12-22
+
+### Changed
+
+* `map_config()` and `unit_config()` accepts `IntoServiceFactory` type
+
+
 ## [1.0.0] - 2019-12-11
 
 ### Added

actix-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-service"
-version = "1.0.0"
+version = "1.0.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix service"
 keywords = ["network", "framework", "async", "futures"]
@@ -26,4 +26,4 @@ futures-util = "0.3.1"
 pin-project = "0.4.6"
 
 [dev-dependencies]
-actix-rt = "1.0.0-alpha.3"
+actix-rt = "1.0.0"

actix-service/src/map_config.rs

@@ -1,28 +1,30 @@
 use std::marker::PhantomData;
 
-use super::ServiceFactory;
+use super::{IntoServiceFactory, ServiceFactory};
 
 /// Adapt external config argument to a config for provided service factory
 ///
 /// Note that this function consumes the receiving service factory and returns
 /// a wrapped version of it.
-pub fn map_config<T, F, C>(factory: T, f: F) -> MapConfig<T, F, C>
+pub fn map_config<T, U, F, C>(factory: U, f: F) -> MapConfig<T, F, C>
 where
     T: ServiceFactory,
+    U: IntoServiceFactory<T>,
     F: Fn(C) -> T::Config,
 {
-    MapConfig::new(factory, f)
+    MapConfig::new(factory.into_factory(), f)
 }
 
 /// Replace config with unit
-pub fn unit_config<T, C>(new_service: T) -> UnitConfig<T, C>
+pub fn unit_config<T, U, C>(factory: U) -> UnitConfig<T, C>
 where
     T: ServiceFactory<Config = ()>,
+    U: IntoServiceFactory<T>,
 {
-    UnitConfig::new(new_service)
+    UnitConfig::new(factory.into_factory())
 }
 
-/// `.map_config()` service combinator
+/// `map_config()` adapter service factory
 pub struct MapConfig<A, F, C> {
     a: A,
     f: F,

actix-testing/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.0.0] - 2019-12-11
+
+* Update actix-server to 1.0.0
+
 ## [1.0.0-alpha.3] - 2019-12-07
 
 * Migrate to tokio 0.2

actix-testing/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-testing"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix testing utils"
 keywords = ["network", "framework", "async", "futures"]
@@ -17,10 +17,10 @@ name = "actix_testing"
 path = "src/lib.rs"
 
 [dependencies]
-actix-rt = "1.0.0-alpha.3"
-actix-macros = "0.1.0-alpha.1"
-actix-server = "1.0.0-alpha.3"
-actix-service = "1.0.0-alpha.3"
+actix-rt = "1.0.0"
+actix-macros = "0.1.0"
+actix-server = "1.0.0"
+actix-service = "1.0.0"
 
 log = "0.4"
 net2 = "0.2"

actix-threadpool/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [0.3.1] - 2019-12-12
+
+### Changed
+
+* Use parking_lot 0.10
+
 ## [0.3.0] - 2019-12-02
 
 ### Changed

actix-threadpool/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-threadpool"
-version = "0.3.0"
+version = "0.3.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix thread pool for sync code"
 keywords = ["actix", "network", "framework", "async", "futures"]
@@ -19,9 +19,9 @@ path = "src/lib.rs"
 
 [dependencies]
 derive_more = "0.99.2"
-futures = "0.3.1"
-parking_lot = "0.9"
-lazy_static = "1.2"
+futures-channel = "0.3.1"
+parking_lot = "0.10"
+lazy_static = "1.3"
 log = "0.4"
 num_cpus = "1.10"
 threadpool = "1.7"

actix-threadpool/src/lib.rs

@@ -6,7 +6,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use derive_more::Display;
-use futures::channel::oneshot;
+use futures_channel::oneshot;
 use parking_lot::Mutex;
 use threadpool::ThreadPool;
 
@@ -79,9 +79,12 @@ impl<I, E: fmt::Debug> Future for CpuFuture<I, E> {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let rx = Pin::new(&mut self.rx);
-        let res = futures::ready!(rx.poll(cx))
-            .map_err(|_| BlockingError::Canceled)
-            .and_then(|res| res.map_err(BlockingError::Error));
+        let res = match rx.poll(cx) {
+            Poll::Pending => return Poll::Pending,
+            Poll::Ready(res) => res
+                .map_err(|_| BlockingError::Canceled)
+                .and_then(|res| res.map_err(BlockingError::Error)),
+        };
         Poll::Ready(res)
     }
 }

actix-tls/CHANGES.md

@@ -1,6 +1,10 @@
 # Changes
 
-[1.0.0-alpha.3] - 2019-12-07
+## [1.0.0] - 2019-12-11
+
+* 1.0.0 release
+
+## [1.0.0-alpha.3] - 2019-12-07
 
 ### Changed
 

actix-tls/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-tls"
-version = "1.0.0-alpha.3"
+version = "1.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix tls services"
 keywords = ["network", "framework", "async", "futures"]
@@ -32,10 +32,10 @@ rustls = ["rust-tls", "webpki", "webpki-roots", "tokio-rustls"]
 nativetls = ["native-tls", "tokio-tls"]
 
 [dependencies]
-actix-service = "1.0.0-alpha.3"
+actix-service = "1.0.0"
 actix-codec = "0.2.0"
-actix-utils = "1.0.0-alpha.3"
-actix-rt = "1.0.0-alpha.3"
+actix-utils = "1.0.0"
+actix-rt = "1.0.0"
 derive_more = "0.99.2"
 either = "1.5.2"
 futures = "0.3.1"
@@ -57,4 +57,4 @@ tokio-tls = { version="0.3", optional = true }
 
 [dev-dependencies]
 bytes = "0.5"
-actix-testing = { version="1.0.0-alpha.3" }
+actix-testing = { version="1.0.0" }

actix-tls/src/openssl.rs

@@ -88,7 +88,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin + 'static> Service for AcceptorService<T>
                 let acc = acc;
                 tokio_openssl::accept(&acc, req).await
             }
-                .boxed_local(),
+            .boxed_local(),
         }
     }
 }

actix-utils/CHANGES.md

@@ -1,5 +1,27 @@
 # Changes
 
+## [1.0.4] - 2019-12-20
+
+* Add methods to check `LocalWaker` registration state.
+
+## [1.0.3] - 2019-12-11
+
+* Revert InOrder service changes
+
+## [1.0.2] - 2019-12-11
+
+* Allow to create `framed::Dispatcher` with custom `mpsc::Receiver`
+
+* Add `oneshot::Sender::is_canceled()` method
+
+## [1.0.1] - 2019-12-11
+
+* Optimize InOrder service
+
+## [1.0.0] - 2019-12-11
+
+* Simplify oneshot and mpsc implementations
+
 ## [1.0.0-alpha.3] - 2019-12-07
 
 * Migrate to tokio 0.2

actix-utils/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-utils"
-version = "1.0.0-alpha.3"
+version = "1.0.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix utils - various actix net related services"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,20 +9,18 @@ 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 = "1.0.0-alpha.3"
-actix-rt = "1.0.0-alpha.3"
+actix-service = "1.0.0"
+actix-rt = "1.0.0"
 actix-codec = "0.2.0"
-bytes = "0.5.2"
-either = "1.5.2"
+bytes = "0.5.3"
+either = "1.5.3"
 futures = "0.3.1"
 pin-project = "0.4.6"
 log = "0.4"

actix-utils/src/cell.rs

@@ -2,16 +2,12 @@
 
 use std::cell::UnsafeCell;
 use std::fmt;
-use std::rc::{Rc, Weak};
+use std::rc::Rc;
 
 pub(crate) struct Cell<T> {
     pub(crate) inner: Rc<UnsafeCell<T>>,
 }
 
-pub(crate) struct WeakCell<T> {
-    inner: Weak<UnsafeCell<T>>,
-}
-
 impl<T> Clone for Cell<T> {
     fn clone(&self) -> Self {
         Self {
@@ -27,39 +23,26 @@ impl<T: fmt::Debug> fmt::Debug for Cell<T> {
 }
 
 impl<T> Cell<T> {
-    pub fn new(inner: T) -> Self {
+    pub(crate) fn new(inner: T) -> Self {
         Self {
             inner: Rc::new(UnsafeCell::new(inner)),
         }
     }
 
-    pub fn downgrade(&self) -> WeakCell<T> {
-        WeakCell {
-            inner: Rc::downgrade(&self.inner),
-        }
+    pub(crate) fn strong_count(&self) -> usize {
+        Rc::strong_count(&self.inner)
     }
 
-    pub fn get_ref(&self) -> &T {
+    pub(crate) fn get_ref(&self) -> &T {
         unsafe { &*self.inner.as_ref().get() }
     }
 
-    pub fn get_mut(&mut self) -> &mut T {
+    pub(crate) fn get_mut(&mut self) -> &mut T {
         unsafe { &mut *self.inner.as_ref().get() }
     }
-}
 
-impl<T> WeakCell<T> {
-    pub fn upgrade(&self) -> Option<Cell<T>> {
-        if let Some(inner) = self.inner.upgrade() {
-            Some(Cell { inner })
-        } else {
-            None
-        }
-    }
-}
-
-impl<T: fmt::Debug> fmt::Debug for WeakCell<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        self.inner.fmt(f)
+    #[allow(clippy::mut_from_ref)]
+    pub(crate) unsafe fn get_mut_unsafe(&self) -> &mut T {
+        &mut *self.inner.as_ref().get()
     }
 }

actix-utils/src/framed.rs

@@ -1,37 +1,33 @@
 //! Framed dispatcher service and related utilities
 #![allow(type_alias_bounds)]
-use std::collections::VecDeque;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::{fmt, mem};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, Service};
-use futures::future::{ready, FutureExt};
-use futures::{Future, Sink, Stream};
+use futures::{Future, FutureExt, Stream};
 use log::debug;
 
-use crate::cell::Cell;
 use crate::mpsc;
-use crate::task::LocalWaker;
 
 type Request<U> = <U as Decoder>::Item;
 type Response<U> = <U as Encoder>::Item;
 
 /// Framed transport errors
-pub enum FramedTransportError<E, U: Encoder + Decoder> {
+pub enum DispatcherError<E, U: Encoder + Decoder> {
     Service(E),
     Encoder(<U as Encoder>::Error),
     Decoder(<U as Decoder>::Error),
 }
 
-impl<E, U: Encoder + Decoder> From<E> for FramedTransportError<E, U> {
+impl<E, U: Encoder + Decoder> From<E> for DispatcherError<E, U> {
     fn from(err: E) -> Self {
-        FramedTransportError::Service(err)
+        DispatcherError::Service(err)
     }
 }
 
-impl<E, U: Encoder + Decoder> fmt::Debug for FramedTransportError<E, U>
+impl<E, U: Encoder + Decoder> fmt::Debug for DispatcherError<E, U>
 where
     E: fmt::Debug,
     <U as Encoder>::Error: fmt::Debug,
@@ -39,20 +35,14 @@ where
 {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
-            FramedTransportError::Service(ref e) => {
-                write!(fmt, "FramedTransportError::Service({:?})", e)
-            }
-            FramedTransportError::Encoder(ref e) => {
-                write!(fmt, "FramedTransportError::Encoder({:?})", e)
-            }
-            FramedTransportError::Decoder(ref e) => {
-                write!(fmt, "FramedTransportError::Encoder({:?})", e)
-            }
+            DispatcherError::Service(ref e) => write!(fmt, "DispatcherError::Service({:?})", e),
+            DispatcherError::Encoder(ref e) => write!(fmt, "DispatcherError::Encoder({:?})", e),
+            DispatcherError::Decoder(ref e) => write!(fmt, "DispatcherError::Decoder({:?})", e),
         }
     }
 }
 
-impl<E, U: Encoder + Decoder> fmt::Display for FramedTransportError<E, U>
+impl<E, U: Encoder + Decoder> fmt::Display for DispatcherError<E, U>
 where
     E: fmt::Display,
     <U as Encoder>::Error: fmt::Debug,
@@ -60,25 +50,22 @@ where
 {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
-            FramedTransportError::Service(ref e) => write!(fmt, "{}", e),
-            FramedTransportError::Encoder(ref e) => write!(fmt, "{:?}", e),
-            FramedTransportError::Decoder(ref e) => write!(fmt, "{:?}", e),
+            DispatcherError::Service(ref e) => write!(fmt, "{}", e),
+            DispatcherError::Encoder(ref e) => write!(fmt, "{:?}", e),
+            DispatcherError::Decoder(ref e) => write!(fmt, "{:?}", e),
         }
     }
 }
 
-pub enum FramedMessage<T> {
-    Message(T),
+pub enum Message<T> {
+    Item(T),
     Close,
 }
 
-type Rx<U> = Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>;
-type Inner<S: Service, U> = Cell<FramedTransportInner<<U as Encoder>::Item, S::Error>>;
-
 /// FramedTransport - is a future that reads frames from Framed object
 /// and pass then to the service.
 #[pin_project::pin_project]
-pub struct FramedTransport<S, T, U>
+pub struct Dispatcher<S, T, U>
 where
     S: Service<Request = Request<U>, Response = Response<U>>,
     S::Error: 'static,
@@ -89,26 +76,37 @@ where
     <U as Encoder>::Error: std::fmt::Debug,
 {
     service: S,
-    state: TransportState<S, U>,
+    state: State<S, U>,
     framed: Framed<T, U>,
-    rx: Option<mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>>,
-    inner: Cell<FramedTransportInner<<U as Encoder>::Item, S::Error>>,
+    rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
+    tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
 }
 
-enum TransportState<S: Service, U: Encoder + Decoder> {
+enum State<S: Service, U: Encoder + Decoder> {
     Processing,
-    Error(FramedTransportError<S::Error, U>),
-    FramedError(FramedTransportError<S::Error, U>),
+    Error(DispatcherError<S::Error, U>),
+    FramedError(DispatcherError<S::Error, U>),
     FlushAndStop,
     Stopping,
 }
 
-struct FramedTransportInner<I, E> {
-    buf: VecDeque<Result<I, E>>,
-    task: LocalWaker,
+impl<S: Service, U: Encoder + Decoder> State<S, U> {
+    fn take_error(&mut self) -> DispatcherError<S::Error, U> {
+        match mem::replace(self, State::Processing) {
+            State::Error(err) => err,
+            _ => panic!(),
+        }
+    }
+
+    fn take_framed_error(&mut self) -> DispatcherError<S::Error, U> {
+        match mem::replace(self, State::Processing) {
+            State::FramedError(err) => err,
+            _ => panic!(),
+        }
+    }
 }
 
-impl<S, T, U> FramedTransport<S, T, U>
+impl<S, T, U> Dispatcher<S, T, U>
 where
     S: Service<Request = Request<U>, Response = Response<U>>,
     S::Error: 'static,
@@ -119,52 +117,150 @@ where
     <U as Encoder>::Error: std::fmt::Debug,
 {
     pub fn new<F: IntoService<S>>(framed: Framed<T, U>, service: F) -> Self {
-        FramedTransport {
+        let (tx, rx) = mpsc::channel();
+        Dispatcher {
             framed,
-            rx: None,
+            rx,
+            tx,
             service: service.into_service(),
-            state: TransportState::Processing,
-            inner: Cell::new(FramedTransportInner {
-                buf: VecDeque::new(),
-                task: LocalWaker::new(),
-            }),
+            state: State::Processing,
         }
     }
 
-    /// Get Sender
-    pub fn set_receiver(
-        mut self,
-        rx: mpsc::Receiver<FramedMessage<<U as Encoder>::Item>>,
+    /// Construct new `Dispatcher` instance with customer `mpsc::Receiver`
+    pub fn with_rx<F: IntoService<S>>(
+        framed: Framed<T, U>,
+        service: F,
+        rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
     ) -> Self {
-        self.rx = Some(rx);
-        self
+        let tx = rx.sender();
+        Dispatcher {
+            framed,
+            rx,
+            tx,
+            service: service.into_service(),
+            state: State::Processing,
+        }
     }
 
-    /// Get reference to a service wrapped by `FramedTransport` instance.
+    /// Get sink
+    pub fn get_sink(&self) -> mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>> {
+        self.tx.clone()
+    }
+
+    /// Get reference to a service wrapped by `Dispatcher` instance.
     pub fn get_ref(&self) -> &S {
         &self.service
     }
 
-    /// Get mutable reference to a service wrapped by `FramedTransport`
-    /// instance.
+    /// Get mutable reference to a service wrapped by `Dispatcher` instance.
     pub fn get_mut(&mut self) -> &mut S {
         &mut self.service
     }
 
-    /// Get reference to a framed instance wrapped by `FramedTransport`
+    /// Get reference to a framed instance wrapped by `Dispatcher`
     /// instance.
     pub fn get_framed(&self) -> &Framed<T, U> {
         &self.framed
     }
 
-    /// Get mutable reference to a framed instance wrapped by `FramedTransport`
-    /// instance.
+    /// Get mutable reference to a framed instance wrapped by `Dispatcher` instance.
     pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
         &mut self.framed
     }
+
+    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool
+    where
+        S: Service<Request = Request<U>, Response = Response<U>>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder,
+        <U as Encoder>::Item: 'static,
+        <U as Encoder>::Error: std::fmt::Debug,
+    {
+        loop {
+            match self.service.poll_ready(cx) {
+                Poll::Ready(Ok(_)) => {
+                    let item = match self.framed.next_item(cx) {
+                        Poll::Ready(Some(Ok(el))) => el,
+                        Poll::Ready(Some(Err(err))) => {
+                            self.state = State::FramedError(DispatcherError::Decoder(err));
+                            return true;
+                        }
+                        Poll::Pending => return false,
+                        Poll::Ready(None) => {
+                            self.state = State::Stopping;
+                            return true;
+                        }
+                    };
+
+                    let tx = self.tx.clone();
+                    actix_rt::spawn(self.service.call(item).map(move |item| {
+                        let _ = tx.send(item.map(Message::Item));
+                    }));
+                }
+                Poll::Pending => return false,
+                Poll::Ready(Err(err)) => {
+                    self.state = State::Error(DispatcherError::Service(err));
+                    return true;
+                }
+            }
+        }
+    }
+
+    /// write to framed object
+    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool
+    where
+        S: Service<Request = Request<U>, Response = Response<U>>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder,
+        <U as Encoder>::Item: 'static,
+        <U as Encoder>::Error: std::fmt::Debug,
+    {
+        loop {
+            while !self.framed.is_write_buf_full() {
+                match Pin::new(&mut self.rx).poll_next(cx) {
+                    Poll::Ready(Some(Ok(Message::Item(msg)))) => {
+                        if let Err(err) = self.framed.write(msg) {
+                            self.state = State::FramedError(DispatcherError::Encoder(err));
+                            return true;
+                        }
+                    }
+                    Poll::Ready(Some(Ok(Message::Close))) => {
+                        self.state = State::FlushAndStop;
+                        return true;
+                    }
+                    Poll::Ready(Some(Err(err))) => {
+                        self.state = State::Error(DispatcherError::Service(err));
+                        return true;
+                    }
+                    Poll::Ready(None) | Poll::Pending => break,
+                }
+            }
+
+            if !self.framed.is_write_buf_empty() {
+                match self.framed.flush(cx) {
+                    Poll::Pending => break,
+                    Poll::Ready(Ok(_)) => (),
+                    Poll::Ready(Err(err)) => {
+                        debug!("Error sending data: {:?}", err);
+                        self.state = State::FramedError(DispatcherError::Encoder(err));
+                        return true;
+                    }
+                }
+            } else {
+                break;
+            }
+        }
+
+        false
+    }
 }
 
-impl<S, T, U> Future for FramedTransport<S, T, U>
+impl<S, T, U> Future for Dispatcher<S, T, U>
 where
     S: Service<Request = Request<U>, Response = Response<U>>,
     S::Error: 'static,
@@ -175,210 +271,46 @@ where
     <U as Encoder>::Error: std::fmt::Debug,
     <U as Decoder>::Error: std::fmt::Debug,
 {
-    type Output = Result<(), FramedTransportError<S::Error, U>>;
+    type Output = Result<(), DispatcherError<S::Error, U>>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.inner.get_ref().task.register(cx.waker());
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        loop {
+            let this = self.as_mut().project();
 
-        let this = self.project();
-        poll(
-            cx,
-            this.service,
-            this.state,
-            this.framed,
-            this.rx,
-            this.inner,
-        )
-    }
-}
-
-fn poll<S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut TransportState<S, U>,
-    framed: &mut Framed<T, U>,
-    rx: &mut Rx<U>,
-    inner: &mut Inner<S, U>,
-) -> Poll<Result<(), FramedTransportError<S::Error, U>>>
-where
-    S: Service<Request = Request<U>, Response = Response<U>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    match mem::replace(state, TransportState::Processing) {
-        TransportState::Processing => {
-            if poll_read(cx, srv, state, framed, inner)
-                || poll_write(cx, state, framed, rx, inner)
-            {
-                poll(cx, srv, state, framed, rx, inner)
-            } else {
-                Poll::Pending
-            }
-        }
-        TransportState::Error(err) => {
-            let is_empty = framed.is_write_buf_empty();
-            if is_empty || poll_write(cx, state, framed, rx, inner) {
-                Poll::Ready(Err(err))
-            } else {
-                *state = TransportState::Error(err);
-                Poll::Pending
-            }
-        }
-        TransportState::FlushAndStop => {
-            if !framed.is_write_buf_empty() {
-                match Pin::new(framed).poll_flush(cx) {
-                    Poll::Ready(Err(err)) => {
-                        debug!("Error sending data: {:?}", err);
+            return match this.state {
+                State::Processing => {
+                    if self.poll_read(cx) || self.poll_write(cx) {
+                        continue;
+                    } else {
+                        Poll::Pending
+                    }
+                }
+                State::Error(_) => {
+                    // flush write buffer
+                    if !self.framed.is_write_buf_empty() {
+                        if let Poll::Pending = self.framed.flush(cx) {
+                            return Poll::Pending;
+                        }
+                    }
+                    Poll::Ready(Err(self.state.take_error()))
+                }
+                State::FlushAndStop => {
+                    if !this.framed.is_write_buf_empty() {
+                        match this.framed.flush(cx) {
+                            Poll::Ready(Err(err)) => {
+                                debug!("Error sending data: {:?}", err);
+                                Poll::Ready(Ok(()))
+                            }
+                            Poll::Pending => Poll::Pending,
+                            Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
+                        }
+                    } else {
                         Poll::Ready(Ok(()))
                     }
-                    Poll::Pending => Poll::Pending,
-                    Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
                 }
-            } else {
-                Poll::Ready(Ok(()))
-            }
-        }
-        TransportState::FramedError(err) => Poll::Ready(Err(err)),
-        TransportState::Stopping => Poll::Ready(Ok(())),
-    }
-}
-
-fn poll_read<S, T, U>(
-    cx: &mut Context<'_>,
-    srv: &mut S,
-    state: &mut TransportState<S, U>,
-    framed: &mut Framed<T, U>,
-    inner: &mut Inner<S, U>,
-) -> bool
-where
-    S: Service<Request = Request<U>, Response = Response<U>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    loop {
-        match srv.poll_ready(cx) {
-            Poll::Ready(Ok(_)) => {
-                let item = match framed.next_item(cx) {
-                    Poll::Ready(Some(Ok(el))) => el,
-                    Poll::Ready(Some(Err(err))) => {
-                        *state =
-                            TransportState::FramedError(FramedTransportError::Decoder(err));
-                        return true;
-                    }
-                    Poll::Pending => return false,
-                    Poll::Ready(None) => {
-                        *state = TransportState::Stopping;
-                        return true;
-                    }
-                };
-
-                let mut cell = inner.clone();
-                let fut = srv.call(item).then(move |item| {
-                    let inner = cell.get_mut();
-                    inner.buf.push_back(item);
-                    inner.task.wake();
-                    ready(())
-                });
-                actix_rt::spawn(fut);
-            }
-            Poll::Pending => return false,
-            Poll::Ready(Err(err)) => {
-                *state = TransportState::Error(FramedTransportError::Service(err));
-                return true;
-            }
+                State::FramedError(_) => Poll::Ready(Err(this.state.take_framed_error())),
+                State::Stopping => Poll::Ready(Ok(())),
+            };
         }
     }
 }
-
-/// write to framed object
-fn poll_write<S, T, U>(
-    cx: &mut Context<'_>,
-    state: &mut TransportState<S, U>,
-    framed: &mut Framed<T, U>,
-    rx: &mut Rx<U>,
-    inner: &mut Inner<S, U>,
-) -> bool
-where
-    S: Service<Request = Request<U>, Response = Response<U>>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder,
-    <U as Encoder>::Item: 'static,
-    <U as Encoder>::Error: std::fmt::Debug,
-{
-    // let this = self.project();
-
-    let inner = inner.get_mut();
-    let mut rx_done = rx.is_none();
-    let mut buf_empty = inner.buf.is_empty();
-    loop {
-        while !framed.is_write_buf_full() {
-            if !buf_empty {
-                match inner.buf.pop_front().unwrap() {
-                    Ok(msg) => {
-                        if let Err(err) = framed.write(msg) {
-                            *state =
-                                TransportState::FramedError(FramedTransportError::Encoder(err));
-                            return true;
-                        }
-                        buf_empty = inner.buf.is_empty();
-                    }
-                    Err(err) => {
-                        *state = TransportState::Error(FramedTransportError::Service(err));
-                        return true;
-                    }
-                }
-            }
-
-            if !rx_done && rx.is_some() {
-                match Pin::new(rx.as_mut().unwrap()).poll_next(cx) {
-                    Poll::Ready(Some(FramedMessage::Message(msg))) => {
-                        if let Err(err) = framed.write(msg) {
-                            *state =
-                                TransportState::FramedError(FramedTransportError::Encoder(err));
-                            return true;
-                        }
-                    }
-                    Poll::Ready(Some(FramedMessage::Close)) => {
-                        *state = TransportState::FlushAndStop;
-                        return true;
-                    }
-                    Poll::Ready(None) => {
-                        rx_done = true;
-                        let _ = rx.take();
-                    }
-                    Poll::Pending => rx_done = true,
-                }
-            }
-
-            if rx_done && buf_empty {
-                break;
-            }
-        }
-
-        if !framed.is_write_buf_empty() {
-            match framed.flush(cx) {
-                Poll::Pending => break,
-                Poll::Ready(Err(err)) => {
-                    debug!("Error sending data: {:?}", err);
-                    *state = TransportState::FramedError(FramedTransportError::Encoder(err));
-                    return true;
-                }
-                Poll::Ready(Ok(_)) => (),
-            }
-        } else {
-            break;
-        }
-    }
-
-    false
-}

actix-utils/src/mpsc.rs

@@ -1,34 +1,27 @@
-//! A multi-producer, single-consumer, futures-aware, FIFO queue with back
-//! pressure, for use communicating between tasks on the same thread.
-//!
-//! These queues are the same as those in `futures::sync`, except they're not
-//! intended to be sent across threads.
-
+//! A multi-producer, single-consumer, futures-aware, FIFO queue.
 use std::any::Any;
-use std::cell::RefCell;
 use std::collections::VecDeque;
 use std::error::Error;
+use std::fmt;
 use std::pin::Pin;
-use std::rc::{Rc, Weak};
 use std::task::{Context, Poll};
-use std::{fmt, mem};
 
 use futures::{Sink, Stream};
 
+use crate::cell::Cell;
 use crate::task::LocalWaker;
 
 /// Creates a unbounded in-memory channel with buffered storage.
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
-    let shared = Rc::new(RefCell::new(Shared {
+    let shared = Cell::new(Shared {
+        has_receiver: true,
         buffer: VecDeque::new(),
         blocked_recv: LocalWaker::new(),
-    }));
+    });
     let sender = Sender {
-        shared: Rc::downgrade(&shared),
-    };
-    let receiver = Receiver {
-        state: State::Open(shared),
+        shared: shared.clone(),
     };
+    let receiver = Receiver { shared };
     (sender, receiver)
 }
 
@@ -36,6 +29,7 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
 struct Shared<T> {
     buffer: VecDeque<T>,
     blocked_recv: LocalWaker,
+    has_receiver: bool,
 }
 
 /// The transmission end of a channel.
@@ -43,22 +37,30 @@ struct Shared<T> {
 /// This is created by the `channel` function.
 #[derive(Debug)]
 pub struct Sender<T> {
-    shared: Weak<RefCell<Shared<T>>>,
+    shared: Cell<Shared<T>>,
 }
 
+impl<T> Unpin for Sender<T> {}
+
 impl<T> Sender<T> {
     /// Sends the provided message along this channel.
     pub fn send(&self, item: T) -> Result<(), SendError<T>> {
-        let shared = match self.shared.upgrade() {
-            Some(shared) => shared,
-            None => return Err(SendError(item)), // receiver was dropped
+        let shared = unsafe { self.shared.get_mut_unsafe() };
+        if !shared.has_receiver {
+            return Err(SendError(item)); // receiver was dropped
         };
-        let mut shared = shared.borrow_mut();
-
         shared.buffer.push_back(item);
         shared.blocked_recv.wake();
         Ok(())
     }
+
+    /// Closes the sender half
+    ///
+    /// This prevents any further messages from being sent on the channel while
+    /// still enabling the receiver to drain messages that are buffered.
+    pub fn close(&mut self) {
+        self.shared.get_mut().has_receiver = false;
+    }
 }
 
 impl<T> Clone for Sender<T> {
@@ -91,17 +93,13 @@ impl<T> Sink<T> for Sender<T> {
 
 impl<T> Drop for Sender<T> {
     fn drop(&mut self) {
-        let shared = match self.shared.upgrade() {
-            Some(shared) => shared,
-            None => return,
-        };
-        // The number of existing `Weak` indicates if we are possibly the last
-        // `Sender`. If we are the last, we possibly must notify a blocked
-        // `Receiver`. `self.shared` is always one of the `Weak` to this shared
-        // data. Therefore the smallest possible Rc::weak_count(&shared) is 1.
-        if Rc::weak_count(&shared) == 1 {
+        let count = self.shared.strong_count();
+        let shared = self.shared.get_mut();
+
+        // check is last sender is about to drop
+        if shared.has_receiver && count == 2 {
             // Wake up receiver as its stream has ended
-            shared.borrow_mut().blocked_recv.wake();
+            shared.blocked_recv.wake();
         }
     }
 }
@@ -111,56 +109,32 @@ impl<T> Drop for Sender<T> {
 /// This is created by the `channel` function.
 #[derive(Debug)]
 pub struct Receiver<T> {
-    state: State<T>,
-}
-
-impl<T> Unpin for Receiver<T> {}
-
-/// Possible states of a receiver. We're either Open (can receive more messages)
-/// or we're closed with a list of messages we have left to receive.
-#[derive(Debug)]
-enum State<T> {
-    Open(Rc<RefCell<Shared<T>>>),
-    Closed(VecDeque<T>),
+    shared: Cell<Shared<T>>,
 }
 
 impl<T> Receiver<T> {
-    /// Closes the receiving half
-    ///
-    /// This prevents any further messages from being sent on the channel while
-    /// still enabling the receiver to drain messages that are buffered.
-    pub fn close(&mut self) {
-        let items = match self.state {
-            State::Open(ref state) => {
-                let mut state = state.borrow_mut();
-                mem::replace(&mut state.buffer, VecDeque::new())
-            }
-            State::Closed(_) => return,
-        };
-        self.state = State::Closed(items);
+    /// Create Sender
+    pub fn sender(&self) -> Sender<T> {
+        Sender {
+            shared: self.shared.clone(),
+        }
     }
 }
 
+impl<T> Unpin for Receiver<T> {}
+
 impl<T> Stream for Receiver<T> {
     type Item = T;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let me = match self.state {
-            State::Open(ref mut me) => me,
-            State::Closed(ref mut items) => return Poll::Ready(items.pop_front()),
-        };
-
-        if let Some(shared) = Rc::get_mut(me) {
+        if self.shared.strong_count() == 1 {
             // All senders have been dropped, so drain the buffer and end the
             // stream.
-            return Poll::Ready(shared.borrow_mut().buffer.pop_front());
-        }
-
-        let mut shared = me.borrow_mut();
-        if let Some(msg) = shared.buffer.pop_front() {
+            Poll::Ready(self.shared.get_mut().buffer.pop_front())
+        } else if let Some(msg) = self.shared.get_mut().buffer.pop_front() {
             Poll::Ready(Some(msg))
         } else {
-            shared.blocked_recv.register(cx.waker());
+            self.shared.get_mut().blocked_recv.register(cx.waker());
             Poll::Pending
         }
     }
@@ -168,7 +142,9 @@ impl<T> Stream for Receiver<T> {
 
 impl<T> Drop for Receiver<T> {
     fn drop(&mut self) {
-        self.close();
+        let shared = self.shared.get_mut();
+        shared.buffer.clear();
+        shared.has_receiver = false;
     }
 }
 
@@ -200,3 +176,44 @@ impl<T> SendError<T> {
         self.0
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::lazy;
+    use futures::{Stream, StreamExt};
+
+    #[actix_rt::test]
+    async fn test_mpsc() {
+        let (tx, mut rx) = channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.next().await.unwrap(), "test");
+
+        let tx2 = tx.clone();
+        tx2.send("test2").unwrap();
+        assert_eq!(rx.next().await.unwrap(), "test2");
+
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut rx).poll_next(cx)).await,
+            Poll::Pending
+        );
+        drop(tx2);
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut rx).poll_next(cx)).await,
+            Poll::Pending
+        );
+        drop(tx);
+        assert_eq!(rx.next().await, None);
+
+        let (tx, rx) = channel();
+        tx.send("test").unwrap();
+        drop(rx);
+        assert!(tx.send("test").is_err());
+
+        let (mut tx, _) = channel();
+        let tx2 = tx.clone();
+        tx.close();
+        assert!(tx.send("test").is_err());
+        assert!(tx2.send("test").is_err());
+    }
+}

actix-utils/src/oneshot.rs

@@ -1,69 +1,45 @@
-//! A one-shot, futures-aware channel
-//!
-//! This channel is similar to that in `sync::oneshot` but cannot be sent across
-//! threads.
-
+//! A one-shot, futures-aware channel.
 use std::future::Future;
 use std::pin::Pin;
-use std::rc::Rc;
 use std::task::{Context, Poll};
 
 pub use futures::channel::oneshot::Canceled;
 
-use crate::cell::{Cell, WeakCell};
+use crate::cell::Cell;
 use crate::task::LocalWaker;
 
 /// Creates a new futures-aware, one-shot channel.
-///
-/// This function is the same as `sync::oneshot::channel` except that the
-/// returned values cannot be sent across threads.
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     let inner = Cell::new(Inner {
         value: None,
         rx_task: LocalWaker::new(),
     });
     let tx = Sender {
-        inner: inner.downgrade(),
-    };
-    let rx = Receiver {
-        state: State::Open(inner),
+        inner: inner.clone(),
     };
+    let rx = Receiver { inner };
     (tx, rx)
 }
 
 /// Represents the completion half of a oneshot through which the result of a
 /// computation is signaled.
-///
-/// This is created by the `unsync::oneshot::channel` function and is equivalent
-/// in functionality to `sync::oneshot::Sender` except that it cannot be sent
-/// across threads.
 #[derive(Debug)]
 pub struct Sender<T> {
-    inner: WeakCell<Inner<T>>,
+    inner: Cell<Inner<T>>,
 }
 
 /// A future representing the completion of a computation happening elsewhere in
 /// memory.
-///
-/// This is created by the `unsync::oneshot::channel` function and is equivalent
-/// in functionality to `sync::oneshot::Receiver` except that it cannot be sent
-/// across threads.
 #[derive(Debug)]
 #[must_use = "futures do nothing unless polled"]
 pub struct Receiver<T> {
-    state: State<T>,
+    inner: Cell<Inner<T>>,
 }
 
 // The channels do not ever project Pin to the inner T
 impl<T> Unpin for Receiver<T> {}
 impl<T> Unpin for Sender<T> {}
 
-#[derive(Debug)]
-enum State<T> {
-    Open(Cell<Inner<T>>),
-    Closed(Option<T>),
-}
-
 #[derive(Debug)]
 struct Inner<T> {
     value: Option<T>,
@@ -78,12 +54,12 @@ impl<T> Sender<T> {
     /// represents.
     ///
     /// If the value is successfully enqueued for the remote end to receive,
-    /// then `Ok(())` is returned. If the receiving end was deallocated before
+    /// then `Ok(())` is returned. If the receiving end was dropped before
     /// this function was called, however, then `Err` is returned with the value
     /// provided.
-    pub fn send(self, val: T) -> Result<(), T> {
-        if let Some(mut inner) = self.inner.upgrade() {
-            let inner = inner.get_mut();
+    pub fn send(mut self, val: T) -> Result<(), T> {
+        if self.inner.strong_count() == 2 {
+            let inner = self.inner.get_mut();
             inner.value = Some(val);
             inner.rx_task.wake();
             Ok(())
@@ -94,44 +70,15 @@ impl<T> Sender<T> {
 
     /// Tests to see whether this `Sender`'s corresponding `Receiver`
     /// has gone away.
-    ///
-    /// This function can be used to learn about when the `Receiver` (consumer)
-    /// half has gone away and nothing will be able to receive a message sent
-    /// from `send`.
-    ///
-    /// Note that this function is intended to *not* be used in the context of a
-    /// future. If you're implementing a future you probably want to call the
-    /// `poll_cancel` function which will block the current task if the
-    /// cancellation hasn't happened yet. This can be useful when working on a
-    /// non-futures related thread, though, which would otherwise panic if
-    /// `poll_cancel` were called.
     pub fn is_canceled(&self) -> bool {
-        self.inner.upgrade().is_none()
+        self.inner.strong_count() == 1
     }
 }
 
 impl<T> Drop for Sender<T> {
     fn drop(&mut self) {
-        if let Some(inner) = self.inner.upgrade() {
-            inner.get_ref().rx_task.wake();
-        };
-    }
-}
-
-impl<T> Receiver<T> {
-    /// Gracefully close this receiver, preventing sending any future messages.
-    ///
-    /// Any `send` operation which happens after this method returns is
-    /// guaranteed to fail. Once this method is called the normal `poll` method
-    /// can be used to determine whether a message was actually sent or not. If
-    /// `Canceled` is returned from `poll` then no message was sent.
-    pub fn close(&mut self) {
-        match self.state {
-            State::Open(ref mut inner) => {
-                let value = inner.get_mut().value.take();
-                self.state = State::Closed(value);
-            }
-            State::Closed(_) => {}
+        if self.inner.strong_count() == 2 {
+            self.inner.get_ref().rx_task.wake();
         };
     }
 }
@@ -142,33 +89,50 @@ impl<T> Future for Receiver<T> {
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.get_mut();
 
-        let inner = match this.state {
-            State::Open(ref mut inner) => inner,
-            State::Closed(ref mut item) => match item.take() {
-                Some(item) => return Poll::Ready(Ok(item)),
-                None => return Poll::Ready(Err(Canceled)),
-            },
-        };
-
         // If we've got a value, then skip the logic below as we're done.
-        if let Some(val) = inner.get_mut().value.take() {
+        if let Some(val) = this.inner.get_mut().value.take() {
             return Poll::Ready(Ok(val));
         }
 
-        // If we can get mutable access, then the sender has gone away. We
-        // didn't see a value above, so we're canceled. Otherwise we park
-        // our task and wait for a value to come in.
-        if Rc::get_mut(&mut inner.inner).is_some() {
+        // Check if sender is dropped and return error if it is.
+        if this.inner.strong_count() == 1 {
             Poll::Ready(Err(Canceled))
         } else {
-            inner.get_ref().rx_task.register(cx.waker());
+            this.inner.get_ref().rx_task.register(cx.waker());
             Poll::Pending
         }
     }
 }
 
-impl<T> Drop for Receiver<T> {
-    fn drop(&mut self) {
-        self.close();
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::lazy;
+
+    #[actix_rt::test]
+    async fn test_oneshot() {
+        let (tx, rx) = channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.await.unwrap(), "test");
+
+        let (tx, rx) = channel();
+        assert!(!tx.is_canceled());
+        drop(rx);
+        assert!(tx.is_canceled());
+        assert!(tx.send("test").is_err());
+
+        let (tx, rx) = channel::<&'static str>();
+        drop(tx);
+        assert!(rx.await.is_err());
+
+        let (tx, mut rx) = channel::<&'static str>();
+        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
+        tx.send("test").unwrap();
+        assert_eq!(rx.await.unwrap(), "test");
+
+        let (tx, mut rx) = channel::<&'static str>();
+        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
+        drop(tx);
+        assert!(rx.await.is_err());
     }
 }

actix-utils/src/order.rs

@@ -105,7 +105,7 @@ where
 
 pub struct InOrderService<S: Service> {
     service: S,
-    task: Rc<LocalWaker>,
+    waker: Rc<LocalWaker>,
     acks: VecDeque<Record<S::Response, S::Error>>,
 }
 
@@ -123,7 +123,7 @@ where
         Self {
             service: service.into_service(),
             acks: VecDeque::new(),
-            task: Rc::new(LocalWaker::new()),
+            waker: Rc::new(LocalWaker::new()),
         }
     }
 }
@@ -142,7 +142,7 @@ where
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         // poll_ready could be called from different task
-        self.task.register(cx.waker());
+        self.waker.register(cx.waker());
 
         // check acks
         while !self.acks.is_empty() {
@@ -172,11 +172,11 @@ where
         let (tx2, rx2) = oneshot::channel();
         self.acks.push_back(Record { rx: rx1, tx: tx2 });
 
-        let task = self.task.clone();
+        let waker = self.waker.clone();
         let fut = self.service.call(request);
         actix_rt::spawn(async move {
             let res = fut.await;
-            task.wake();
+            waker.wake();
             let _ = tx1.send(res);
         });
 
@@ -211,7 +211,7 @@ mod tests {
     use super::*;
     use actix_service::Service;
     use futures::channel::oneshot;
-    use futures::future::{lazy, FutureExt, LocalBoxFuture};
+    use futures::future::{lazy, poll_fn, FutureExt, LocalBoxFuture};
 
     struct Srv;
 
@@ -245,15 +245,18 @@ mod tests {
             let _ = actix_rt::System::new("test").block_on(async {
                 let mut srv = InOrderService::new(Srv);
 
+                let _ = lazy(|cx| srv.poll_ready(cx)).await;
                 let res1 = srv.call(rx1);
                 let res2 = srv.call(rx2);
                 let res3 = srv.call(rx3);
 
-                let _ = lazy(|cx| srv.poll_ready(cx)).await;
-
-                // dispatcher do this
-                actix_rt::time::delay_for(Duration::from_millis(100)).await;
-                let _ = lazy(|cx| srv.poll_ready(cx)).await;
+                actix_rt::spawn(async move {
+                    let _ = poll_fn(|cx| {
+                        let _ = srv.poll_ready(cx);
+                        Poll::<()>::Pending
+                    })
+                    .await;
+                });
 
                 assert_eq!(res1.await.unwrap(), 1);
                 assert_eq!(res2.await.unwrap(), 2);

actix-utils/src/stream.rs

@@ -3,12 +3,12 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_service::{IntoService, Service};
-use futures::Stream;
+use futures::{FutureExt, Stream};
 
 use crate::mpsc;
 
 #[pin_project::pin_project]
-pub struct StreamDispatcher<S, T>
+pub struct Dispatcher<S, T>
 where
     S: Stream,
     T: Service<Request = S::Item, Response = ()> + 'static,
@@ -20,7 +20,7 @@ where
     err_tx: mpsc::Sender<T::Error>,
 }
 
-impl<S, T> StreamDispatcher<S, T>
+impl<S, T> Dispatcher<S, T>
 where
     S: Stream,
     T: Service<Request = S::Item, Response = ()> + 'static,
@@ -30,7 +30,7 @@ where
         F: IntoService<T>,
     {
         let (err_tx, err_rx) = mpsc::channel();
-        StreamDispatcher {
+        Dispatcher {
             err_rx,
             err_tx,
             stream,
@@ -39,7 +39,7 @@ where
     }
 }
 
-impl<S, T> Future for StreamDispatcher<S, T>
+impl<S, T> Future for Dispatcher<S, T>
 where
     S: Stream,
     T: Service<Request = S::Item, Response = ()> + 'static,
@@ -54,47 +54,23 @@ where
         }
 
         loop {
-            match this.service.poll_ready(cx)? {
+            return match this.service.poll_ready(cx)? {
                 Poll::Ready(_) => match this.stream.poll_next(cx) {
                     Poll::Ready(Some(item)) => {
-                        actix_rt::spawn(StreamDispatcherService {
-                            fut: this.service.call(item),
-                            stop: self.err_tx.clone(),
-                        });
+                        let stop = this.err_tx.clone();
+                        actix_rt::spawn(this.service.call(item).map(move |res| {
+                            if let Err(e) = res {
+                                let _ = stop.send(e);
+                            }
+                        }));
                         this = self.as_mut().project();
+                        continue;
                     }
-                    Poll::Pending => return Poll::Pending,
-                    Poll::Ready(None) => return Poll::Ready(Ok(())),
+                    Poll::Pending => Poll::Pending,
+                    Poll::Ready(None) => Poll::Ready(Ok(())),
                 },
-                Poll::Pending => return Poll::Pending,
-            }
-        }
-    }
-}
-
-#[pin_project::pin_project]
-struct StreamDispatcherService<F: Future, E> {
-    #[pin]
-    fut: F,
-    stop: mpsc::Sender<E>,
-}
-
-impl<F, E> Future for StreamDispatcherService<F, E>
-where
-    F: Future<Output = Result<(), E>>,
-{
-    type Output = ();
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.project();
-
-        match this.fut.poll(cx) {
-            Poll::Ready(Ok(_)) => Poll::Ready(()),
-            Poll::Pending => Poll::Pending,
-            Poll::Ready(Err(e)) => {
-                let _ = this.stop.send(e);
-                Poll::Ready(())
-            }
+                Poll::Pending => Poll::Pending,
+            };
         }
     }
 }

actix-utils/src/task.rs

@@ -34,11 +34,22 @@ impl LocalWaker {
         }
     }
 
+    #[inline]
+    /// Check if waker has been registered.
+    pub fn is_registed(&self) -> bool {
+        unsafe { (*self.waker.get()).is_some() }
+    }
+
     #[inline]
     /// Registers the waker to be notified on calls to `wake`.
-    pub fn register(&self, waker: &Waker) {
+    ///
+    /// Returns `true` if waker was registered before.
+    pub fn register(&self, waker: &Waker) -> bool {
         unsafe {
-            *self.waker.get() = Some(waker.clone());
+            let w = self.waker.get();
+            let is_registered = (*w).is_some();
+            *w = Some(waker.clone());
+            is_registered
         }
     }
 

router/CHANGES.txt

@@ -1,5 +1,19 @@
 # Changes
 
+## [0.2.3] - 2019-12-25
+
+* Add impl `IntoPattern` for `&String`
+
+## [0.2.2] - 2019-12-25
+
+* Use `IntoPattern` for `RouterBuilder::path()`
+
+## [0.2.1] - 2019-12-25
+
+* Add `IntoPattern` trait
+
+* Add multi-pattern resources
+
 ## [0.2.0] - 2019-12-07
 
 * Update http to 0.2

router/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-router"
-version = "0.2.0"
+version = "0.2.3"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Path router"
 keywords = ["actix"]
@@ -8,9 +8,7 @@ homepage = "https://actix.rs"
 repository = "https://github.com/actix/actix-net.git"
 documentation = "https://docs.rs/actix-router/"
 license = "MIT/Apache-2.0"
-exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
 edition = "2018"
-workspace = ".."
 
 [lib]
 name = "actix_router"
@@ -21,8 +19,8 @@ default = ["http"]
 
 [dependencies]
 regex = "1.3.1"
-serde = "1.0.80"
-bytestring = "0.1.0"
+serde = "1.0.104"
+bytestring = "0.1.2"
 log = "0.4.8"
 http = { version="0.2.0", optional=true }
 

router/src/de.rs

@@ -492,7 +492,7 @@ mod tests {
 
     #[derive(Deserialize)]
     struct Id {
-        id: String,
+        _id: String,
     }
 
     #[derive(Debug, Deserialize)]

router/src/lib.rs

@@ -35,6 +35,100 @@ impl ResourcePath for bytestring::ByteString {
     }
 }
 
+/// Helper trait for type that could be converted to path pattern
+pub trait IntoPattern {
+    /// Signle patter
+    fn is_single(&self) -> bool;
+
+    fn patterns(&self) -> Vec<String>;
+}
+
+impl IntoPattern for String {
+    fn is_single(&self) -> bool {
+        true
+    }
+
+    fn patterns(&self) -> Vec<String> {
+        vec![self.clone()]
+    }
+}
+
+impl<'a> IntoPattern for &'a String {
+    fn is_single(&self) -> bool {
+        true
+    }
+
+    fn patterns(&self) -> Vec<String> {
+        vec![self.as_str().to_string()]
+    }
+}
+
+impl<'a> IntoPattern for &'a str {
+    fn is_single(&self) -> bool {
+        true
+    }
+
+    fn patterns(&self) -> Vec<String> {
+        vec![self.to_string()]
+    }
+}
+
+impl<T: AsRef<str>> IntoPattern for Vec<T> {
+    fn is_single(&self) -> bool {
+        self.len() == 1
+    }
+
+    fn patterns(&self) -> Vec<String> {
+        self.into_iter().map(|v| v.as_ref().to_string()).collect()
+    }
+}
+
+macro_rules! array_patterns (($tp:ty, $num:tt) => {
+    impl IntoPattern for [$tp; $num] {
+        fn is_single(&self) -> bool {
+            $num == 1
+        }
+
+        fn patterns(&self) -> Vec<String> {
+            self.iter().map(|v| v.to_string()).collect()
+        }
+    }
+});
+
+array_patterns!(&str, 1);
+array_patterns!(&str, 2);
+array_patterns!(&str, 3);
+array_patterns!(&str, 4);
+array_patterns!(&str, 5);
+array_patterns!(&str, 6);
+array_patterns!(&str, 7);
+array_patterns!(&str, 8);
+array_patterns!(&str, 9);
+array_patterns!(&str, 10);
+array_patterns!(&str, 11);
+array_patterns!(&str, 12);
+array_patterns!(&str, 13);
+array_patterns!(&str, 14);
+array_patterns!(&str, 15);
+array_patterns!(&str, 16);
+
+array_patterns!(String, 1);
+array_patterns!(String, 2);
+array_patterns!(String, 3);
+array_patterns!(String, 4);
+array_patterns!(String, 5);
+array_patterns!(String, 6);
+array_patterns!(String, 7);
+array_patterns!(String, 8);
+array_patterns!(String, 9);
+array_patterns!(String, 10);
+array_patterns!(String, 11);
+array_patterns!(String, 12);
+array_patterns!(String, 13);
+array_patterns!(String, 14);
+array_patterns!(String, 15);
+array_patterns!(String, 16);
+
 #[cfg(feature = "http")]
 mod url;
 

router/src/resource.rs

@@ -2,10 +2,10 @@ use std::cmp::min;
 use std::hash::{Hash, Hasher};
 use std::rc::Rc;
 
-use regex::{escape, Regex};
+use regex::{escape, Regex, RegexSet};
 
 use crate::path::{Path, PathItem};
-use crate::{Resource, ResourcePath};
+use crate::{IntoPattern, Resource, ResourcePath};
 
 const MAX_DYNAMIC_SEGMENTS: usize = 16;
 
@@ -32,21 +32,53 @@ enum PatternType {
     Static(String),
     Prefix(String),
     Dynamic(Regex, Vec<Rc<String>>, usize),
+    DynamicSet(RegexSet, Vec<(Regex, Vec<Rc<String>>, usize)>),
 }
 
 impl ResourceDef {
     /// Parse path pattern and create new `Pattern` instance.
     ///
-    /// Panics if path pattern is wrong.
-    pub fn new(path: &str) -> Self {
-        ResourceDef::with_prefix(path, false)
+    /// Panics if path pattern is malformed.
+    pub fn new<T: IntoPattern>(path: T) -> Self {
+        if path.is_single() {
+            let patterns = path.patterns();
+            ResourceDef::with_prefix(&patterns[0], false)
+        } else {
+            let set = path.patterns();
+            let mut data = Vec::new();
+            let mut re_set = Vec::new();
+
+            for path in set {
+                let (pattern, _, _, len) = ResourceDef::parse(&path, false);
+
+                let re = match Regex::new(&pattern) {
+                    Ok(re) => re,
+                    Err(err) => panic!("Wrong path pattern: \"{}\" {}", path, err),
+                };
+                // actix creates one router per thread
+                let names: Vec<_> = re
+                    .capture_names()
+                    .filter_map(|name| name.map(|name| Rc::new(name.to_owned())))
+                    .collect();
+                data.push((re, names, len));
+                re_set.push(pattern);
+            }
+
+            ResourceDef {
+                id: 0,
+                tp: PatternType::DynamicSet(RegexSet::new(re_set).unwrap(), data),
+                elements: Vec::new(),
+                name: String::new(),
+                pattern: "".to_owned(),
+            }
+        }
     }
 
     /// Parse path pattern and create new `Pattern` instance.
     ///
     /// Use `prefix` type instead of `static`.
     ///
-    /// Panics if path regex pattern is wrong.
+    /// Panics if path regex pattern is malformed.
     pub fn prefix(path: &str) -> Self {
         ResourceDef::with_prefix(path, true)
     }
@@ -57,7 +89,7 @@ impl ResourceDef {
     ///
     /// Use `prefix` type instead of `static`.
     ///
-    /// Panics if path regex pattern is wrong.
+    /// Panics if path regex pattern is malformed.
     pub fn root_prefix(path: &str) -> Self {
         ResourceDef::with_prefix(&insert_slash(path), true)
     }
@@ -123,8 +155,9 @@ impl ResourceDef {
     pub fn is_match(&self, path: &str) -> bool {
         match self.tp {
             PatternType::Static(ref s) => s == path,
-            PatternType::Dynamic(ref re, _, _) => re.is_match(path),
             PatternType::Prefix(ref s) => path.starts_with(s),
+            PatternType::Dynamic(ref re, _, _) => re.is_match(path),
+            PatternType::DynamicSet(ref re, _) => re.is_match(path),
         }
     }
 
@@ -176,6 +209,30 @@ impl ResourceDef {
                 };
                 Some(min(plen, len))
             }
+            PatternType::DynamicSet(ref re, ref params) => {
+                if let Some(idx) = re.matches(path).into_iter().next() {
+                    let (ref pattern, _, len) = params[idx];
+                    if let Some(captures) = pattern.captures(path) {
+                        let mut pos = 0;
+                        let mut passed = false;
+                        for capture in captures.iter() {
+                            if let Some(ref m) = capture {
+                                if !passed {
+                                    passed = true;
+                                    continue;
+                                }
+
+                                pos = m.end();
+                            }
+                        }
+                        Some(pos + len)
+                    } else {
+                        None
+                    }
+                } else {
+                    None
+                }
+            }
         }
     }
 
@@ -190,6 +247,25 @@ impl ResourceDef {
                     false
                 }
             }
+            PatternType::Prefix(ref s) => {
+                let rpath = path.path();
+                let len = if s == rpath {
+                    s.len()
+                } else if rpath.starts_with(s)
+                    && (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
+                {
+                    if s.ends_with('/') {
+                        s.len() - 1
+                    } else {
+                        s.len()
+                    }
+                } else {
+                    return false;
+                };
+                let rpath_len = rpath.len();
+                path.skip(min(rpath_len, len) as u16);
+                true
+            }
             PatternType::Dynamic(ref re, ref names, len) => {
                 let mut idx = 0;
                 let mut pos = 0;
@@ -219,24 +295,40 @@ impl ResourceDef {
                 path.skip((pos + len) as u16);
                 true
             }
-            PatternType::Prefix(ref s) => {
-                let rpath = path.path();
-                let len = if s == rpath {
-                    s.len()
-                } else if rpath.starts_with(s)
-                    && (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
-                {
-                    if s.ends_with('/') {
-                        s.len() - 1
+            PatternType::DynamicSet(ref re, ref params) => {
+                if let Some(idx) = re.matches(path.path()).into_iter().next() {
+                    let (ref pattern, ref names, len) = params[idx];
+                    let mut idx = 0;
+                    let mut pos = 0;
+                    let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
+                        [PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
+
+                    if let Some(captures) = pattern.captures(path.path()) {
+                        for (no, name) in names.iter().enumerate() {
+                            if let Some(m) = captures.name(&name) {
+                                idx += 1;
+                                pos = m.end();
+                                segments[no] =
+                                    PathItem::Segment(m.start() as u16, m.end() as u16);
+                            } else {
+                                log::error!(
+                                    "Dynamic path match but not all segments found: {}",
+                                    name
+                                );
+                                return false;
+                            }
+                        }
                     } else {
-                        s.len()
+                        return false;
                     }
+                    for idx in 0..idx {
+                        path.add(names[idx].clone(), segments[idx]);
+                    }
+                    path.skip((pos + len) as u16);
+                    true
                 } else {
-                    return false;
-                };
-                let rpath_len = rpath.len();
-                path.skip(min(rpath_len, len) as u16);
-                true
+                    false
+                }
             }
         }
     }
@@ -263,6 +355,30 @@ impl ResourceDef {
                     false
                 }
             }
+            PatternType::Prefix(ref s) => {
+                let len = {
+                    let rpath = res.resource_path().path();
+                    if s == rpath {
+                        s.len()
+                    } else if rpath.starts_with(s)
+                        && (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
+                    {
+                        if s.ends_with('/') {
+                            s.len() - 1
+                        } else {
+                            s.len()
+                        }
+                    } else {
+                        return false;
+                    }
+                };
+                if !check(res, user_data) {
+                    return false;
+                }
+                let path = res.resource_path();
+                path.skip(min(path.path().len(), len) as u16);
+                true
+            }
             PatternType::Dynamic(ref re, ref names, len) => {
                 let mut idx = 0;
                 let mut pos = 0;
@@ -298,29 +414,47 @@ impl ResourceDef {
                 path.skip((pos + len) as u16);
                 true
             }
-            PatternType::Prefix(ref s) => {
-                let len = {
-                    let rpath = res.resource_path().path();
-                    if s == rpath {
-                        s.len()
-                    } else if rpath.starts_with(s)
-                        && (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
-                    {
-                        if s.ends_with('/') {
-                            s.len() - 1
-                        } else {
-                            s.len()
+            PatternType::DynamicSet(ref re, ref params) => {
+                let path = res.resource_path().path();
+                if let Some(idx) = re.matches(path).into_iter().next() {
+                    let (ref pattern, ref names, len) = params[idx];
+                    let mut idx = 0;
+                    let mut pos = 0;
+                    let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
+                        [PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
+
+                    if let Some(captures) = pattern.captures(path) {
+                        for (no, name) in names.iter().enumerate() {
+                            if let Some(m) = captures.name(&name) {
+                                idx += 1;
+                                pos = m.end();
+                                segments[no] =
+                                    PathItem::Segment(m.start() as u16, m.end() as u16);
+                            } else {
+                                log::error!(
+                                    "Dynamic path match but not all segments found: {}",
+                                    name
+                                );
+                                return false;
+                            }
                         }
                     } else {
                         return false;
                     }
-                };
-                if !check(res, user_data) {
-                    return false;
+
+                    if !check(res, user_data) {
+                        return false;
+                    }
+
+                    let path = res.resource_path();
+                    for idx in 0..idx {
+                        path.add(names[idx].clone(), segments[idx]);
+                    }
+                    path.skip((pos + len) as u16);
+                    true
+                } else {
+                    false
                 }
-                let path = res.resource_path();
-                path.skip(min(path.path().len(), len) as u16);
-                true
             }
         }
     }
@@ -348,7 +482,10 @@ impl ResourceDef {
                     }
                 }
             }
-        };
+            PatternType::DynamicSet(..) => {
+                return false;
+            }
+        }
         true
     }
 
@@ -476,7 +613,7 @@ impl<'a> From<&'a str> for ResourceDef {
 
 impl From<String> for ResourceDef {
     fn from(path: String) -> ResourceDef {
-        ResourceDef::new(&path)
+        ResourceDef::new(path)
     }
 }
 
@@ -558,6 +695,65 @@ mod tests {
         assert_eq!(path.get("id").unwrap(), "012345");
     }
 
+    #[test]
+    fn test_dynamic_set() {
+        let re = ResourceDef::new(vec![
+            "/user/{id}",
+            "/v{version}/resource/{id}",
+            "/{id:[[:digit:]]{6}}",
+        ]);
+        assert!(re.is_match("/user/profile"));
+        assert!(re.is_match("/user/2345"));
+        assert!(!re.is_match("/user/2345/"));
+        assert!(!re.is_match("/user/2345/sdg"));
+
+        let mut path = Path::new("/user/profile");
+        assert!(re.match_path(&mut path));
+        assert_eq!(path.get("id").unwrap(), "profile");
+
+        let mut path = Path::new("/user/1245125");
+        assert!(re.match_path(&mut path));
+        assert_eq!(path.get("id").unwrap(), "1245125");
+
+        assert!(re.is_match("/v1/resource/320120"));
+        assert!(!re.is_match("/v/resource/1"));
+        assert!(!re.is_match("/resource"));
+
+        let mut path = Path::new("/v151/resource/adahg32");
+        assert!(re.match_path(&mut path));
+        assert_eq!(path.get("version").unwrap(), "151");
+        assert_eq!(path.get("id").unwrap(), "adahg32");
+
+        assert!(re.is_match("/012345"));
+        assert!(!re.is_match("/012"));
+        assert!(!re.is_match("/01234567"));
+        assert!(!re.is_match("/XXXXXX"));
+
+        let mut path = Path::new("/012345");
+        assert!(re.match_path(&mut path));
+        assert_eq!(path.get("id").unwrap(), "012345");
+
+        let re = ResourceDef::new([
+            "/user/{id}",
+            "/v{version}/resource/{id}",
+            "/{id:[[:digit:]]{6}}",
+        ]);
+        assert!(re.is_match("/user/profile"));
+        assert!(re.is_match("/user/2345"));
+        assert!(!re.is_match("/user/2345/"));
+        assert!(!re.is_match("/user/2345/sdg"));
+
+        let re = ResourceDef::new([
+            "/user/{id}".to_string(),
+            "/v{version}/resource/{id}".to_string(),
+            "/{id:[[:digit:]]{6}}".to_string(),
+        ]);
+        assert!(re.is_match("/user/profile"));
+        assert!(re.is_match("/user/2345"));
+        assert!(!re.is_match("/user/2345/"));
+        assert!(!re.is_match("/user/2345/sdg"));
+    }
+
     #[test]
     fn test_parse_tail() {
         let re = ResourceDef::new("/user/-{id}*");

router/src/router.rs

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

string/CHANGES.md

@@ -1,9 +1,15 @@
 # Changes
 
-[0.1.1] - 2019-12-07
+## [0.1.2] - 2019-12-22
+
+* Fix `new()` method
+
+* Make `ByteString::from_static()` and `ByteString::from_bytes_unchecked()` methods const.
+
+## [0.1.1] - 2019-12-07
 
 * Fix hash impl
 
-[0.1.0] - 2019-12-07
+## [0.1.0] - 2019-12-07
 
 * Initial release

string/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bytestring"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "A UTF-8 encoded string with Bytes as a storage"
 keywords = ["actix"]
@@ -9,11 +9,10 @@ repository = "https://github.com/actix/actix-net.git"
 documentation = "https://docs.rs/bytestring/"
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
 
 [lib]
 name = "bytestring"
 path = "src/lib.rs"
 
 [dependencies]
-bytes = "0.5.2"
+bytes = "0.5.3"

string/src/lib.rs

@@ -6,14 +6,14 @@ use bytes::Bytes;
 
 /// A utf-8 encoded string with [`Bytes`] as a storage.
 ///
-/// [`Bytes`]: https://docs.rs/bytes/0.5.2/bytes/struct.Bytes.html
+/// [`Bytes`]: https://docs.rs/bytes/0.5.3/bytes/struct.Bytes.html
 #[derive(Clone, Eq, PartialEq, Ord, PartialOrd, Default)]
 pub struct ByteString(Bytes);
 
 impl ByteString {
     /// Creates a new `ByteString`.
-    pub fn new() -> String {
-        String::default()
+    pub fn new() -> Self {
+        ByteString(Bytes::new())
     }
 
     /// Get a reference to the underlying bytes object.
@@ -27,12 +27,12 @@ impl ByteString {
     }
 
     /// Creates a new `ByteString` from a static str.
-    pub fn from_static(src: &'static str) -> ByteString {
-        Self(Bytes::from_static(src.as_ref()))
+    pub const fn from_static(src: &'static str) -> ByteString {
+        Self(Bytes::from_static(src.as_bytes()))
     }
 
     /// Creates a new `ByteString` from a Bytes.
-    pub unsafe fn from_bytes_unchecked(src: Bytes) -> ByteString {
+    pub const unsafe fn from_bytes_unchecked(src: Bytes) -> ByteString {
         Self(src)
     }
 }
@@ -147,6 +147,11 @@ mod test {
     use std::collections::hash_map::DefaultHasher;
     use std::hash::{Hash, Hasher};
 
+    #[test]
+    fn test_new() {
+        let _: ByteString = ByteString::new();
+    }
+
     #[test]
     fn test_hash() {
         let mut hasher1 = DefaultHasher::default();
@@ -171,6 +176,7 @@ mod test {
 
     #[test]
     fn test_from_static_str() {
+        const _S: ByteString = ByteString::from_static("hello");
         let _ = ByteString::from_static("str");
     }