relax unpin bounds on payload types

actix-http/CHANGES.md

@@ -3,8 +3,17 @@
 ## Unreleased - 2021-xx-xx
 ### Changes
 - `HeaderMap::get_all` now returns a `std::slice::Iter`. [#2527]
+- `Payload` inner fields are now named. [#????]
+- `impl Stream` for `Payload` no longer requires the `Stream` variant be `Unpin`. [#????]
+- `impl Future` for `h1::SendResponse` no longer requires the body type be `Unpin`. [#????]
+- `impl Stream` for `encoding::Decoder` no longer requires the stream type be `Unpin`. [#????]
+- Rename `PayloadStream` to `BoxedPayloadStream`. [#????]
+
+### Removed
+- `h1::Payload::readany`. [#????]
 
 [#2527]: https://github.com/actix/actix-web/pull/2527
+[#????]: https://github.com/actix/actix-web/pull/????
 
 
 ## 3.0.0-beta.16 - 2021-12-17

actix-http/src/encoding/decoder.rs

@@ -28,11 +28,14 @@ use crate::{
 
 const MAX_CHUNK_SIZE_DECODE_IN_PLACE: usize = 2049;
 
-pub struct Decoder<S> {
-    decoder: Option<ContentDecoder>,
-    stream: S,
-    eof: bool,
-    fut: Option<JoinHandle<Result<(Option<Bytes>, ContentDecoder), io::Error>>>,
+pin_project_lite::pin_project! {
+    pub struct Decoder<S> {
+        decoder: Option<ContentDecoder>,
+        #[pin]
+        stream: S,
+        eof: bool,
+        fut: Option<JoinHandle<Result<(Option<Bytes>, ContentDecoder), io::Error>>>,
+    }
 }
 
 impl<S> Decoder<S>
@@ -89,42 +92,44 @@ where
 
 impl<S> Stream for Decoder<S>
 where
-    S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
+    S: Stream<Item = Result<Bytes, PayloadError>>,
 {
     type Item = Result<Bytes, PayloadError>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
+
         loop {
-            if let Some(ref mut fut) = self.fut {
+            if let Some(ref mut fut) = this.fut {
                 let (chunk, decoder) =
                     ready!(Pin::new(fut).poll(cx)).map_err(|_| BlockingError)??;
 
-                self.decoder = Some(decoder);
-                self.fut.take();
+                *this.decoder = Some(decoder);
+                this.fut.take();
 
                 if let Some(chunk) = chunk {
                     return Poll::Ready(Some(Ok(chunk)));
                 }
             }
 
-            if self.eof {
+            if *this.eof {
                 return Poll::Ready(None);
             }
 
-            match ready!(Pin::new(&mut self.stream).poll_next(cx)) {
+            match ready!(this.stream.as_mut().poll_next(cx)) {
                 Some(Err(err)) => return Poll::Ready(Some(Err(err))),
 
                 Some(Ok(chunk)) => {
-                    if let Some(mut decoder) = self.decoder.take() {
+                    if let Some(mut decoder) = this.decoder.take() {
                         if chunk.len() < MAX_CHUNK_SIZE_DECODE_IN_PLACE {
                             let chunk = decoder.feed_data(chunk)?;
-                            self.decoder = Some(decoder);
+                            *this.decoder = Some(decoder);
 
                             if let Some(chunk) = chunk {
                                 return Poll::Ready(Some(Ok(chunk)));
                             }
                         } else {
-                            self.fut = Some(spawn_blocking(move || {
+                            *this.fut = Some(spawn_blocking(move || {
                                 let chunk = decoder.feed_data(chunk)?;
                                 Ok((chunk, decoder))
                             }));
@@ -137,9 +142,9 @@ where
                 }
 
                 None => {
-                    self.eof = true;
+                    *this.eof = true;
 
-                    return if let Some(mut decoder) = self.decoder.take() {
+                    return if let Some(mut decoder) = this.decoder.take() {
                         match decoder.feed_eof() {
                             Ok(Some(res)) => Poll::Ready(Some(Ok(res))),
                             Ok(None) => Poll::Ready(None),

actix-http/src/h1/dispatcher.rs

@@ -646,10 +646,11 @@ where
                                     Payload is attached to Request and passed to Service::call
                                     where the state can be collected and consumed.
                                     */
-                                    let (ps, pl) = Payload::create(false);
-                                    let (req1, _) = req.replace_payload(crate::Payload::H1(pl));
+                                    let (sender, payload) = Payload::create(false);
+                                    let (req1, _) =
+                                        req.replace_payload(crate::Payload::H1 { payload });
                                     req = req1;
-                                    *this.payload = Some(ps);
+                                    *this.payload = Some(sender);
                                 }
 
                                 // Request has no payload.

actix-http/src/h1/payload.rs

@@ -22,9 +22,9 @@ pub enum PayloadStatus {
 
 /// Buffered stream of bytes chunks
 ///
-/// Payload stores chunks in a vector. First chunk can be received with
-/// `.readany()` method. Payload stream is not thread safe. Payload does not
-/// notify current task when new data is available.
+/// Payload stores chunks in a vector. First chunk can be received with `.readany()` method.
+/// Payload stream is not thread safe. Payload does not notify current task when new data
+/// is available.
 ///
 /// Payload stream can be used as `Response` body stream.
 #[derive(Debug)]
@@ -77,14 +77,6 @@ impl Payload {
     pub fn unread_data(&mut self, data: Bytes) {
         self.inner.borrow_mut().unread_data(data);
     }
-
-    #[inline]
-    pub fn readany(
-        &mut self,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Bytes, PayloadError>>> {
-        self.inner.borrow_mut().readany(cx)
-    }
 }
 
 impl Stream for Payload {
@@ -257,8 +249,12 @@ impl Inner {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use actix_utils::future::poll_fn;
+    use static_assertions::assert_impl_all;
+
+    use super::*;
+
+    assert_impl_all!(Payload: Unpin);
 
     #[actix_rt::test]
     async fn test_unread_data() {
@@ -270,7 +266,10 @@ mod tests {
 
         assert_eq!(
             Bytes::from("data"),
-            poll_fn(|cx| payload.readany(cx)).await.unwrap().unwrap()
+            poll_fn(|cx| Pin::new(&mut payload).poll_next(cx))
+                .await
+                .unwrap()
+                .unwrap()
         );
     }
 }

actix-http/src/h1/utils.rs

@@ -45,7 +45,7 @@ where
 impl<T, B> Future for SendResponse<T, B>
 where
     T: AsyncRead + AsyncWrite + Unpin,
-    B: MessageBody + Unpin,
+    B: MessageBody,
     B::Error: Into<Error>,
 {
     type Output = Result<Framed<T, Codec>, Error>;
@@ -81,7 +81,7 @@ where
                             // body is done when item is None
                             body_done = item.is_none();
                             if body_done {
-                                let _ = this.body.take();
+                                this.body.set(None);
                             }
                             let framed = this.framed.as_mut().as_pin_mut().unwrap();
                             framed

actix-http/src/h2/dispatcher.rs

@@ -108,8 +108,8 @@ where
             match Pin::new(&mut this.connection).poll_accept(cx)? {
                 Poll::Ready(Some((req, tx))) => {
                     let (parts, body) = req.into_parts();
-                    let pl = crate::h2::Payload::new(body);
-                    let pl = Payload::H2(pl);
+                    let payload = crate::h2::Payload::new(body);
+                    let pl = Payload::H2 { payload };
                     let mut req = Request::with_payload(pl);
 
                     let head = req.head_mut();

actix-http/src/h2/mod.rs

@@ -98,3 +98,12 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use static_assertions::assert_impl_all;
+
+    use super::*;
+
+    assert_impl_all!(Payload: Unpin);
+}

actix-http/src/lib.rs

@@ -58,7 +58,8 @@ pub use self::header::ContentEncoding;
 pub use self::http_message::HttpMessage;
 pub use self::message::ConnectionType;
 pub use self::message::Message;
-pub use self::payload::{Payload, PayloadStream};
+#[allow(deprecated)]
+pub use self::payload::{BoxedPayloadStream, Payload, PayloadStream};
 pub use self::requests::{Request, RequestHead, RequestHeadType};
 pub use self::responses::{Response, ResponseBuilder, ResponseHead};
 pub use self::service::HttpService;

actix-http/src/payload.rs

@@ -1,4 +1,5 @@
 use std::{
+    mem,
     pin::Pin,
     task::{Context, Poll},
 };
@@ -9,62 +10,79 @@ use h2::RecvStream;
 
 use crate::error::PayloadError;
 
-// TODO: rename to boxed payload
-/// A boxed payload.
-pub type PayloadStream = Pin<Box<dyn Stream<Item = Result<Bytes, PayloadError>>>>;
+/// A boxed payload stream.
+pub type BoxedPayloadStream = Pin<Box<dyn Stream<Item = Result<Bytes, PayloadError>>>>;
 
-/// A streaming payload.
-pub enum Payload<S = PayloadStream> {
-    None,
-    H1(crate::h1::Payload),
-    H2(crate::h2::Payload),
-    Stream(S),
+#[deprecated(since = "4.0.0", note = "Renamed to `BoxedStream`.")]
+pub type PayloadStream = BoxedPayloadStream;
+
+pin_project_lite::pin_project! {
+    /// A streaming payload.
+    #[project = PayloadProj]
+    pub enum Payload<S = BoxedPayloadStream> {
+        None,
+        H1 { payload: crate::h1::Payload },
+        H2 { payload: crate::h2::Payload },
+        Stream { #[pin] payload: S },
+}
 }
 
 impl<S> From<crate::h1::Payload> for Payload<S> {
-    fn from(v: crate::h1::Payload) -> Self {
-        Payload::H1(v)
+    fn from(payload: crate::h1::Payload) -> Self {
+        Payload::H1 { payload }
     }
 }
 
 impl<S> From<crate::h2::Payload> for Payload<S> {
-    fn from(v: crate::h2::Payload) -> Self {
-        Payload::H2(v)
+    fn from(payload: crate::h2::Payload) -> Self {
+        Payload::H2 { payload }
     }
 }
 
 impl<S> From<RecvStream> for Payload<S> {
-    fn from(v: RecvStream) -> Self {
-        Payload::H2(crate::h2::Payload::new(v))
+    fn from(stream: RecvStream) -> Self {
+        Payload::H2 {
+            payload: crate::h2::Payload::new(stream),
+        }
     }
 }
 
-impl From<PayloadStream> for Payload {
-    fn from(pl: PayloadStream) -> Self {
-        Payload::Stream(pl)
+impl From<BoxedPayloadStream> for Payload {
+    fn from(payload: BoxedPayloadStream) -> Self {
+        Payload::Stream { payload }
     }
 }
 
 impl<S> Payload<S> {
     /// Takes current payload and replaces it with `None` value
     pub fn take(&mut self) -> Payload<S> {
-        std::mem::replace(self, Payload::None)
+        mem::replace(self, Payload::None)
     }
 }
 
 impl<S> Stream for Payload<S>
 where
-    S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
+    S: Stream<Item = Result<Bytes, PayloadError>>,
 {
     type Item = Result<Bytes, PayloadError>;
 
     #[inline]
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        match self.get_mut() {
-            Payload::None => Poll::Ready(None),
-            Payload::H1(ref mut pl) => pl.readany(cx),
-            Payload::H2(ref mut pl) => Pin::new(pl).poll_next(cx),
-            Payload::Stream(ref mut pl) => Pin::new(pl).poll_next(cx),
+        match self.project() {
+            PayloadProj::None => Poll::Ready(None),
+            PayloadProj::H1 { payload } => Pin::new(payload).poll_next(cx),
+            PayloadProj::H2 { payload } => Pin::new(payload).poll_next(cx),
+            PayloadProj::Stream { payload } => payload.poll_next(cx),
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use static_assertions::assert_impl_all;
+
+    use super::*;
+
+    assert_impl_all!(RecvStream: Unpin);
+    assert_impl_all!(Payload: Unpin);
+}

actix-http/src/requests/request.rs

@@ -10,11 +10,12 @@ use std::{
 use http::{header, Method, Uri, Version};
 
 use crate::{
-    header::HeaderMap, Extensions, HttpMessage, Message, Payload, PayloadStream, RequestHead,
+    header::HeaderMap, BoxedPayloadStream, Extensions, HttpMessage, Message, Payload,
+    RequestHead,
 };
 
 /// An HTTP request.
-pub struct Request<P = PayloadStream> {
+pub struct Request<P = BoxedPayloadStream> {
     pub(crate) payload: Payload<P>,
     pub(crate) head: Message<RequestHead>,
     pub(crate) conn_data: Option<Rc<Extensions>>,
@@ -46,7 +47,7 @@ impl<P> HttpMessage for Request<P> {
     }
 }
 
-impl From<Message<RequestHead>> for Request<PayloadStream> {
+impl From<Message<RequestHead>> for Request<BoxedPayloadStream> {
     fn from(head: Message<RequestHead>) -> Self {
         Request {
             head,
@@ -57,10 +58,10 @@ impl From<Message<RequestHead>> for Request<PayloadStream> {
     }
 }
 
-impl Request<PayloadStream> {
+impl Request<BoxedPayloadStream> {
     /// Create new Request instance
     #[allow(clippy::new_without_default)]
-    pub fn new() -> Request<PayloadStream> {
+    pub fn new() -> Request<BoxedPayloadStream> {
         Request {
             head: Message::new(),
             payload: Payload::None,

actix-http/tests/test_rustls.rs

@@ -7,6 +7,7 @@ use std::{
     io::{self, BufReader, Write},
     net::{SocketAddr, TcpStream as StdTcpStream},
     sync::Arc,
+    task::Poll,
 };
 
 use actix_http::{
@@ -16,25 +17,37 @@ use actix_http::{
     Error, HttpService, Method, Request, Response, StatusCode, Version,
 };
 use actix_http_test::test_server;
+use actix_rt::pin;
 use actix_service::{fn_factory_with_config, fn_service};
 use actix_tls::connect::rustls::webpki_roots_cert_store;
-use actix_utils::future::{err, ok};
+use actix_utils::future::{err, ok, poll_fn};
 use bytes::{Bytes, BytesMut};
 use derive_more::{Display, Error};
-use futures_core::Stream;
-use futures_util::stream::{once, StreamExt as _};
+use futures_core::{ready, Stream};
+use futures_util::stream::once;
 use rustls::{Certificate, PrivateKey, ServerConfig as RustlsServerConfig, ServerName};
 use rustls_pemfile::{certs, pkcs8_private_keys};
 
-async fn load_body<S>(mut stream: S) -> Result<BytesMut, PayloadError>
+async fn load_body<S>(stream: S) -> Result<BytesMut, PayloadError>
 where
-    S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
+    S: Stream<Item = Result<Bytes, PayloadError>>,
 {
-    let mut body = BytesMut::new();
-    while let Some(item) = stream.next().await {
-        body.extend_from_slice(&item?)
-    }
-    Ok(body)
+    let mut buf = BytesMut::new();
+
+    pin!(stream);
+
+    poll_fn(|cx| loop {
+        let body = stream.as_mut();
+
+        match ready!(body.poll_next(cx)) {
+            Some(Ok(bytes)) => buf.extend_from_slice(&*bytes),
+            None => return Poll::Ready(Ok(())),
+            Some(Err(err)) => return Poll::Ready(Err(err)),
+        }
+    })
+    .await?;
+
+    Ok(buf)
 }
 
 fn tls_config() -> RustlsServerConfig {

actix-multipart/src/server.rs

@@ -1233,7 +1233,7 @@ mod tests {
 
         // and should not consume the payload
         match payload {
-            actix_web::dev::Payload::H1(_) => {} //expected
+            actix_web::dev::Payload::H1 { .. } => {} //expected
             _ => unreachable!(),
         }
     }

awc/src/client/connection.rs

@@ -267,7 +267,9 @@ where
                 Connection::Tls(ConnectionType::H2(conn)) => {
                     h2proto::send_request(conn, head.into(), body).await
                 }
-                _ => unreachable!("Plain Tcp connection can be used only in Http1 protocol"),
+                _ => {
+                    unreachable!("Plain TCP connection can be used only with HTTP/1.1 protocol")
+                }
             }
         })
     }

awc/src/client/h1proto.rs

@@ -123,7 +123,12 @@ where
 
             Ok((head, Payload::None))
         }
-        _ => Ok((head, Payload::Stream(Box::pin(PlStream::new(framed))))),
+        _ => Ok((
+            head,
+            Payload::Stream {
+                payload: Box::pin(PlStream::new(framed)),
+            },
+        )),
     }
 }
 

awc/src/response.rs

@@ -10,8 +10,8 @@ use std::{
 };
 
 use actix_http::{
-    error::PayloadError, header, header::HeaderMap, Extensions, HttpMessage, Payload,
-    PayloadStream, ResponseHead, StatusCode, Version,
+    error::PayloadError, header, header::HeaderMap, BoxedPayloadStream, Extensions,
+    HttpMessage, Payload, ResponseHead, StatusCode, Version,
 };
 use actix_rt::time::{sleep, Sleep};
 use bytes::{Bytes, BytesMut};
@@ -23,7 +23,7 @@ use crate::cookie::{Cookie, ParseError as CookieParseError};
 use crate::error::JsonPayloadError;
 
 /// Client Response
-pub struct ClientResponse<S = PayloadStream> {
+pub struct ClientResponse<S = BoxedPayloadStream> {
     pub(crate) head: ResponseHead,
     pub(crate) payload: Payload<S>,
     pub(crate) timeout: ResponseTimeout,

awc/src/sender.rs

@@ -20,7 +20,7 @@ use futures_core::Stream;
 use serde::Serialize;
 
 #[cfg(feature = "__compress")]
-use actix_http::{encoding::Decoder, header::ContentEncoding, Payload, PayloadStream};
+use actix_http::{encoding::Decoder, header::ContentEncoding, Payload};
 
 use crate::{
     any_body::AnyBody,
@@ -91,7 +91,7 @@ impl SendClientRequest {
 
 #[cfg(feature = "__compress")]
 impl Future for SendClientRequest {
-    type Output = Result<ClientResponse<Decoder<Payload<PayloadStream>>>, SendRequestError>;
+    type Output = Result<ClientResponse<Decoder<Payload>>, SendRequestError>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.get_mut();
@@ -108,12 +108,13 @@ impl Future for SendClientRequest {
                     res.into_client_response()._timeout(delay.take()).map_body(
                         |head, payload| {
                             if *response_decompress {
-                                Payload::Stream(Decoder::from_headers(payload, &head.headers))
+                                Payload::Stream {
+                                    payload: Decoder::from_headers(payload, &head.headers),
+                                }
                             } else {
-                                Payload::Stream(Decoder::new(
-                                    payload,
-                                    ContentEncoding::Identity,
-                                ))
+                                Payload::Stream {
+                                    payload: Decoder::new(payload, ContentEncoding::Identity),
+                                }
                             }
                         },
                     )

src/dev.rs

@@ -14,7 +14,7 @@ pub use crate::types::form::UrlEncoded;
 pub use crate::types::json::JsonBody;
 pub use crate::types::readlines::Readlines;
 
-pub use actix_http::{Extensions, Payload, PayloadStream, RequestHead, Response, ResponseHead};
+pub use actix_http::{Extensions, Payload, RequestHead, Response, ResponseHead};
 pub use actix_router::{Path, ResourceDef, ResourcePath, Url};
 pub use actix_server::{Server, ServerHandle};
 pub use actix_service::{

src/service.rs

@@ -7,7 +7,7 @@ use std::{
 use actix_http::{
     body::{BoxBody, EitherBody, MessageBody},
     header::HeaderMap,
-    Extensions, HttpMessage, Method, Payload, PayloadStream, RequestHead, Response,
+    BoxedPayloadStream, Extensions, HttpMessage, Method, Payload, RequestHead, Response,
     ResponseHead, StatusCode, Uri, Version,
 };
 use actix_router::{IntoPatterns, Path, Patterns, Resource, ResourceDef, Url};
@@ -293,7 +293,7 @@ impl Resource<Url> for ServiceRequest {
 }
 
 impl HttpMessage for ServiceRequest {
-    type Stream = PayloadStream;
+    type Stream = BoxedPayloadStream;
 
     #[inline]
     /// Returns Request's headers.