Split async and blocking code

Author: frostyplanet

src/rx.rs src/async_rx.rssrc/rx.rs renamed to src/async_rx.rs

@@ -1,105 +1,14 @@
 use crate::channel::*;
-use crate::m_rx::*;
 use crate::stream::AsyncStream;
 use async_trait::async_trait;
 use crossbeam::channel::Receiver;
-pub use crossbeam::channel::{RecvError, RecvTimeoutError, TryRecvError};
+pub use crossbeam::channel::{RecvError, TryRecvError};
 use std::fmt;
 use std::future::Future;
+use std::ops::{Deref, DerefMut};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
-use std::time::Duration;
-
-/// Receiver that works in blocking context
-pub struct Rx<T> {
-    pub(crate) recv: Receiver<T>,
-    pub(crate) shared: Arc<ChannelShared>,
-}
-
-impl<T> fmt::Debug for Rx<T> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "Rx")
-    }
-}
-
-impl<T> Drop for Rx<T> {
-    fn drop(&mut self) {
-        self.shared.close_rx();
-    }
-}
-
-impl<T> Rx<T> {
-    #[inline]
-    pub(crate) fn new(recv: Receiver<T>, shared: Arc<ChannelShared>) -> Self {
-        Self { recv, shared }
-    }
-
-    /// Receive message, will block when channel is empty.
-    ///
-    /// Returns Ok(T) when successful.
-    ///
-    /// Returns Err([RecvError]) when all Tx dropped.
-    #[inline]
-    pub fn recv<'a>(&'a self) -> Result<T, RecvError> {
-        match self.recv.recv() {
-            Err(e) => return Err(e),
-            Ok(i) => {
-                self.shared.on_recv();
-                return Ok(i);
-            }
-        }
-    }
-
-    /// Try to receive message, non-blocking.
-    ///
-    /// Returns Ok(T) when successful.
-    ///
-    /// Returns Err([TryRecvError::Empty]) when channel is empty.
-    ///
-    /// returns Err([TryRecvError::Disconnected]) when all Tx dropped.
-    #[inline]
-    pub fn try_recv(&self) -> Result<T, TryRecvError> {
-        match self.recv.try_recv() {
-            Err(e) => return Err(e),
-            Ok(i) => {
-                self.shared.on_recv();
-                return Ok(i);
-            }
-        }
-    }
-
-    /// Waits for a message to be received from the channel, but only for a limited time.
-    /// Will block when channel is empty.
-    ///
-    /// Returns Ok(T) when successful.
-    ///
-    /// Returns Err([RecvTimeoutError::Timeout]) when a message could not be received because the channel is empty and the operation timed out.
-    ///
-    /// returns Err([RecvTimeoutError::Disconnected]) when all Tx dropped.
-    #[inline]
-    pub fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError> {
-        match self.recv.recv_timeout(timeout) {
-            Err(e) => return Err(e),
-            Ok(i) => {
-                self.shared.on_recv();
-                return Ok(i);
-            }
-        }
-    }
-
-    /// Probe possible messages in the channel (not accurate)
-    #[inline]
-    pub fn len(&self) -> usize {
-        self.recv.len()
-    }
-
-    /// Whether there's message in the channel (not accurate)
-    #[inline]
-    pub fn is_empty(&self) -> bool {
-        self.recv.is_empty()
-    }
-}
 
 /// Receiver that works in async context
 ///
@@ -317,143 +226,158 @@ impl<T> Future for ReceiveFuture<'_, T> {
     }
 }
 
