migrate to -utils beta 4 (#2127)

actix-http/CHANGES.md

@@ -9,8 +9,12 @@
 ### Changed
 * `client::Connector` type now only have one generic type for `actix_service::Service`. [#2063]
 
+### Removed
+* `ResponseError` impl for `actix_utils::timeout::TimeoutError`. [#2127]
+
 [#2063]: https://github.com/actix/actix-web/pull/2063
 [#2081]: https://github.com/actix/actix-web/pull/2081
+[#2127]: https://github.com/actix/actix-web/pull/2127
 
 
 ## 3.0.0-beta.4 - 2021-03-08

actix-http/Cargo.toml

@@ -46,7 +46,7 @@ trust-dns = ["trust-dns-resolver"]
 [dependencies]
 actix-service = "2.0.0-beta.4"
 actix-codec = "0.4.0-beta.1"
-actix-utils = "3.0.0-beta.2"
+actix-utils = "3.0.0-beta.4"
 actix-rt = "2.2"
 actix-tls = "3.0.0-beta.5"
 
@@ -65,11 +65,13 @@ http = "0.2.2"
 httparse = "1.3"
 itoa = "0.4"
 language-tags = "0.2"
+local-channel = "0.1"
 once_cell = "1.5"
 log = "0.4"
 mime = "0.3"
 percent-encoding = "2.1"
 pin-project = "1.0.0"
+pin-project-lite = "0.2"
 rand = "0.8"
 regex = "1.3"
 serde = "1.0"

actix-http/examples/hello-world.rs

@@ -2,7 +2,7 @@ use std::{env, io};
 
 use actix_http::{HttpService, Response};
 use actix_server::Server;
-use futures_util::future;
+use actix_utils::future;
 use http::header::HeaderValue;
 use log::info;
 

actix-http/src/body/mod.rs

@@ -20,8 +20,9 @@ mod tests {
     use std::pin::Pin;
 
     use actix_rt::pin;
+    use actix_utils::future::poll_fn;
     use bytes::{Bytes, BytesMut};
-    use futures_util::{future::poll_fn, stream};
+    use futures_util::stream;
 
     use super::*;
 

actix-http/src/client/h1proto.rs

@@ -5,10 +5,11 @@ use std::{
 };
 
 use actix_codec::Framed;
+use actix_utils::future::poll_fn;
 use bytes::buf::BufMut;
 use bytes::{Bytes, BytesMut};
 use futures_core::{ready, Stream};
-use futures_util::{future::poll_fn, SinkExt as _};
+use futures_util::SinkExt as _;
 
 use crate::error::PayloadError;
 use crate::h1;

actix-http/src/client/h2proto.rs

@@ -1,7 +1,7 @@
 use std::future::Future;
 
+use actix_utils::future::poll_fn;
 use bytes::Bytes;
-use futures_util::future::poll_fn;
 use h2::{
     client::{Builder, Connection, SendRequest},
     SendStream,

actix-http/src/error.rs

@@ -6,9 +6,6 @@ use std::str::Utf8Error;
 use std::string::FromUtf8Error;
 use std::{fmt, io, result};
 
-use actix_codec::{Decoder, Encoder};
-use actix_utils::dispatcher::DispatcherError as FramedDispatcherError;
-use actix_utils::timeout::TimeoutError;
 use bytes::BytesMut;
 use derive_more::{Display, From};
 use http::uri::InvalidUri;
@@ -148,19 +145,6 @@ impl From<ResponseBuilder> for Error {
     }
 }
 
-/// Inspects the underlying enum and returns an appropriate status code.
-///
-/// If the variant is [`TimeoutError::Service`], the error code of the service is returned.
-/// Otherwise, [`StatusCode::GATEWAY_TIMEOUT`] is returned.
-impl<E: ResponseError> ResponseError for TimeoutError<E> {
-    fn status_code(&self) -> StatusCode {
-        match self {
-            TimeoutError::Service(e) => e.status_code(),
-            TimeoutError::Timeout => StatusCode::GATEWAY_TIMEOUT,
-        }
-    }
-}
-
 #[derive(Debug, Display)]
 #[display(fmt = "UnknownError")]
 struct UnitError;
@@ -469,14 +453,6 @@ impl ResponseError for ContentTypeError {
     }
 }
 
-impl<E, U: Encoder<I> + Decoder, I> ResponseError for FramedDispatcherError<E, U, I>
-where
-    E: fmt::Debug + fmt::Display,
-    <U as Encoder<I>>::Error: fmt::Debug,
-    <U as Decoder>::Error: fmt::Debug,
-{
-}
-
 /// Helper type that can wrap any error and generate custom response.
 ///
 /// In following example any `io::Error` will be converted into "BAD REQUEST"

actix-http/src/h1/dispatcher.rs

@@ -951,7 +951,8 @@ mod tests {
     use std::str;
 
     use actix_service::fn_service;
-    use futures_util::future::{lazy, ready, Ready};
+    use actix_utils::future::{ready, Ready};
+    use futures_util::future::lazy;
 
     use super::*;
     use crate::{

actix-http/src/h1/expect.rs

@@ -1,5 +1,5 @@
 use actix_service::{Service, ServiceFactory};
-use futures_util::future::{ready, Ready};
+use actix_utils::future::{ready, Ready};
 
 use crate::error::Error;
 use crate::request::Request;

actix-http/src/h1/payload.rs

@@ -263,7 +263,7 @@ impl Inner {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use futures_util::future::poll_fn;
+    use actix_utils::future::poll_fn;
 
     #[actix_rt::test]
     async fn test_unread_data() {

actix-http/src/h1/service.rs

@@ -6,8 +6,8 @@ use std::{fmt, net};
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
 use actix_rt::net::TcpStream;
 use actix_service::{pipeline_factory, IntoServiceFactory, Service, ServiceFactory};
+use actix_utils::future::ready;
 use futures_core::future::LocalBoxFuture;
-use futures_util::future::ready;
 
 use crate::body::MessageBody;
 use crate::config::ServiceConfig;

actix-http/src/h2/service.rs

@@ -10,9 +10,9 @@ use actix_service::{
     fn_factory, fn_service, pipeline_factory, IntoServiceFactory, Service,
     ServiceFactory,
 };
+use actix_utils::future::ready;
 use bytes::Bytes;
 use futures_core::{future::LocalBoxFuture, ready};
-use futures_util::future::ready;
 use h2::server::{handshake, Handshake};
 use log::error;
 

actix-http/src/ws/dispatcher.rs

@@ -4,7 +4,6 @@ use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
 use actix_service::{IntoService, Service};
-use actix_utils::dispatcher::{Dispatcher as InnerDispatcher, DispatcherError};
 
 use super::{Codec, Frame, Message};
 
@@ -15,7 +14,7 @@ where
     T: AsyncRead + AsyncWrite,
 {
     #[pin]
-    inner: InnerDispatcher<S, T, Codec, Message>,
+    inner: inner::Dispatcher<S, T, Codec, Message>,
 }
 
 impl<S, T> Dispatcher<S, T>
@@ -27,13 +26,13 @@ where
 {
     pub fn new<F: IntoService<S, Frame>>(io: T, service: F) -> Self {
         Dispatcher {
-            inner: InnerDispatcher::new(Framed::new(io, Codec::new()), service),
+            inner: inner::Dispatcher::new(Framed::new(io, Codec::new()), service),
         }
     }
 
     pub fn with<F: IntoService<S, Frame>>(framed: Framed<T, Codec>, service: F) -> Self {
         Dispatcher {
-            inner: InnerDispatcher::new(framed, service),
+            inner: inner::Dispatcher::new(framed, service),
         }
     }
 }
@@ -45,9 +44,393 @@ where
     S::Future: 'static,
     S::Error: 'static,
 {
-    type Output = Result<(), DispatcherError<S::Error, Codec, Message>>;
+    type Output = Result<(), inner::DispatcherError<S::Error, Codec, Message>>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         self.project().inner.poll(cx)
     }
 }
+
+/// Framed dispatcher service and related utilities.
+mod inner {
+    // allow dead code since this mod was ripped from actix-utils
+    #![allow(dead_code)]
+
+    use core::{
+        fmt,
+        future::Future,
+        mem,
+        pin::Pin,
+        task::{Context, Poll},
+    };
+
+    use actix_service::{IntoService, Service};
+    use futures_core::stream::Stream;
+    use local_channel::mpsc;
+    use log::debug;
+    use pin_project_lite::pin_project;
+
+    use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
+
+    use crate::ResponseError;
+
+    /// Framed transport errors
+    pub enum DispatcherError<E, U, I>
+    where
+        U: Encoder<I> + Decoder,
+    {
+        /// Inner service error.
+        Service(E),
+
+        /// Frame encoding error.
+        Encoder(<U as Encoder<I>>::Error),
+
+        /// Frame decoding error.
+        Decoder(<U as Decoder>::Error),
+    }
+
+    impl<E, U, I> From<E> for DispatcherError<E, U, I>
+    where
+        U: Encoder<I> + Decoder,
+    {
+        fn from(err: E) -> Self {
+            DispatcherError::Service(err)
+        }
+    }
+
+    impl<E, U, I> fmt::Debug for DispatcherError<E, U, I>
+    where
+        E: fmt::Debug,
+        U: Encoder<I> + Decoder,
+        <U as Encoder<I>>::Error: fmt::Debug,
+        <U as Decoder>::Error: fmt::Debug,
+    {
+        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+            match *self {
+                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, I> fmt::Display for DispatcherError<E, U, I>
+    where
+        E: fmt::Display,
+        U: Encoder<I> + Decoder,
+        <U as Encoder<I>>::Error: fmt::Debug,
+        <U as Decoder>::Error: fmt::Debug,
+    {
+        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+            match *self {
+                DispatcherError::Service(ref e) => write!(fmt, "{}", e),
+                DispatcherError::Encoder(ref e) => write!(fmt, "{:?}", e),
+                DispatcherError::Decoder(ref e) => write!(fmt, "{:?}", e),
+            }
+        }
+    }
+
+    impl<E, U, I> ResponseError for DispatcherError<E, U, I>
+    where
+        E: fmt::Debug + fmt::Display,
+        U: Encoder<I> + Decoder,
+        <U as Encoder<I>>::Error: fmt::Debug,
+        <U as Decoder>::Error: fmt::Debug,
+    {
+    }
+
+    /// Message type wrapper for signalling end of message stream.
+    pub enum Message<T> {
+        /// Message item.
+        Item(T),
+
+        /// Signal from service to flush all messages and stop processing.
+        Close,
+    }
+
+    pin_project! {
+        /// A future that reads frames from a [`Framed`] object and passes them to a [`Service`].
+        pub struct Dispatcher<S, T, U, I>
+        where
+            S: Service<<U as Decoder>::Item, Response = I>,
+            S::Error: 'static,
+            S::Future: 'static,
+            T: AsyncRead,
+            T: AsyncWrite,
+            U: Encoder<I>,
+            U: Decoder,
+            I: 'static,
+            <U as Encoder<I>>::Error: fmt::Debug,
+        {
+            service: S,
+            state: State<S, U, I>,
+            #[pin]
+            framed: Framed<T, U>,
+            rx: mpsc::Receiver<Result<Message<I>, S::Error>>,
+            tx: mpsc::Sender<Result<Message<I>, S::Error>>,
+        }
+    }
+
+    enum State<S, U, I>
+    where
+        S: Service<<U as Decoder>::Item>,
+        U: Encoder<I> + Decoder,
+    {
+        Processing,
+        Error(DispatcherError<S::Error, U, I>),
+        FramedError(DispatcherError<S::Error, U, I>),
+        FlushAndStop,
+        Stopping,
+    }
+
+    impl<S, U, I> State<S, U, I>
+    where
+        S: Service<<U as Decoder>::Item>,
+        U: Encoder<I> + Decoder,
+    {
+        fn take_error(&mut self) -> DispatcherError<S::Error, U, I> {
+            match mem::replace(self, State::Processing) {
+                State::Error(err) => err,
+                _ => panic!(),
+            }
+        }
+
+        fn take_framed_error(&mut self) -> DispatcherError<S::Error, U, I> {
+            match mem::replace(self, State::Processing) {
+                State::FramedError(err) => err,
+                _ => panic!(),
+            }
+        }
+    }
+
+    impl<S, T, U, I> Dispatcher<S, T, U, I>
+    where
+        S: Service<<U as Decoder>::Item, Response = I>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder<I>,
+        I: 'static,
+        <U as Decoder>::Error: fmt::Debug,
+        <U as Encoder<I>>::Error: fmt::Debug,
+    {
+        /// Create new `Dispatcher`.
+        pub fn new<F>(framed: Framed<T, U>, service: F) -> Self
+        where
+            F: IntoService<S, <U as Decoder>::Item>,
+        {
+            let (tx, rx) = mpsc::channel();
+            Dispatcher {
+                framed,
+                rx,
+                tx,
+                service: service.into_service(),
+                state: State::Processing,
+            }
+        }
+
+        /// Construct new `Dispatcher` instance with customer `mpsc::Receiver`
+        pub fn with_rx<F>(
+            framed: Framed<T, U>,
+            service: F,
+            rx: mpsc::Receiver<Result<Message<I>, S::Error>>,
+        ) -> Self
+        where
+            F: IntoService<S, <U as Decoder>::Item>,
+        {
+            let tx = rx.sender();
+            Dispatcher {
+                framed,
+                rx,
+                tx,
+                service: service.into_service(),
+                state: State::Processing,
+            }
+        }
+
+        /// Get sender handle.
+        pub fn tx(&self) -> mpsc::Sender<Result<Message<I>, S::Error>> {
+            self.tx.clone()
+        }
+
+        /// Get reference to a service wrapped by `Dispatcher` instance.
+        pub fn service(&self) -> &S {
+            &self.service
+        }
+
+        /// Get mutable reference to a service wrapped by `Dispatcher` instance.
+        pub fn service_mut(&mut self) -> &mut S {
+            &mut self.service
+        }
+
+        /// Get reference to a framed instance wrapped by `Dispatcher` instance.
+        pub fn framed(&self) -> &Framed<T, U> {
+            &self.framed
+        }
+
+        /// Get mutable reference to a framed instance wrapped by `Dispatcher` instance.
+        pub fn framed_mut(&mut self) -> &mut Framed<T, U> {
+            &mut self.framed
+        }
+
+        /// Read from framed object.
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
+        where
+            S: Service<<U as Decoder>::Item, Response = I>,
+            S::Error: 'static,
+            S::Future: 'static,
+            T: AsyncRead + AsyncWrite,
+            U: Decoder + Encoder<I>,
+            I: 'static,
+            <U as Encoder<I>>::Error: fmt::Debug,
+        {
+            loop {
+                let this = self.as_mut().project();
+                match this.service.poll_ready(cx) {
+                    Poll::Ready(Ok(_)) => {
+                        let item = match this.framed.next_item(cx) {
+                            Poll::Ready(Some(Ok(el))) => el,
+                            Poll::Ready(Some(Err(err))) => {
+                                *this.state =
+                                    State::FramedError(DispatcherError::Decoder(err));
+                                return true;
+                            }
+                            Poll::Pending => return false,
+                            Poll::Ready(None) => {
+                                *this.state = State::Stopping;
+                                return true;
+                            }
+                        };
+
+                        let tx = this.tx.clone();
+                        let fut = this.service.call(item);
+                        actix_rt::spawn(async move {
+                            let item = fut.await;
+                            let _ = tx.send(item.map(Message::Item));
+                        });
+                    }
+                    Poll::Pending => return false,
+                    Poll::Ready(Err(err)) => {
+                        *this.state = State::Error(DispatcherError::Service(err));
+                        return true;
+                    }
+                }
+            }
+        }
+
+        /// Write to framed object.
+        fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
+        where
+            S: Service<<U as Decoder>::Item, Response = I>,
+            S::Error: 'static,
+            S::Future: 'static,
+            T: AsyncRead + AsyncWrite,
+            U: Decoder + Encoder<I>,
+            I: 'static,
+            <U as Encoder<I>>::Error: fmt::Debug,
+        {
+            loop {
+                let mut this = self.as_mut().project();
+                while !this.framed.is_write_buf_full() {
+                    match Pin::new(&mut this.rx).poll_next(cx) {
+                        Poll::Ready(Some(Ok(Message::Item(msg)))) => {
+                            if let Err(err) = this.framed.as_mut().write(msg) {
+                                *this.state =
+                                    State::FramedError(DispatcherError::Encoder(err));
+                                return true;
+                            }
+                        }
+                        Poll::Ready(Some(Ok(Message::Close))) => {
+                            *this.state = State::FlushAndStop;
+                            return true;
+                        }
+                        Poll::Ready(Some(Err(err))) => {
+                            *this.state = State::Error(DispatcherError::Service(err));
+                            return true;
+                        }
+                        Poll::Ready(None) | Poll::Pending => break,
+                    }
+                }
+
+                if !this.framed.is_write_buf_empty() {
+                    match this.framed.flush(cx) {
+                        Poll::Pending => break,
+                        Poll::Ready(Ok(_)) => {}
+                        Poll::Ready(Err(err)) => {
+                            debug!("Error sending data: {:?}", err);
+                            *this.state =
+                                State::FramedError(DispatcherError::Encoder(err));
+                            return true;
+                        }
+                    }
+                } else {
+                    break;
+                }
+            }
+
+            false
+        }
+    }
+
+    impl<S, T, U, I> Future for Dispatcher<S, T, U, I>
+    where
+        S: Service<<U as Decoder>::Item, Response = I>,
+        S::Error: 'static,
+        S::Future: 'static,
+        T: AsyncRead + AsyncWrite,
+        U: Decoder + Encoder<I>,
+        I: 'static,
+        <U as Encoder<I>>::Error: fmt::Debug,
+        <U as Decoder>::Error: fmt::Debug,
+    {
+        type Output = Result<(), DispatcherError<S::Error, U, I>>;
+
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            loop {
+                let this = self.as_mut().project();
+
+                return match this.state {
+                    State::Processing => {
+                        if self.as_mut().poll_read(cx) || self.as_mut().poll_write(cx) {
+                            continue;
+                        } else {
+                            Poll::Pending
+                        }
+                    }
+                    State::Error(_) => {
+                        // flush write buffer
+                        if !this.framed.is_write_buf_empty()
+                            && this.framed.flush(cx).is_pending()
+                        {
+                            return Poll::Pending;
+                        }
+                        Poll::Ready(Err(this.state.take_error()))
+                    }
+                    State::FlushAndStop => {
+                        if !this.framed.is_write_buf_empty() {
+                            this.framed.flush(cx).map(|res| {
+                                if let Err(err) = res {
+                                    debug!("Error sending data: {:?}", err);
+                                }
+
+                                Ok(())
+                            })
+                        } else {
+                            Poll::Ready(Ok(()))
+                        }
+                    }
+                    State::FramedError(_) => {
+                        Poll::Ready(Err(this.state.take_framed_error()))
+                    }
+                    State::Stopping => Poll::Ready(Ok(())),
+                };
+            }
+        }
+    }
+}

actix-http/tests/test_client.rs

@@ -3,11 +3,9 @@ use actix_http::{
 };
 use actix_http_test::test_server;
 use actix_service::ServiceFactoryExt;
+use actix_utils::future;
 use bytes::Bytes;
-use futures_util::{
-    future::{self, ok},
-    StreamExt as _,
-};
+use futures_util::StreamExt as _;
 
 const STR: &str = "Hello World Hello World Hello World Hello World Hello World \
                    Hello World Hello World Hello World Hello World Hello World \
@@ -63,7 +61,7 @@ async fn test_h1_v2() {
 async fn test_connection_close() {
     let srv = test_server(move || {
         HttpService::build()
-            .finish(|_| ok::<_, ()>(Response::Ok().body(STR)))
+            .finish(|_| future::ok::<_, ()>(Response::Ok().body(STR)))
             .tcp()
             .map(|_| ())
     })
@@ -79,9 +77,9 @@ async fn test_with_query_parameter() {
         HttpService::build()
             .finish(|req: Request| {
                 if req.uri().query().unwrap().contains("qp=") {
-                    ok::<_, ()>(Response::Ok().finish())
+                    future::ok::<_, ()>(Response::Ok().finish())
                 } else {
-                    ok::<_, ()>(Response::BadRequest().finish())
+                    future::ok::<_, ()>(Response::BadRequest().finish())
                 }
             })
             .tcp()

actix-http/tests/test_openssl.rs

@@ -11,12 +11,10 @@ use actix_http::HttpMessage;
 use actix_http::{body, Error, HttpService, Request, Response};
 use actix_http_test::test_server;
 use actix_service::{fn_service, ServiceFactoryExt};
+use actix_utils::future::{err, ok, ready};
 use bytes::{Bytes, BytesMut};
 use futures_core::Stream;
-use futures_util::{
-    future::{err, ok, ready},
-    stream::{once, StreamExt as _},
-};
+use futures_util::stream::{once, StreamExt as _};
 use openssl::{
     pkey::PKey,
     ssl::{SslAcceptor, SslMethod},

actix-http/tests/test_rustls.rs

@@ -8,10 +8,10 @@ use actix_http::http::{Method, StatusCode, Version};
 use actix_http::{body, error, Error, HttpService, Request, Response};
 use actix_http_test::test_server;
 use actix_service::{fn_factory_with_config, fn_service};
+use actix_utils::future::{err, ok};
 
 use bytes::{Bytes, BytesMut};
 use futures_core::Stream;
-use futures_util::future::{self, err, ok};
 use futures_util::stream::{once, StreamExt as _};
 use rustls::{
     internal::pemfile::{certs, pkcs8_private_keys},
@@ -51,7 +51,7 @@ fn tls_config() -> RustlsServerConfig {
 async fn test_h1() -> io::Result<()> {
     let srv = test_server(move || {
         HttpService::build()
-            .h1(|_| future::ok::<_, Error>(Response::Ok().finish()))
+            .h1(|_| ok::<_, Error>(Response::Ok().finish()))
             .rustls(tls_config())
     })
     .await;
@@ -65,7 +65,7 @@ async fn test_h1() -> io::Result<()> {
 async fn test_h2() -> io::Result<()> {
     let srv = test_server(move || {
         HttpService::build()
-            .h2(|_| future::ok::<_, Error>(Response::Ok().finish()))
+            .h2(|_| ok::<_, Error>(Response::Ok().finish()))
             .rustls(tls_config())
     })
     .await;
@@ -82,7 +82,7 @@ async fn test_h1_1() -> io::Result<()> {
             .h1(|req: Request| {
                 assert!(req.peer_addr().is_some());
                 assert_eq!(req.version(), Version::HTTP_11);
-                future::ok::<_, Error>(Response::Ok().finish())
+                ok::<_, Error>(Response::Ok().finish())
             })
             .rustls(tls_config())
     })
@@ -100,7 +100,7 @@ async fn test_h2_1() -> io::Result<()> {
             .finish(|req: Request| {
                 assert!(req.peer_addr().is_some());
                 assert_eq!(req.version(), Version::HTTP_2);
-                future::ok::<_, Error>(Response::Ok().finish())
+                ok::<_, Error>(Response::Ok().finish())
             })
             .rustls(tls_config())
     })
@@ -144,7 +144,7 @@ async fn test_h2_content_length() {
                     StatusCode::OK,
                     StatusCode::NOT_FOUND,
                 ];
-                future::ok::<_, ()>(Response::new(statuses[indx]))
+                ok::<_, ()>(Response::new(statuses[indx]))
             })
             .rustls(tls_config())
     })
@@ -213,7 +213,7 @@ async fn test_h2_headers() {
                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST ",
                 ));
             }
-            future::ok::<_, ()>(config.body(data.clone()))
+            ok::<_, ()>(config.body(data.clone()))
         })
             .rustls(tls_config())
     }).await;
@@ -252,7 +252,7 @@ const STR: &str = "Hello World Hello World Hello World Hello World Hello World \
 async fn test_h2_body2() {
     let mut srv = test_server(move || {
         HttpService::build()
-            .h2(|_| future::ok::<_, ()>(Response::Ok().body(STR)))
+            .h2(|_| ok::<_, ()>(Response::Ok().body(STR)))
             .rustls(tls_config())
     })
     .await;

actix-http/tests/test_server.rs

@@ -5,9 +5,10 @@ use std::{net, thread};
 use actix_http_test::test_server;
 use actix_rt::time::sleep;
 use actix_service::fn_service;
+use actix_utils::future::{err, ok, ready};
 use bytes::Bytes;
-use futures_util::future::{self, err, ok, ready, FutureExt};
 use futures_util::stream::{once, StreamExt as _};
+use futures_util::FutureExt as _;
 use regex::Regex;
 
 use actix_http::HttpMessage;
@@ -24,7 +25,7 @@ async fn test_h1() {
             .client_disconnect(1000)
             .h1(|req: Request| {
                 assert!(req.peer_addr().is_some());
-                future::ok::<_, ()>(Response::Ok().finish())
+                ok::<_, ()>(Response::Ok().finish())
             })
             .tcp()
     })
@@ -44,7 +45,7 @@ async fn test_h1_2() {
             .finish(|req: Request| {
                 assert!(req.peer_addr().is_some());
                 assert_eq!(req.version(), http::Version::HTTP_11);
-                future::ok::<_, ()>(Response::Ok().finish())
+                ok::<_, ()>(Response::Ok().finish())
             })
             .tcp()
     })
@@ -65,7 +66,7 @@ async fn test_expect_continue() {
                     err(error::ErrorPreconditionFailed("error"))
                 }
             }))
