restore 0.1 behavior

actix-threadpool/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [0.3.0] - 2019-12-02
+
+### Changed
+
+* Expect `Result` type as a function return type
+
 ## [0.2.0] - 2019-11-21
 
 ### Changed

actix-threadpool/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-threadpool"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix thread pool for sync code"
 keywords = ["actix", "network", "framework", "async", "futures"]
@@ -18,7 +18,7 @@ name = "actix_threadpool"
 path = "src/lib.rs"
 
 [dependencies]
-derive_more = "0.99"
+derive_more = "0.99.2"
 futures = "0.3.1"
 parking_lot = "0.9"
 lazy_static = "1.2"

actix-threadpool/src/lib.rs

@@ -1,15 +1,15 @@
 //! Thread pool for blocking operations
 
+use std::fmt;
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
+use derive_more::Display;
 use futures::channel::oneshot;
 use parking_lot::Mutex;
 use threadpool::ThreadPool;
 
-pub use futures::channel::oneshot::Canceled;
-
 /// Env variable for default cpu pool size.
 const ENV_CPU_POOL_VAR: &str = "ACTIX_THREADPOOL";
 
@@ -39,12 +39,22 @@ thread_local! {
     };
 }
 
+/// Blocking operation execution error
+#[derive(Debug, Display)]
+pub enum BlockingError<E: fmt::Debug> {
+    #[display(fmt = "{:?}", _0)]
+    Error(E),
+    #[display(fmt = "Thread pool is gone")]
+    Canceled,
+}
+
 /// Execute blocking function on a thread pool, returns future that resolves
 /// to result of the function execution.
-pub fn run<F, I>(f: F) -> CpuFuture<I>
+pub fn run<F, I, E>(f: F) -> CpuFuture<I, E>
 where
-    F: FnOnce() -> I + Send + 'static,
+    F: FnOnce() -> Result<I, E> + Send + 'static,
     I: Send + 'static,
+    E: Send + fmt::Debug + 'static,
 {
     let (tx, rx) = oneshot::channel();
     POOL.with(|pool| {
@@ -60,16 +70,18 @@ where
 
 /// Blocking operation completion future. It resolves with results
 /// of blocking function execution.
-pub struct CpuFuture<I> {
+pub struct CpuFuture<I, E> {
-    rx: oneshot::Receiver<I>,
+    rx: oneshot::Receiver<Result<I, E>>,
 }
 
-impl<I> Future for CpuFuture<I> {
+impl<I, E: fmt::Debug> Future for CpuFuture<I, E> {
-    type Output = Result<I, Canceled>;
+    type Output = Result<I, BlockingError<E>>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let rx = Pin::new(&mut Pin::get_mut(self).rx);
+        let rx = Pin::new(&mut self.rx);
-        let res = futures::ready!(rx.poll(cx));
+        let res = futures::ready!(rx.poll(cx))
-        Poll::Ready(res.map_err(|_| Canceled))
+            .map_err(|_| BlockingError::Canceled)
+            .and_then(|res| res.map_err(BlockingError::Error));
+        Poll::Ready(res)
     }
 }