-impl<T> From<MRx<T>> for Rx<T> {
-    fn from(rx: MRx<T>) -> Self {
-        rx.0
-    }
-}
-
-impl<T> From<MAsyncRx<T>> for AsyncRx<T> {
-    fn from(rx: MAsyncRx<T>) -> Self {
-        rx.0
-    }
-}
-
-/// For writing generic code with MRx & Rx
-pub trait BlockingRxTrait<T: Send + 'static>: Send + 'static {
-    /// Receive message, will block when channel is empty.
+/// For writing generic code with MAsyncRx & AsyncRx
+#[async_trait]
+pub trait AsyncRxTrait<T: Unpin + Send + 'static>: Send + Sync + 'static {
+    /// Receive message, will await when channel is empty.
     ///
     /// Returns `Ok(T)` when successful.
     ///
-    /// Returns Err([RecvError]) when all Tx dropped.
-    fn recv<'a>(&'a self) -> Result<T, RecvError>;
+    /// returns Err([RecvError]) when all Tx dropped.
+    async fn recv(&self) -> Result<T, RecvError>;
 
     /// Try to receive message, non-blocking.
     ///
-    /// Returns `Ok(T)` when successful.
+    /// Returns Ok(T) when successful.
     ///
     /// Returns Err([TryRecvError::Empty]) when channel is empty.
     ///
     /// Returns Err([TryRecvError::Disconnected]) when all Tx dropped.
     fn try_recv(&self) -> Result<T, TryRecvError>;
 
-    /// Waits for a message to be received from the channel, but only for a limited time.
-    /// Will block when channel is empty.
-    ///
-    /// Returns Ok(T) when successful.
-    ///
-    /// Returns Err([RecvTimeoutError::Timeout]) when a message could not be received because the channel is empty and the operation timed out.
-    ///
-    /// returns Err([RecvTimeoutError::Disconnected]) when all Tx dropped.
-    fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError>;
+    /// Generate a fixed Sized future object that receive a message
+    fn make_recv_future<'a>(&'a self) -> ReceiveFuture<'a, T>;
 
     /// Probe possible messages in the channel (not accurate)
     fn len(&self) -> usize;
 
     /// Whether there's message in the channel (not accurate)
     fn is_empty(&self) -> bool;
+
+    /// Returns count of tx / rx wakers stored in channel for debug purpose
+    #[cfg(test)]
+    fn get_waker_size(&self) -> (usize, usize);
 }
 
-impl<T: Send + 'static> BlockingRxTrait<T> for Rx<T> {
+#[async_trait]
+impl<T: Unpin + Send + 'static> AsyncRxTrait<T> for AsyncRx<T> {
     #[inline(always)]
-    fn recv<'a>(&'a self) -> Result<T, RecvError> {
-        Rx::recv(self)
+    async fn recv(&self) -> Result<T, RecvError> {
+        AsyncRx::recv(self).await
     }
 
     #[inline(always)]
     fn try_recv(&self) -> Result<T, TryRecvError> {
-        Rx::try_recv(self)
+        AsyncRx::try_recv(self)
     }
 
     #[inline(always)]
-    fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError> {
-        Rx::recv_timeout(self, timeout)
+    fn make_recv_future<'a>(&'a self) -> ReceiveFuture<'a, T> {
+        AsyncRx::make_recv_future(self)
     }
 
     #[inline(always)]
     fn len(&self) -> usize {
-        Rx::len(self)
+        AsyncRx::len(self)
     }
 
     #[inline(always)]
     fn is_empty(&self) -> bool {
-        Rx::is_empty(self)
+        AsyncRx::is_empty(self)
+    }
+
+    #[inline(always)]
+    #[cfg(test)]
+    fn get_waker_size(&self) -> (usize, usize) {
+        AsyncRx::get_waker_size(self)
     }
 }
 
