use released versions of actix-net

actix-http/src/client/connector.rs

@@ -6,10 +6,10 @@ use actix_codec::{AsyncRead, AsyncWrite};
 use actix_connect::{
     default_connector, Connect as TcpConnect, Connection as TcpConnection,
 };
+use actix_rt::net::TcpStream;
 use actix_service::{apply_fn, Service};
 use actix_utils::timeout::{TimeoutError, TimeoutService};
 use http::Uri;
-use tokio_net::tcp::TcpStream;
 
 use super::connection::Connection;
 use super::error::ConnectError;

actix-http/src/encoding/decoder.rs

@@ -21,7 +21,7 @@ pub struct Decoder<S> {
     decoder: Option<ContentDecoder>,
     stream: S,
     eof: bool,
-    fut: Option<CpuFuture<Result<(Option<Bytes>, ContentDecoder), io::Error>>>,
+    fut: Option<CpuFuture<(Option<Bytes>, ContentDecoder), io::Error>>,
 }
 
 impl<S> Decoder<S>
@@ -85,8 +85,7 @@ where
         loop {
             if let Some(ref mut fut) = self.fut {
                 let (chunk, decoder) = match ready!(Pin::new(fut).poll(cx)) {
-                    Ok(Ok(item)) => item,
-                    Ok(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
+                    Ok(item) => item,
                     Err(e) => return Poll::Ready(Some(Err(e.into()))),
                 };
                 self.decoder = Some(decoder);

actix-http/src/encoding/encoder.rs

@@ -24,7 +24,7 @@ pub struct Encoder<B> {
     eof: bool,
     body: EncoderBody<B>,
     encoder: Option<ContentEncoder>,
-    fut: Option<CpuFuture<Result<ContentEncoder, io::Error>>>,
+    fut: Option<CpuFuture<ContentEncoder, io::Error>>,
 }
 
 impl<B: MessageBody> Encoder<B> {
@@ -104,8 +104,7 @@ impl<B: MessageBody> MessageBody for Encoder<B> {
 
             if let Some(ref mut fut) = self.fut {
                 let mut encoder = match futures::ready!(Pin::new(fut).poll(cx)) {
-                    Ok(Ok(item)) => item,
-                    Ok(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
+                    Ok(item) => item,
                     Err(e) => return Poll::Ready(Some(Err(e.into()))),
                 };
                 let chunk = encoder.take();

actix-http/src/error.rs

@@ -6,6 +6,7 @@ use std::str::Utf8Error;
 use std::string::FromUtf8Error;
 use std::{fmt, io, result};
 
+pub use actix_threadpool::BlockingError;
 use actix_utils::timeout::TimeoutError;
 use bytes::BytesMut;
 use derive_more::{Display, From};
@@ -197,6 +198,9 @@ impl ResponseError for DeError {
 /// `InternalServerError` for `Canceled`
 impl ResponseError for Canceled {}
 
+/// `InternalServerError` for `BlockingError`
+impl<E: fmt::Debug> ResponseError for BlockingError<E> {}
+
 /// Return `BAD_REQUEST` for `Utf8Error`
 impl ResponseError for Utf8Error {
     fn status_code(&self) -> StatusCode {
@@ -359,12 +363,15 @@ impl From<io::Error> for PayloadError {
     }
 }
 
-impl From<Canceled> for PayloadError {
-    fn from(_: Canceled) -> Self {
-        PayloadError::Io(io::Error::new(
-            io::ErrorKind::Other,
-            "Operation is canceled",
-        ))
+impl From<BlockingError<io::Error>> for PayloadError {
+    fn from(err: BlockingError<io::Error>) -> Self {
+        match err {
+            BlockingError::Error(e) => PayloadError::Io(e),
+            BlockingError::Canceled => PayloadError::Io(io::Error::new(
+                io::ErrorKind::Other,
+                "Operation is canceled",
+            )),
+        }
     }
 }