-            .finish(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .finish(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -96,7 +97,7 @@ async fn test_expect_continue_h1() {
                     }
                 })
             }))
-            .h1(fn_service(|_| future::ok::<_, ()>(Response::Ok().finish())))
+            .h1(fn_service(|_| ok::<_, ()>(Response::Ok().finish())))
             .tcp()
     })
     .await;
@@ -175,7 +176,7 @@ async fn test_slow_request() {
     let srv = test_server(|| {
         HttpService::build()
             .client_timeout(100)
-            .finish(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .finish(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -191,7 +192,7 @@ async fn test_slow_request() {
 async fn test_http1_malformed_request() {
     let srv = test_server(|| {
         HttpService::build()
-            .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .h1(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -207,7 +208,7 @@ async fn test_http1_malformed_request() {
 async fn test_http1_keepalive() {
     let srv = test_server(|| {
         HttpService::build()
-            .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .h1(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -229,7 +230,7 @@ async fn test_http1_keepalive_timeout() {
     let srv = test_server(|| {
         HttpService::build()
             .keep_alive(1)
-            .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .h1(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -250,7 +251,7 @@ async fn test_http1_keepalive_timeout() {
 async fn test_http1_keepalive_close() {
     let srv = test_server(|| {
         HttpService::build()
-            .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .h1(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -271,7 +272,7 @@ async fn test_http1_keepalive_close() {
 async fn test_http10_keepalive_default_close() {
     let srv = test_server(|| {
         HttpService::build()
-            .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .h1(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -291,7 +292,7 @@ async fn test_http10_keepalive_default_close() {
 async fn test_http10_keepalive() {
     let srv = test_server(|| {
         HttpService::build()
-            .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .h1(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -319,7 +320,7 @@ async fn test_http1_keepalive_disabled() {
     let srv = test_server(|| {
         HttpService::build()
             .keep_alive(KeepAlive::Disabled)
-            .h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
+            .h1(|_| ok::<_, ()>(Response::Ok().finish()))
             .tcp()
     })
     .await;
@@ -354,7 +355,7 @@ async fn test_content_length() {
                     StatusCode::OK,
                     StatusCode::NOT_FOUND,
                 ];
-                future::ok::<_, ()>(Response::new(statuses[indx]))
+                ok::<_, ()>(Response::new(statuses[indx]))
             })
             .tcp()
     })
@@ -409,7 +410,7 @@ async fn test_h1_headers() {
                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST ",
                 ));
             }
-            future::ok::<_, ()>(builder.body(data.clone()))
+            ok::<_, ()>(builder.body(data.clone()))
         }).tcp()
     }).await;
 
@@ -645,7 +646,7 @@ async fn test_h1_response_http_error_handling() {
 async fn test_h1_service_error() {
     let mut srv = test_server(|| {
         HttpService::build()
-            .h1(|_| future::err::<Response, Error>(error::ErrorBadRequest("error")))
+            .h1(|_| err::<Response, _>(error::ErrorBadRequest("error")))
             .tcp()
     })
     .await;
@@ -667,7 +668,7 @@ async fn test_h1_on_connect() {
             })
             .h1(|req: Request| {
                 assert!(req.extensions().contains::<isize>());
-                future::ok::<_, ()>(Response::Ok().finish())
+                ok::<_, ()>(Response::Ok().finish())
             })
             .tcp()
     })

actix-http/tests/test_ws.rs

@@ -9,11 +9,12 @@ use actix_codec::{AsyncRead, AsyncWrite, Framed};
 use actix_http::{body, h1, ws, Error, HttpService, Request, Response};
 use actix_http_test::test_server;
 use actix_service::{fn_factory, Service};
-use actix_utils::dispatcher::Dispatcher;
+use actix_utils::future;
 use bytes::Bytes;
-use futures_util::future;
 use futures_util::{SinkExt as _, StreamExt as _};
 
+use crate::ws::Dispatcher;
+
 struct WsService<T>(Arc<Mutex<(PhantomData<T>, Cell<bool>)>>);
 
 impl<T> WsService<T> {
@@ -58,7 +59,7 @@ where
                 .await
                 .unwrap();
 
-            Dispatcher::new(framed.replace_codec(ws::Codec::new()), service)
+            Dispatcher::with(framed.replace_codec(ws::Codec::new()), service)
                 .await
                 .map_err(|_| panic!())
         };