propagate into<error> change

actix-http/src/body/body.rs

@@ -72,6 +72,8 @@ impl MessageBody for Body {
                     Poll::Ready(Some(Ok(mem::take(bin))))
                 }
             }
+
+            // TODO: MSRV 1.51: poll_map_err
             Body::Message(body) => match ready!(body.as_mut().poll_next(cx)) {
                 Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
                 Some(Ok(val)) => Poll::Ready(Some(Ok(val))),

actix-http/src/body/message_body.rs

@@ -45,7 +45,8 @@ impl MessageBody for () {
 
 impl<T> MessageBody for Box<T>
 where
-    T: MessageBody<Error = Error> + Unpin,
+    T: MessageBody + Unpin,
+    T::Error: Into<Error>,
 {
     type Error = Error;
 
@@ -57,11 +58,19 @@ where
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Bytes, Self::Error>>> {
-        Pin::new(self.get_mut().as_mut()).poll_next(cx)
+        match ready!(Pin::new(self.get_mut().as_mut()).poll_next(cx)) {
+            Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
+            Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+            None => Poll::Ready(None),
+        }
     }
 }
 
-impl<T: MessageBody> MessageBody for Pin<Box<T>> {
+impl<T> MessageBody for Pin<Box<T>>
+where
+    T: MessageBody,
+    T::Error: Into<Error>,
+{
     type Error = Error;
 
     fn size(&self) -> BodySize {
@@ -72,7 +81,11 @@ impl<T: MessageBody> MessageBody for Pin<Box<T>> {
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Bytes, Self::Error>>> {
-        self.as_mut().poll_next(cx)
+        match ready!(self.as_mut().poll_next(cx)) {
+            Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
+            Some(Ok(val)) => Poll::Ready(Some(Ok(val))),
+            None => Poll::Ready(None),
+        }
     }
 }
 

actix-http/src/builder.rs

@@ -208,7 +208,7 @@ where
         S::Response: Into<Response<B>> + 'static,
 
         B: MessageBody + 'static,
-        B: MessageBody<Error = Error>,
+        B::Error: Into<Error>,
     {
         let cfg = ServiceConfig::new(
             self.keep_alive,
@@ -225,11 +225,13 @@ where
     /// Finish service configuration and create `HttpService` instance.
     pub fn finish<F, B>(self, service: F) -> HttpService<T, S, B, X, U>
     where
-        B: MessageBody + 'static,
         F: IntoServiceFactory<S, Request>,
         S::Error: Into<Error> + 'static,
         S::InitError: fmt::Debug,
         S::Response: Into<Response<B>> + 'static,
+
+        B: MessageBody + 'static,
+        B::Error: Into<Error>,
     {
         let cfg = ServiceConfig::new(
             self.keep_alive,

actix-http/src/client/connection.rs

@@ -257,7 +257,8 @@ where
     ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>
     where
         H: Into<RequestHeadType> + 'static,
-        RB: MessageBody<Error = Error> + 'static,
+        RB: MessageBody + 'static,
+        RB::Error: Into<Error>,
     {
         Box::pin(async move {
             match self {

actix-http/src/client/h1proto.rs

@@ -31,7 +31,8 @@ pub(crate) async fn send_request<Io, B>(
 ) -> Result<(ResponseHead, Payload), SendRequestError>
 where
     Io: ConnectionIo,
-    B: MessageBody<Error = Error>,
+    B: MessageBody,
+    B::Error: Into<Error>,
 {
     // set request host header
     if !head.as_ref().headers.contains_key(HOST)
@@ -153,7 +154,8 @@ pub(crate) async fn send_body<Io, B>(
 ) -> Result<(), SendRequestError>
 where
     Io: ConnectionIo,
-    B: MessageBody<Error = Error>,
+    B: MessageBody,
+    B::Error: Into<Error>,
 {
     actix_rt::pin!(body);
 
@@ -161,9 +163,10 @@ where
     while !eof {
         while !eof && !framed.as_ref().is_write_buf_full() {
             match poll_fn(|cx| body.as_mut().poll_next(cx)).await {
-                Some(result) => {
+                Some(Ok(chunk)) => {
-                    framed.as_mut().write(h1::Message::Chunk(Some(result?)))?;
+                    framed.as_mut().write(h1::Message::Chunk(Some(chunk)))?;
                 }
+                Some(Err(err)) => return Err(err.into().into()),
                 None => {
                     eof = true;
                     framed.as_mut().write(h1::Message::Chunk(None))?;

actix-http/src/client/h2proto.rs

@@ -30,7 +30,8 @@ pub(crate) async fn send_request<Io, B>(
 ) -> Result<(ResponseHead, Payload), SendRequestError>
 where
     Io: ConnectionIo,
-    B: MessageBody<Error = Error>,
+    B: MessageBody,
+    B::Error: Into<Error>,
 {
     trace!("Sending client request: {:?} {:?}", head, body.size());
 
@@ -135,7 +136,8 @@ async fn send_body<B>(
     mut send: SendStream<Bytes>,
 ) -> Result<(), SendRequestError>
 where
-    B: MessageBody<Error = Error>,
+    B: MessageBody,
+    B::Error: Into<Error>,
 {
     let mut buf = None;
     actix_rt::pin!(body);
@@ -146,7 +148,7 @@ where
                     send.reserve_capacity(b.len());
                     buf = Some(b);
                 }
-                Some(Err(e)) => return Err(e.into()),
+                Some(Err(e)) => return Err(e.into().into()),
                 None => {
                     if let Err(e) = send.send_data(Bytes::new(), true) {
                         return Err(e.into());

actix-http/src/h1/service.rs

@@ -66,7 +66,7 @@ where
     S::Response: Into<Response<B>>,
 
     B: MessageBody,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
@@ -115,7 +115,7 @@ mod openssl {
         S::Response: Into<Response<B>>,
 
         B: MessageBody,
-        B: MessageBody<Error = Error>,
+        B::Error: Into<Error>,
 
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
@@ -175,7 +175,7 @@ mod rustls {
         S::Response: Into<Response<B>>,
 
         B: MessageBody,
-        B: MessageBody<Error = Error>,
+        B::Error: Into<Error>,
 
         X: ServiceFactory<Request, Config = (), Response = Request>,
         X::Future: 'static,
@@ -273,7 +273,7 @@ where
     S::InitError: fmt::Debug,
 
     B: MessageBody,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 
     X: ServiceFactory<Request, Config = (), Response = Request>,
     X::Future: 'static,
@@ -342,7 +342,7 @@ where
     S::Response: Into<Response<B>>,
 
     B: MessageBody,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 
     X: Service<Request, Response = Request>,
     X::Error: Into<Error>,

actix-http/src/h1/utils.rs

@@ -22,7 +22,7 @@ pub struct SendResponse<T, B> {
 impl<T, B> SendResponse<T, B>
 where
     B: MessageBody,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 {
     pub fn new(framed: Framed<T, Codec>, response: Response<B>) -> Self {
         let (res, body) = response.into_parts();
@@ -39,7 +39,7 @@ impl<T, B> Future for SendResponse<T, B>
 where
     T: AsyncRead + AsyncWrite + Unpin,
     B: MessageBody + Unpin,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 {
     type Output = Result<Framed<T, Codec>, Error>;
 

actix-http/src/h2/dispatcher.rs

@@ -76,7 +76,7 @@ where
     S::Response: Into<Response<B>> + 'static,
 
     B: MessageBody + 'static,
-    B::Error: Into<Error> + 'static,
+    B::Error: Into<Error>,
 {
     type Output = Result<(), DispatchError>;
 

actix-http/src/h2/service.rs

@@ -42,7 +42,7 @@ where
     <S::Service as Service<Request>>::Future: 'static,
 
     B: MessageBody + 'static,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 {
     /// Create new `H2Service` instance with config.
     pub(crate) fn with_config<F: IntoServiceFactory<S, Request>>(
@@ -73,7 +73,7 @@ where
     <S::Service as Service<Request>>::Future: 'static,
 
     B: MessageBody + 'static,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 {
     /// Create plain TCP based service
     pub fn tcp(
@@ -112,7 +112,7 @@ mod openssl {
         <S::Service as Service<Request>>::Future: 'static,
 
         B: MessageBody + 'static,
-        B: MessageBody<Error = Error>,
+        B::Error: Into<Error>,
     {
         /// Create OpenSSL based service
         pub fn openssl(
@@ -158,7 +158,7 @@ mod rustls {
         <S::Service as Service<Request>>::Future: 'static,
 
         B: MessageBody + 'static,
-        B: MessageBody<Error = Error>,
+        B::Error: Into<Error>,
     {
         /// Create Rustls based service
         pub fn rustls(
@@ -201,7 +201,7 @@ where
     <S::Service as Service<Request>>::Future: 'static,
 
     B: MessageBody + 'static,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 {
     type Response = ();
     type Error = DispatchError;
@@ -263,7 +263,7 @@ where
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
     B: MessageBody + 'static,
-    B: MessageBody<Error = Error>,
+    B::Error: Into<Error>,
 {
     type Response = ();
     type Error = DispatchError;
@@ -327,7 +327,8 @@ where
     S::Error: Into<Error> + 'static,
     S::Future: 'static,
     S::Response: Into<Response<B>> + 'static,
-    B: MessageBody<Error = Error>,
+    B: MessageBody,
+    B::Error: Into<Error>,
 {
     type Output = Result<(), DispatchError>;
 

actix-http/src/response.rs

@@ -244,7 +244,8 @@ impl<B> Response<B> {
 
 impl<B> fmt::Debug for Response<B>
 where
-    B: MessageBody<Error = Error>,
+    B: MessageBody,
+    B::Error: Into<Error>,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let res = writeln!(

actix-http/src/service.rs

@@ -59,6 +59,7 @@ where
     S::Response: Into<Response<B>> + 'static,
     <S::Service as Service<Request>>::Future: 'static,
     B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     /// Create new `HttpService` instance.
     pub fn new<F: IntoServiceFactory<S, Request>>(service: F) -> Self {

src/middleware/compat.rs

@@ -113,8 +113,10 @@ pub trait MapServiceResponseBody {
     fn map_body(self) -> ServiceResponse;
 }
 
-impl<B: MessageBody<Error = Error> + Unpin + 'static> MapServiceResponseBody
+impl<B> MapServiceResponseBody for ServiceResponse<B>
-    for ServiceResponse<B>
+where
+    B: MessageBody + Unpin + 'static,
+    B::Error: Into<Error>,
 {
     fn map_body(self) -> ServiceResponse {
         self.map_body(|_, body| ResponseBody::Other(Body::from_message(body)))

src/middleware/logger.rs

@@ -326,7 +326,11 @@ impl<B> PinnedDrop for StreamLog<B> {
     }
 }
 
-impl<B: MessageBody<Error = Error>> MessageBody for StreamLog<B> {
+impl<B> MessageBody for StreamLog<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
     type Error = Error;
 
     fn size(&self) -> BodySize {

src/response/response.rs

@@ -243,7 +243,11 @@ impl<B> HttpResponse<B> {
     }
 }
 
-impl<B: MessageBody<Error = Error>> fmt::Debug for HttpResponse<B> {
+impl<B> fmt::Debug for HttpResponse<B>
+where
+    B: MessageBody,
+    B::Error: Into<Error>,
+{
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("HttpResponse")
             .field("error", &self.error)

src/server.rs

@@ -80,7 +80,8 @@ where
     <S::Service as Service<Request>>::Future: 'static,
     S::Service: 'static,
     // S::Service: 'static,
-    B: MessageBody<Error = Error> + 'static,
+    B: MessageBody + 'static,
+    B::Error: Into<Error>,
 {
     /// Create new HTTP server with application factory
     pub fn new(factory: F) -> Self {

src/service.rs

@@ -445,7 +445,8 @@ impl<B> From<ServiceResponse<B>> for Response<B> {
 
 impl<B> fmt::Debug for ServiceResponse<B>
 where
-    B: MessageBody<Error = Error>,
+    B: MessageBody,
+    B::Error: Into<Error>,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let res = writeln!(

src/test.rs

@@ -150,7 +150,8 @@ where
 pub async fn read_response<S, B>(app: &S, req: Request) -> Bytes
 where
     S: Service<Request, Response = ServiceResponse<B>, Error = Error>,
-    B: MessageBody<Error = Error> + Unpin,
+    B: MessageBody + Unpin,
+    B::Error: Into<Error>,
 {
     let mut resp = app
         .call(req)
@@ -195,7 +196,8 @@ where
 /// ```
 pub async fn read_body<B>(mut res: ServiceResponse<B>) -> Bytes
 where
-    B: MessageBody<Error = Error> + Unpin,
+    B: MessageBody + Unpin,
+    B::Error: Into<Error>,
 {
     let mut body = res.take_body();
     let mut bytes = BytesMut::new();
@@ -244,7 +246,8 @@ where
 /// ```
 pub async fn read_body_json<T, B>(res: ServiceResponse<B>) -> T
 where
-    B: MessageBody<Error = Error> + Unpin,
+    B: MessageBody + Unpin,
+    B::Error: Into<Error>,
     T: DeserializeOwned,
 {
     let body = read_body(res).await;
@@ -305,7 +308,8 @@ where
 pub async fn read_response_json<S, B, T>(app: &S, req: Request) -> T
 where
     S: Service<Request, Response = ServiceResponse<B>, Error = Error>,
-    B: MessageBody<Error = Error> + Unpin,
+    B: MessageBody + Unpin,
+    B::Error: Into<Error>,
     T: DeserializeOwned,
 {
     let body = read_response(app, req).await;