channel-receiver 0.2.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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use tokio::sync::mpsc::error::TryRecvError as TryRecvErrorInner;
pub use tokio::sync::mpsc::{
    Receiver as TokioReceiver, UnboundedReceiver as TokioUnboundedReceiver,
};

use crate::{AsyncReceiver, TryRecvError};

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

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

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

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

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

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

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

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

    #[tokio::test]
    async fn test_with_channel() {
        {
            let (tx, rx) = tokio::sync::mpsc::channel(1);
            let mut receiver: Box<dyn AsyncReceiver<usize>> = Box::new(rx);
            assert!(tx.send(1).await.is_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::Disconnected));
            assert_eq!(
                tokio::time::timeout(tokio::time::Duration::from_millis(200), receiver.recv())
                    .await,
                Ok(None)
            );
        }
    }

    #[tokio::test]
    async fn test_with_unbounded_channel() {
        {
            let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
            let mut receiver: Box<dyn AsyncReceiver<usize>> = Box::new(rx);
            assert!(tx.send(1).is_ok());
            assert!(tx.send(2).is_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::Disconnected));
            assert_eq!(
                tokio::time::timeout(tokio::time::Duration::from_millis(200), receiver.recv())
                    .await,
                Ok(None)
            );
        }
    }
}