body ergonomics v3 (#2468)

awc/Cargo.toml

@@ -105,6 +105,7 @@ brotli2 = "0.3.2"
 env_logger = "0.9"
 flate2 = "1.0.13"
 futures-util = { version = "0.3.7", default-features = false }
+static_assertions = "1.1"
 rcgen = "0.8"
 rustls-pemfile = "0.2"
 

awc/src/any_body.rs

@@ -0,0 +1,266 @@
+use std::{
+    borrow::Cow,
+    fmt, mem,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use bytes::{Bytes, BytesMut};
+use futures_core::Stream;
+use pin_project_lite::pin_project;
+
+use actix_http::body::{BodySize, BodyStream, BoxBody, MessageBody, SizedStream};
+
+use crate::BoxError;
+
+pin_project! {
+    /// Represents various types of HTTP message body.
+    #[derive(Clone)]
+    #[project = AnyBodyProj]
+    pub enum AnyBody<B = BoxBody> {
+        /// Empty response. `Content-Length` header is not set.
+        None,
+
+        /// Complete, in-memory response body.
+        Bytes { body: Bytes },
+
+        /// Generic / Other message body.
+        Body { #[pin] body: B },
+    }
+}
+
+impl AnyBody {
+    /// Constructs a "body" representing an empty response.
+    pub fn none() -> Self {
+        Self::None
+    }
+
+    /// Constructs a new, 0-length body.
+    pub fn empty() -> Self {
+        Self::Bytes { body: Bytes::new() }
+    }
+
+    /// Create boxed body from generic message body.
+    pub fn new_boxed<B>(body: B) -> Self
+    where
+        B: MessageBody + 'static,
+    {
+        Self::Body {
+            body: BoxBody::new(body),
+        }
+    }
+
+    /// Constructs new `AnyBody` instance from a slice of bytes by copying it.
+    ///
+    /// If your bytes container is owned, it may be cheaper to use a `From` impl.
+    pub fn copy_from_slice(s: &[u8]) -> Self {
+        Self::Bytes {
+            body: Bytes::copy_from_slice(s),
+        }
+    }
+
+    #[doc(hidden)]
+    #[deprecated(since = "4.0.0", note = "Renamed to `copy_from_slice`.")]
+    pub fn from_slice(s: &[u8]) -> Self {
+        Self::Bytes {
+            body: Bytes::copy_from_slice(s),
+        }
+    }
+}
+
+impl<B> AnyBody<B> {
+    /// Create body from generic message body.
+    pub fn new(body: B) -> Self {
+        Self::Body { body }
+    }
+}
+
+impl<B> AnyBody<B>
+where
+    B: MessageBody + 'static,
+{
+    pub fn into_boxed(self) -> AnyBody {
+        match self {
+            Self::None => AnyBody::None,
+            Self::Bytes { body: bytes } => AnyBody::Bytes { body: bytes },
+            Self::Body { body } => AnyBody::new_boxed(body),
+        }
+    }
+}
+
+impl<B> MessageBody for AnyBody<B>
+where
+    B: MessageBody,
+{
+    type Error = crate::BoxError;
+
+    fn size(&self) -> BodySize {
+        match self {
+            AnyBody::None => BodySize::None,
+            AnyBody::Bytes { ref body } => BodySize::Sized(body.len() as u64),
+            AnyBody::Body { ref body } => body.size(),
+        }
+    }
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
+        match self.project() {
+            AnyBodyProj::None => Poll::Ready(None),
+            AnyBodyProj::Bytes { body } => {
+                let len = body.len();
+                if len == 0 {
+                    Poll::Ready(None)
+                } else {
+                    Poll::Ready(Some(Ok(mem::take(body))))
+                }
+            }
+
+            AnyBodyProj::Body { body } => body.poll_next(cx).map_err(|err| err.into()),
+        }
+    }
+}
+
+impl PartialEq for AnyBody {
+    fn eq(&self, other: &AnyBody) -> bool {
+        match self {
+            AnyBody::None => matches!(*other, AnyBody::None),
+            AnyBody::Bytes { body } => match other {
+                AnyBody::Bytes { body: b2 } => body == b2,
+                _ => false,
+            },
+            AnyBody::Body { .. } => false,
+        }
+    }
+}
+
+impl<S: fmt::Debug> fmt::Debug for AnyBody<S> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match *self {
+            AnyBody::None => write!(f, "AnyBody::None"),
+            AnyBody::Bytes { ref body } => write!(f, "AnyBody::Bytes({:?})", body),
+            AnyBody::Body { ref body } => write!(f, "AnyBody::Message({:?})", body),
+        }
+    }
+}
+
+impl<B> From<&'static str> for AnyBody<B> {
+    fn from(string: &'static str) -> Self {
+        Self::Bytes {
+            body: Bytes::from_static(string.as_ref()),
+        }
+    }
+}
+
+impl<B> From<&'static [u8]> for AnyBody<B> {
+    fn from(bytes: &'static [u8]) -> Self {
+        Self::Bytes {
+            body: Bytes::from_static(bytes),
+        }
+    }
+}
+
+impl<B> From<Vec<u8>> for AnyBody<B> {
+    fn from(vec: Vec<u8>) -> Self {
+        Self::Bytes {
+            body: Bytes::from(vec),
+        }
+    }
+}
+
+impl<B> From<String> for AnyBody<B> {
+    fn from(string: String) -> Self {
+        Self::Bytes {
+            body: Bytes::from(string),
+        }
+    }
+}
+
+impl<B> From<&'_ String> for AnyBody<B> {
+    fn from(string: &String) -> Self {
+        Self::Bytes {
+            body: Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(&string)),
+        }
+    }
+}
+
+impl<B> From<Cow<'_, str>> for AnyBody<B> {
+    fn from(string: Cow<'_, str>) -> Self {
+        match string {
+            Cow::Owned(s) => Self::from(s),
+            Cow::Borrowed(s) => Self::Bytes {
+                body: Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(s)),
+            },
+        }
+    }
+}
+
+impl<B> From<Bytes> for AnyBody<B> {
+    fn from(bytes: Bytes) -> Self {
+        Self::Bytes { body: bytes }
+    }
+}
+
+impl<B> From<BytesMut> for AnyBody<B> {
+    fn from(bytes: BytesMut) -> Self {
+        Self::Bytes {
+            body: bytes.freeze(),
+        }
+    }
+}
+
+impl<S, E> From<SizedStream<S>> for AnyBody
+where
+    S: Stream<Item = Result<Bytes, E>> + 'static,
+    E: Into<BoxError> + 'static,
+{
+    fn from(stream: SizedStream<S>) -> Self {
+        AnyBody::new_boxed(stream)
+    }
+}
+
+impl<S, E> From<BodyStream<S>> for AnyBody
+where
+    S: Stream<Item = Result<Bytes, E>> + 'static,
+    E: Into<BoxError> + 'static,
+{
+    fn from(stream: BodyStream<S>) -> Self {
+        AnyBody::new_boxed(stream)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::marker::PhantomPinned;
+
+    use static_assertions::{assert_impl_all, assert_not_impl_all};
+
+    use super::*;
+
+    struct PinType(PhantomPinned);
+
+    impl MessageBody for PinType {
+        type Error = crate::BoxError;
+
+        fn size(&self) -> BodySize {
+            unimplemented!()
+        }
+
+        fn poll_next(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Option<Result<Bytes, Self::Error>>> {
+            unimplemented!()
+        }
+    }
+
+    assert_impl_all!(AnyBody<()>: MessageBody, fmt::Debug, Send, Sync, Unpin);
+    assert_impl_all!(AnyBody<AnyBody<()>>: MessageBody, fmt::Debug, Send, Sync, Unpin);
+    assert_impl_all!(AnyBody<Bytes>: MessageBody, fmt::Debug, Send, Sync, Unpin);
+    assert_impl_all!(AnyBody: MessageBody, fmt::Debug, Unpin);
+    assert_impl_all!(AnyBody<PinType>: MessageBody);
+
+    assert_not_impl_all!(AnyBody: Send, Sync, Unpin);
+    assert_not_impl_all!(AnyBody<PinType>: Send, Sync, Unpin);
+}

awc/src/client/connection.rs

@@ -12,9 +12,9 @@ use bytes::Bytes;
 use futures_core::future::LocalBoxFuture;
 use h2::client::SendRequest;
 
-use actix_http::{
-    body::MessageBody, h1::ClientCodec, Error, Payload, RequestHeadType, ResponseHead,
-};
+use actix_http::{body::MessageBody, h1::ClientCodec, Payload, RequestHeadType, ResponseHead};
+
+use crate::BoxError;
 
 use super::error::SendRequestError;
 use super::pool::Acquired;
@@ -254,7 +254,7 @@ where
     where
         H: Into<RequestHeadType> + 'static,
         RB: MessageBody + 'static,
-        RB::Error: Into<Error>,
+        RB::Error: Into<BoxError>,
     {
         Box::pin(async move {
             match self {

awc/src/client/error.rs

@@ -1,13 +1,13 @@
-use std::{error::Error as StdError, fmt, io};
+use std::{fmt, io};
 
 use derive_more::{Display, From};
 
-use actix_http::{
-    error::{Error, ParseError},
-    http::Error as HttpError,
-};
+use actix_http::{error::ParseError, http::Error as HttpError};
+
 #[cfg(feature = "openssl")]
-use actix_tls::accept::openssl::reexports::Error as OpenSslError;
+use actix_tls::accept::openssl::reexports::Error as OpensslError;
+
+use crate::BoxError;
 
 /// A set of errors that can occur while connecting to an HTTP host
 #[derive(Debug, Display, From)]
@@ -20,7 +20,7 @@ pub enum ConnectError {
     /// SSL error
     #[cfg(feature = "openssl")]
     #[display(fmt = "{}", _0)]
-    SslError(OpenSslError),
+    SslError(OpensslError),
 
     /// Failed to resolve the hostname
     #[display(fmt = "Failed resolving hostname: {}", _0)]
@@ -118,11 +118,11 @@ pub enum SendRequestError {
     TunnelNotSupported,
 
     /// Error sending request body
-    Body(Error),
+    Body(BoxError),
 
     /// Other errors that can occur after submitting a request.
     #[display(fmt = "{:?}: {}", _1, _0)]
-    Custom(Box<dyn StdError>, Box<dyn fmt::Debug>),
+    Custom(BoxError, Box<dyn fmt::Debug>),
 }
 
 impl std::error::Error for SendRequestError {}
@@ -141,7 +141,7 @@ pub enum FreezeRequestError {
 
     /// Other errors that can occur after submitting a request.
     #[display(fmt = "{:?}: {}", _1, _0)]
-    Custom(Box<dyn StdError>, Box<dyn fmt::Debug>),
+    Custom(BoxError, Box<dyn fmt::Debug>),
 }
 
 impl std::error::Error for FreezeRequestError {}

awc/src/client/h1proto.rs

@@ -13,7 +13,7 @@ use actix_http::{
         header::{HeaderMap, IntoHeaderValue, EXPECT, HOST},
         StatusCode,
     },
-    Error, Payload, RequestHeadType, ResponseHead,
+    Payload, RequestHeadType, ResponseHead,
 };
 use actix_utils::future::poll_fn;
 use bytes::buf::BufMut;
@@ -22,6 +22,8 @@ use futures_core::{ready, Stream};
 use futures_util::SinkExt as _;
 use pin_project_lite::pin_project;
 
+use crate::BoxError;
+
 use super::connection::{ConnectionIo, H1Connection};
 use super::error::{ConnectError, SendRequestError};
 
@@ -33,7 +35,7 @@ pub(crate) async fn send_request<Io, B>(
 where
     Io: ConnectionIo,
     B: MessageBody,
-    B::Error: Into<Error>,
+    B::Error: Into<BoxError>,
 {
     // set request host header
     if !head.as_ref().headers.contains_key(HOST)
@@ -155,7 +157,7 @@ pub(crate) async fn send_body<Io, B>(
 where
     Io: ConnectionIo,
     B: MessageBody,
-    B::Error: Into<Error>,
+    B::Error: Into<BoxError>,
 {
     actix_rt::pin!(body);
 
@@ -166,7 +168,7 @@ where
                 Some(Ok(chunk)) => {
                     framed.as_mut().write(h1::Message::Chunk(Some(chunk)))?;
                 }
-                Some(Err(err)) => return Err(err.into().into()),
+                Some(Err(err)) => return Err(SendRequestError::Body(err.into())),
                 None => {
                     eof = true;
                     framed.as_mut().write(h1::Message::Chunk(None))?;

awc/src/client/h2proto.rs

@@ -13,9 +13,11 @@ use log::trace;
 use actix_http::{
     body::{BodySize, MessageBody},
     header::HeaderMap,
-    Error, Payload, RequestHeadType, ResponseHead,
+    Payload, RequestHeadType, ResponseHead,
 };
 
+use crate::BoxError;
+
 use super::{
     config::ConnectorConfig,
     connection::{ConnectionIo, H2Connection},
@@ -30,7 +32,7 @@ pub(crate) async fn send_request<Io, B>(
 where
     Io: ConnectionIo,
     B: MessageBody,
-    B::Error: Into<Error>,
+    B::Error: Into<BoxError>,
 {
     trace!("Sending client request: {:?} {:?}", head, body.size());
 
@@ -133,10 +135,12 @@ where
 async fn send_body<B>(body: B, mut send: SendStream<Bytes>) -> Result<(), SendRequestError>
 where
     B: MessageBody,
-    B::Error: Into<Error>,
+    B::Error: Into<BoxError>,
 {
     let mut buf = None;
+
     actix_rt::pin!(body);
+
     loop {
         if buf.is_none() {
             match poll_fn(|cx| body.as_mut().poll_next(cx)).await {
@@ -144,10 +148,10 @@ where
                     send.reserve_capacity(b.len());
                     buf = Some(b);
                 }
-                Some(Err(e)) => return Err(e.into().into()),
+                Some(Err(err)) => return Err(SendRequestError::Body(err.into())),
                 None => {
-                    if let Err(e) = send.send_data(Bytes::new(), true) {
-                        return Err(e.into());
+                    if let Err(err) = send.send_data(Bytes::new(), true) {
+                        return Err(err.into());
                     }
                     send.reserve_capacity(0);
                     return Ok(());

awc/src/connect.rs

@@ -7,16 +7,17 @@ use std::{
 };
 
 use actix_codec::Framed;
-use actix_http::{
-    body::AnyBody, h1::ClientCodec, Payload, RequestHead, RequestHeadType, ResponseHead,
-};
+use actix_http::{h1::ClientCodec, Payload, RequestHead, RequestHeadType, ResponseHead};
 use actix_service::Service;
 use futures_core::{future::LocalBoxFuture, ready};
 
-use crate::client::{
-    Connect as ClientConnect, ConnectError, Connection, ConnectionIo, SendRequestError,
+use crate::{
+    any_body::AnyBody,
+    client::{
+        Connect as ClientConnect, ConnectError, Connection, ConnectionIo, SendRequestError,
+    },
+    response::ClientResponse,
 };
-use crate::response::ClientResponse;
 
 pub type BoxConnectorService = Rc<
     dyn Service<

awc/src/error.rs

@@ -11,6 +11,8 @@ use serde_json::error::Error as JsonError;
 
 pub use crate::client::{ConnectError, FreezeRequestError, InvalidUrl, SendRequestError};
 
+// TODO: address display, error, and from impls
+
 /// Websocket client error
 #[derive(Debug, Display, From)]
 pub enum WsClientError {

awc/src/frozen.rs

@@ -1,18 +1,18 @@
-use std::{convert::TryFrom, error::Error as StdError, net, rc::Rc, time::Duration};
+use std::{convert::TryFrom, net, rc::Rc, time::Duration};
 
 use bytes::Bytes;
 use futures_core::Stream;
 use serde::Serialize;
 
 use actix_http::{
-    body::AnyBody,
     http::{header::IntoHeaderValue, Error as HttpError, HeaderMap, HeaderName, Method, Uri},
     RequestHead,
 };
 
 use crate::{
+    any_body::AnyBody,
     sender::{RequestSender, SendClientRequest},
-    ClientConfig,
+    BoxError, ClientConfig,
 };
 
 /// `FrozenClientRequest` struct represents cloneable client request.
@@ -82,7 +82,7 @@ impl FrozenClientRequest {
     pub fn send_stream<S, E>(&self, stream: S) -> SendClientRequest
     where
         S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
-        E: Into<Box<dyn StdError>> + 'static,
+        E: Into<BoxError> + 'static,
     {
         RequestSender::Rc(self.head.clone(), None).send_stream(
             self.addr,
@@ -207,7 +207,7 @@ impl FrozenSendBuilder {
     pub fn send_stream<S, E>(self, stream: S) -> SendClientRequest
     where
         S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
-        E: Into<Box<dyn StdError>> + 'static,
+        E: Into<BoxError> + 'static,
     {
         if let Some(e) = self.err {
             return e.into();

awc/src/lib.rs

@@ -104,6 +104,7 @@
 #![doc(html_logo_url = "https://actix.rs/img/logo.png")]
 #![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
 
+mod any_body;
 mod builder;
 mod client;
 mod connect;
@@ -139,6 +140,8 @@ use actix_service::Service;
 
 use self::client::{ConnectInfo, TcpConnectError, TcpConnection};
 
+pub(crate) type BoxError = Box<dyn std::error::Error>;
+
 /// An asynchronous HTTP and WebSocket client.
 ///
 /// You should take care to create, at most, one `Client` per thread. Otherwise, expect higher CPU

awc/src/middleware/redirect.rs

@@ -8,7 +8,6 @@ use std::{
 };
 
 use actix_http::{
-    body::AnyBody,
     http::{header, Method, StatusCode, Uri},
     RequestHead, RequestHeadType,
 };
@@ -17,10 +16,12 @@ use bytes::Bytes;
 use futures_core::ready;
 
 use super::Transform;
-
-use crate::client::{InvalidUrl, SendRequestError};
-use crate::connect::{ConnectRequest, ConnectResponse};
-use crate::ClientResponse;
+use crate::{
+    any_body::AnyBody,
+    client::{InvalidUrl, SendRequestError},
+    connect::{ConnectRequest, ConnectResponse},
+    ClientResponse,
+};
 
 pub struct Redirect {
     max_redirect_times: u8,
@@ -95,7 +96,7 @@ where
                 };
 
                 let body_opt = match body {
-                    AnyBody::Bytes(ref b) => Some(b.clone()),
+                    AnyBody::Bytes { ref body } => Some(body.clone()),
                     _ => None,
                 };
 
@@ -192,7 +193,9 @@ where
                         let body_new = if is_redirect {
                             // try to reuse body
                             match body {
-                                Some(ref bytes) => AnyBody::Bytes(bytes.clone()),
+                                Some(ref bytes) => AnyBody::Bytes {
+                                    body: bytes.clone(),
+                                },
                                 // TODO: should this be AnyBody::Empty or AnyBody::None.
                                 _ => AnyBody::empty(),
                             }

awc/src/request.rs

@@ -1,11 +1,10 @@
-use std::{convert::TryFrom, error::Error as StdError, fmt, net, rc::Rc, time::Duration};
+use std::{convert::TryFrom, fmt, net, rc::Rc, time::Duration};
 
 use bytes::Bytes;
 use futures_core::Stream;
 use serde::Serialize;
 
 use actix_http::{
-    body::AnyBody,
     http::{
         header::{self, IntoHeaderPair},
         ConnectionType, Error as HttpError, HeaderMap, HeaderValue, Method, Uri, Version,
@@ -13,15 +12,17 @@ use actix_http::{
     RequestHead,
 };
 
-#[cfg(feature = "cookies")]
-use crate::cookie::{Cookie, CookieJar};
 use crate::{
+    any_body::AnyBody,
     error::{FreezeRequestError, InvalidUrl},
     frozen::FrozenClientRequest,
     sender::{PrepForSendingError, RequestSender, SendClientRequest},
-    ClientConfig,
+    BoxError, ClientConfig,
 };
 
+#[cfg(feature = "cookies")]
+use crate::cookie::{Cookie, CookieJar};
+
 /// An HTTP Client request builder
 ///
 /// This type can be used to construct an instance of `ClientRequest` through a
@@ -404,7 +405,7 @@ impl ClientRequest {
     pub fn send_stream<S, E>(self, stream: S) -> SendClientRequest
     where
         S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
-        E: Into<Box<dyn StdError>> + 'static,
+        E: Into<BoxError> + 'static,
     {
         let slf = match self.prep_for_sending() {
             Ok(slf) => slf,

awc/src/sender.rs

@@ -1,5 +1,4 @@
 use std::{
-    error::Error as StdError,
     future::Future,
     net,
     pin::Pin,
@@ -9,12 +8,12 @@ use std::{
 };
 
 use actix_http::{
-    body::{AnyBody, BodyStream},
+    body::BodyStream,
     http::{
         header::{self, HeaderMap, HeaderName, IntoHeaderValue},
         Error as HttpError,
     },
-    Error, RequestHead, RequestHeadType,
+    RequestHead, RequestHeadType,
 };
 use actix_rt::time::{sleep, Sleep};
 use bytes::Bytes;
@@ -26,8 +25,9 @@ use serde::Serialize;
 use actix_http::{encoding::Decoder, http::header::ContentEncoding, Payload, PayloadStream};
 
 use crate::{
+    any_body::AnyBody,
     error::{FreezeRequestError, InvalidUrl, SendRequestError},
-    ClientConfig, ClientResponse, ConnectRequest, ConnectResponse,
+    BoxError, ClientConfig, ClientResponse, ConnectRequest, ConnectResponse,
 };
 
 #[derive(Debug, From)]
@@ -162,12 +162,6 @@ impl From<SendRequestError> for SendClientRequest {
     }
 }
 
-impl From<Error> for SendClientRequest {
-    fn from(e: Error) -> Self {
-        SendClientRequest::Err(Some(e.into()))
-    }
-}
-
 impl From<HttpError> for SendClientRequest {
     fn from(e: HttpError) -> Self {
         SendClientRequest::Err(Some(e.into()))
@@ -236,7 +230,9 @@ impl RequestSender {
             response_decompress,
             timeout,
             config,
-            AnyBody::Bytes(Bytes::from(body)),
+            AnyBody::Bytes {
+                body: Bytes::from(body),
+            },
         )
     }
 
@@ -265,7 +261,9 @@ impl RequestSender {
             response_decompress,
             timeout,
             config,
-            AnyBody::Bytes(Bytes::from(body)),
+            AnyBody::Bytes {
+                body: Bytes::from(body),
+            },
         )
     }
 
@@ -279,7 +277,7 @@ impl RequestSender {
     ) -> SendClientRequest
     where
         S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
-        E: Into<Box<dyn StdError>> + 'static,
+        E: Into<BoxError> + 'static,
     {
         self.send_body(
             addr,