-/// For writing generic code with MAsyncRx & AsyncRx
-#[async_trait]
-pub trait AsyncRxTrait<T: Unpin + Send + 'static>: Send + Sync + 'static {
-    /// Receive message, will await when channel is empty.
-    ///
-    /// Returns `Ok(T)` when successful.
-    ///
-    /// returns Err([RecvError]) when all Tx dropped.
-    async fn recv(&self) -> Result<T, RecvError>;
+/// Receiver that works in async context. MC version of [`AsyncRx<T>`] implements [Clone].
+///
+/// You can use `into()` to convert it to `AsyncRx<T>`.
+pub struct MAsyncRx<T>(pub(crate) AsyncRx<T>);
 
-    /// Try to receive message, non-blocking.
-    ///
-    /// Returns Ok(T) when successful.
-    ///
-    /// Returns Err([TryRecvError::Empty]) when channel is empty.
-    ///
-    /// Returns Err([TryRecvError::Disconnected]) when all Tx dropped.
-    fn try_recv(&self) -> Result<T, TryRecvError>;
+impl<T> Clone for MAsyncRx<T> {
+    #[inline]
+    fn clone(&self) -> Self {
+        let inner = &self.0;
+        inner.shared.add_rx();
+        Self(AsyncRx { recv: inner.recv.clone(), shared: inner.shared.clone() })
+    }
+}
 
-    /// Generate a fixed Sized future object that receive a message
-    fn make_recv_future<'a>(&'a self) -> ReceiveFuture<'a, T>;
+impl<T> From<MAsyncRx<T>> for AsyncRx<T> {
+    fn from(rx: MAsyncRx<T>) -> Self {
+        rx.0
+    }
+}
 
-    /// Probe possible messages in the channel (not accurate)
-    fn len(&self) -> usize;
+impl<T> MAsyncRx<T> {
+    #[inline]
+    pub(crate) fn new(recv: Receiver<T>, shared: Arc<ChannelShared>) -> Self {
+        Self(AsyncRx::new(recv, shared))
+    }
 
-    /// Whether there's message in the channel (not accurate)
-    fn is_empty(&self) -> bool;
+    #[inline]
+    pub fn into_stream(self) -> AsyncStream<T>
+    where
+        T: Sync + Send + Unpin + 'static,
+    {
+        AsyncStream::new(self.0)
+    }
+}
 
-    /// Returns count of tx / rx wakers stored in channel for debug purpose
-    #[cfg(test)]
-    fn get_waker_size(&self) -> (usize, usize);
+impl<T> Deref for MAsyncRx<T> {
+    type Target = AsyncRx<T>;
+
+    /// inherit all the functions of [AsyncRx]
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<T> DerefMut for MAsyncRx<T> {
+    /// inherit all the functions of [AsyncRx]
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
 }
 
 #[async_trait]
-impl<T: Unpin + Send + 'static> AsyncRxTrait<T> for AsyncRx<T> {
+impl<T: Unpin + Send + 'static> AsyncRxTrait<T> for MAsyncRx<T> {
     #[inline(always)]
     async fn recv(&self) -> Result<T, RecvError> {
-        AsyncRx::recv(self).await
+        self.0.recv().await
     }
 
     #[inline(always)]
     fn try_recv(&self) -> Result<T, TryRecvError> {
-        AsyncRx::try_recv(self)
+        self.0.try_recv()
     }
 
+    /// Generate a fixed Sized future object that receive a message
     #[inline(always)]
     fn make_recv_future<'a>(&'a self) -> ReceiveFuture<'a, T> {
-        AsyncRx::make_recv_future(self)
+        self.0.make_recv_future()
     }
 
+    /// Probe possible messages in the channel (not accurate)
     #[inline(always)]
     fn len(&self) -> usize {
-        AsyncRx::len(self)
+        self.0.len()
     }
 
+    /// Whether there's message in the channel (not accurate)
     #[inline(always)]
     fn is_empty(&self) -> bool {
-        AsyncRx::is_empty(self)
+        self.0.is_empty()
     }
 
+    /// Returns count of tx / rx wakers stored in channel for debug purpose
     #[inline(always)]
     #[cfg(test)]
     fn get_waker_size(&self) -> (usize, usize) {
-        AsyncRx::get_waker_size(self)
+        self.0.get_waker_size()
     }
 }