Implement BlockingQueue.take_result()/poll_result() as Future. Avoid dead locking of sahred_thread_pool()

Author: johnteee

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fp_rust"
-version = "0.2.4"
+version = "0.2.5"
 license = "MIT"
 authors = ["JunYi JohnTeee Lee <johnteee@gmail.com>"]
 edition = "2018"

src/monadio.rs

@@ -62,12 +62,14 @@ impl<Y: 'static + Send + Sync + Clone> MonadIO<Y> {
     pub async fn to_future(&self) -> Result<Arc<Y>, Box<dyn Error>> {
         // let mio = self.map(|y| y);
         let mio = self.clone();
-        let result = shared_thread_pool()
-            .inner
-            .lock()
-            .unwrap()
-            .spawn_with_handle(async move { mio.eval() })?
-            .await;
+        let future = {
+            shared_thread_pool()
+                .inner
+                .lock()
+                .unwrap()
+                .spawn_with_handle(async move { mio.eval() })?
+        };
+        let result = future.await;
         Ok(result)
     }
 }

src/sync.rs

@@ -5,10 +5,18 @@ of general async handling features.
 
 use std::sync::{
     atomic::{AtomicBool, Ordering},
-    mpsc, Arc, Condvar, Mutex,
+    mpsc,
+    mpsc::RecvTimeoutError,
+    Arc, Condvar, Mutex,
 };
 use std::time::Duration;
 
+#[cfg(feature = "for_futures")]
+use super::common::shared_thread_pool;
+#[cfg(feature = "for_futures")]
+use crate::futures::task::SpawnExt;
+// #[cfg(feature = "for_futures")]
+use std::error::Error;
 #[cfg(feature = "for_futures")]
 use std::future::Future;
 #[cfg(feature = "for_futures")]
@@ -412,18 +420,16 @@ where
     }
 
     fn poll(&mut self) -> Option<T> {
-        if !self.is_alive() {
-            return None::<T>;
-        }
-
-        {
-            let result = self.blocking_recever.lock().unwrap().try_recv();
+        let result = self.poll_result();
 
-            if self.panic && result.is_err() {
-                std::panic::panic_any(result.err());
-            }
+        if self.panic && result.is_err() {
+            std::panic::panic_any(result.err());
+            // return None;
+        }
 
-            result.ok()
+        match result {
+            Ok(v) => Some(v),
+            Err(_) => None,
         }
     }
 
@@ -434,34 +440,102 @@ where
     }
 
     fn take(&mut self) -> Option<T> {
+        let result = self.take_result();
+
+        if self.panic && result.is_err() {
+            std::panic::panic_any(result.err());
+            // return None;
+        }
+
+        match result {
+            Ok(v) => Some(v),
+            Err(_) => None,
+        }
+    }
+}
+
+impl<T> BlockingQueue<T>
+where
+    T: Send + 'static,
+{
+    pub fn poll_result(&mut self) -> Result<T, Box<dyn Error + Send>> {
         if !self.is_alive() {
-            return None::<T>;
+            return Err(Box::new(RecvTimeoutError::Disconnected));
+        }
+
+        {
+            let result = self.blocking_recever.lock().unwrap().try_recv();
+
+            match result {
+                Ok(v) => Ok(v),
+                Err(e) => Err(Box::new(e)),
+            }
+        }
+    }
+
+    pub fn take_result(&mut self) -> Result<T, Box<dyn Error + Send>> {
+        if !self.is_alive() {
+            return Err(Box::new(RecvTimeoutError::Disconnected));
         }
 
         {
             match self.timeout {
                 Some(duration) => {
                     let result = self.blocking_recever.lock().unwrap().recv_timeout(duration);
 
-                    if self.panic && result.is_err() {
-                        std::panic::panic_any(result.err());
+                    match result {
+                        Ok(v) => Ok(v),
+                        Err(e) => Err(Box::new(e)),
                     }
-
-                    result.ok()
                 }
                 None => {
                     let result = self.blocking_recever.lock().unwrap().recv();
 
-                    if self.panic && result.is_err() {
-                        std::panic::panic_any(result.err());
+                    match result {
+                        Ok(v) => Ok(v),
+                        Err(e) => Err(Box::new(e)),
                     }
-
-                    result.ok()
                 }
             }
         }
     }
 }
+#[cfg(feature = "for_futures")]
+impl<T> BlockingQueue<T>
+where
+    T: Send + 'static + Clone,
+{
+    pub async fn poll_result_as_future(&mut self) -> Result<T, Box<dyn Error + Send>> {
+        let mut queue = self.clone();
+
+        let spawn_future_result = {
+            shared_thread_pool()
+                .inner
+                .lock()
+                .unwrap()
+                .spawn_with_handle(async move { queue.poll_result() })
+        };
+        match spawn_future_result {
+            Ok(future) => future.await,
+            Err(e) => Err(Box::new(e)),
+        }
+    }
+    pub async fn take_result_as_future(&mut self) -> Result<T, Box<dyn Error + Send>> {
+        let mut queue = self.clone();
+
+        let spawn_future_result = {
+            shared_thread_pool()
+                .inner
+                .lock()
+                .unwrap()
+                .spawn_with_handle(async move { queue.take_result() })
+        };
+        match spawn_future_result {
+            Ok(future) => future.await,
+            Err(e) => Err(Box::new(e)),
+        }
+    }
+}
 
 #[cfg(feature = "for_futures")]
 #[futures_test::test]