refined error model (#2253)

actix-http/src/body/body_stream.rs

@@ -1,4 +1,5 @@
 use std::{
+    error::Error as StdError,
     pin::Pin,
     task::{Context, Poll},
 };
@@ -7,8 +8,6 @@ use bytes::Bytes;
 use futures_core::{ready, Stream};
 use pin_project_lite::pin_project;
 
-use crate::error::Error;
-
 use super::{BodySize, MessageBody};
 
 pin_project! {
@@ -24,7 +23,7 @@ pin_project! {
 impl<S, E> BodyStream<S>
 where
     S: Stream<Item = Result<Bytes, E>>,
-    E: Into<Error>,
+    E: Into<Box<dyn StdError>> + 'static,
 {
     pub fn new(stream: S) -> Self {
         BodyStream { stream }
@@ -34,9 +33,9 @@ where
 impl<S, E> MessageBody for BodyStream<S>
 where
     S: Stream<Item = Result<Bytes, E>>,
-    E: Into<Error>,
+    E: Into<Box<dyn StdError>> + 'static,
 {
-    type Error = Error;
+    type Error = E;
 
     fn size(&self) -> BodySize {
         BodySize::Stream
@@ -56,7 +55,7 @@ where
 
             let chunk = match ready!(stream.poll_next(cx)) {
                 Some(Ok(ref bytes)) if bytes.is_empty() => continue,
-                opt => opt.map(|res| res.map_err(Into::into)),
+                opt => opt,
             };
 
             return Poll::Ready(chunk);
@@ -66,9 +65,16 @@ where
 
 #[cfg(test)]
 mod tests {
-    use actix_rt::pin;
+    use std::{convert::Infallible, time::Duration};
+
+    use actix_rt::{
+        pin,
+        time::{sleep, Sleep},
+    };
     use actix_utils::future::poll_fn;
-    use futures_util::stream;
+    use derive_more::{Display, Error};
+    use futures_core::ready;
+    use futures_util::{stream, FutureExt as _};
 
     use super::*;
     use crate::body::to_bytes;
@@ -78,7 +84,7 @@ mod tests {
         let body = BodyStream::new(stream::iter(
             ["1", "", "2"]
                 .iter()
-                .map(|&v| Ok(Bytes::from(v)) as Result<Bytes, ()>),
+                .map(|&v| Ok::<_, Infallible>(Bytes::from(v))),
         ));
         pin!(body);
 
@@ -103,9 +109,63 @@ mod tests {
         let body = BodyStream::new(stream::iter(
             ["1", "", "2"]
                 .iter()
-                .map(|&v| Ok(Bytes::from(v)) as Result<Bytes, ()>),
+                .map(|&v| Ok::<_, Infallible>(Bytes::from(v))),
         ));
 
         assert_eq!(to_bytes(body).await.ok(), Some(Bytes::from("12")));
     }
+    #[derive(Debug, Display, Error)]
+    #[display(fmt = "stream error")]
+    struct StreamErr;
+
+    #[actix_rt::test]
+    async fn stream_immediate_error() {
+        let body = BodyStream::new(stream::once(async { Err(StreamErr) }));
+        assert!(matches!(to_bytes(body).await, Err(StreamErr)));
+    }
+
+    #[actix_rt::test]
+    async fn stream_delayed_error() {
+        let body =
+            BodyStream::new(stream::iter(vec![Ok(Bytes::from("1")), Err(StreamErr)]));
+        assert!(matches!(to_bytes(body).await, Err(StreamErr)));
+
+        #[pin_project::pin_project(project = TimeDelayStreamProj)]
+        #[derive(Debug)]
+        enum TimeDelayStream {
+            Start,
+            Sleep(Pin<Box<Sleep>>),
+            Done,
+        }
+
+        impl Stream for TimeDelayStream {
+            type Item = Result<Bytes, StreamErr>;
+
+            fn poll_next(
+                mut self: Pin<&mut Self>,
+                cx: &mut Context<'_>,
+            ) -> Poll<Option<Self::Item>> {
+                match self.as_mut().get_mut() {
+                    TimeDelayStream::Start => {
+                        let sleep = sleep(Duration::from_millis(1));
+                        self.as_mut().set(TimeDelayStream::Sleep(Box::pin(sleep)));
+                        cx.waker().wake_by_ref();
+                        Poll::Pending
+                    }
+
+                    TimeDelayStream::Sleep(ref mut delay) => {
+                        ready!(delay.poll_unpin(cx));
+                        self.set(TimeDelayStream::Done);
+                        cx.waker().wake_by_ref();
+                        Poll::Pending
+                    }
+
+                    TimeDelayStream::Done => Poll::Ready(Some(Err(StreamErr))),
+                }
+            }
+        }
+
+        let body = BodyStream::new(TimeDelayStream::Start);
+        assert!(matches!(to_bytes(body).await, Err(StreamErr)));
+    }
 }