channel-receiver 0.1.0

Channel Receiver
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

pub use async_channel::Receiver as AsyncChannelReceiver;
use async_channel::TryRecvError as TryRecvErrorInner;

use crate::{AsyncReceiver, TryRecvError};

//
#[async_trait::async_trait]
impl<T> AsyncReceiver<T> for AsyncChannelReceiver<T> {
    async fn recv(&mut self) -> Option<T>
    where
        T: Send,
    {
        AsyncChannelReceiver::recv(self).await.ok()
    }

    fn try_recv(&mut self) -> Result<T, TryRecvError> {
        AsyncChannelReceiver::try_recv(self).map_err(Into::into)
    }

    fn close(&mut self) {
        AsyncChannelReceiver::close(self);
    }
}

//
impl From<TryRecvErrorInner> for TryRecvError {
    fn from(err: TryRecvErrorInner) -> Self {
        match err {
            TryRecvErrorInner::Empty => Self::Empty,
            TryRecvErrorInner::Closed => Self::Closed,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_with_bounded() {
        {
            let (tx, rx) = async_channel::bounded(1);
            let mut receiver: Box<dyn AsyncReceiver<usize>> = Box::new(rx);
            assert_eq!(tx.send(1).await, Ok(()));
            assert_eq!(receiver.recv().await, Some(1));
            assert_eq!(tx.try_send(2), Ok(()));
            assert_eq!(receiver.recv().await, Some(2));
            assert!(
                tokio::time::timeout(tokio::time::Duration::from_millis(200), receiver.recv())
                    .await
                    .is_err()
            );
            assert_eq!(receiver.try_recv(), Err(TryRecvError::Empty));
            drop(tx);
            assert_eq!(receiver.try_recv(), Err(TryRecvError::Closed));
            assert_eq!(
                tokio::time::timeout(tokio::time::Duration::from_millis(200), receiver.recv())
                    .await,
                Ok(None)
            );
        }
    }

    #[tokio::test]
    async fn test_with_unbounded() {
        {
            let (tx, rx) = async_channel::unbounded();
            let mut receiver: Box<dyn AsyncReceiver<usize>> = Box::new(rx);
            assert_eq!(tx.send(1).await, Ok(()));
            assert_eq!(tx.send(2).await, Ok(()));
            assert_eq!(receiver.recv().await, Some(1));
            assert_eq!(receiver.recv().await, Some(2));
            assert!(
                tokio::time::timeout(tokio::time::Duration::from_millis(200), receiver.recv())
                    .await
                    .is_err()
            );
            assert_eq!(receiver.try_recv(), Err(TryRecvError::Empty));
            drop(tx);
            assert_eq!(receiver.try_recv(), Err(TryRecvError::Closed));
            assert_eq!(
                tokio::time::timeout(tokio::time::Duration::from_millis(200), receiver.recv())
                    .await,
                Ok(None)
            );
        }
    }
}