channel_sender/
impl_tokio.rs

1use tokio::sync::mpsc::error::TrySendError;
2pub use tokio::sync::{
3    mpsc::{Sender as TokioMpscSender, UnboundedSender as TokioMpscUnboundedSender},
4    oneshot::Sender as TokioOneshotSender,
5};
6
7//
8mod multi_producer_impl {
9    use super::*;
10
11    use crate::{
12        error::{SendError, SendErrorWithoutFull},
13        multi_producer::{BoundedSender, UnboundedSender},
14    };
15
16    #[async_trait::async_trait]
17    impl<T> BoundedSender<T> for TokioMpscSender<T> {
18        async fn send(&self, t: T) -> Result<(), SendErrorWithoutFull<T>>
19        where
20            T: Send,
21        {
22            TokioMpscSender::send(self, t)
23                .await
24                .map_err(|err| SendErrorWithoutFull::Closed(err.0))
25        }
26
27        fn try_send(&self, t: T) -> Result<(), SendError<T>> {
28            TokioMpscSender::try_send(self, t).map_err(Into::into)
29        }
30    }
31
32    impl<T> UnboundedSender<T> for TokioMpscUnboundedSender<T> {
33        fn send(&self, t: T) -> Result<(), SendErrorWithoutFull<T>> {
34            TokioMpscUnboundedSender::send(self, t)
35                .map_err(|err| SendErrorWithoutFull::Closed(err.0))
36        }
37    }
38}
39
40//
41mod one_shot_impl {
42    use super::*;
43
44    use crate::{
45        error::SendErrorWithoutFull,
46        one_shot::{BoxSender, Sender},
47    };
48
49    impl<T> Sender<T> for TokioOneshotSender<T> {
50        fn send(self, t: T) -> Result<(), SendErrorWithoutFull<T>> {
51            TokioOneshotSender::send(self, t).map_err(|t| SendErrorWithoutFull::Closed(t))
52        }
53    }
54
55    impl<T> BoxSender<T> for TokioOneshotSender<T> {
56        fn send(self: Box<Self>, t: T) -> Result<(), SendErrorWithoutFull<T>> {
57            TokioOneshotSender::send(*self, t).map_err(|t| SendErrorWithoutFull::Closed(t))
58        }
59    }
60}
61
62//
63mod generic_impl {
64    use super::*;
65
66    use crate::{
67        error::SendError,
68        generic::{CloneableSender, Sender},
69    };
70
71    impl<T> Sender<T> for TokioMpscSender<T> {
72        fn send(&self, t: T) -> Result<(), SendError<T>> {
73            TokioMpscSender::try_send(self, t).map_err(Into::into)
74        }
75    }
76
77    impl<T> CloneableSender<T> for TokioMpscSender<T> {
78        fn send(&self, t: T) -> Result<(), SendError<T>> {
79            TokioMpscSender::try_send(self, t).map_err(Into::into)
80        }
81    }
82
83    impl<T> Sender<T> for TokioMpscUnboundedSender<T> {
84        fn send(&self, t: T) -> Result<(), SendError<T>> {
85            TokioMpscUnboundedSender::send(self, t).map_err(|err| SendError::Closed(err.0))
86        }
87    }
88
89    impl<T> CloneableSender<T> for TokioMpscUnboundedSender<T> {
90        fn send(&self, t: T) -> Result<(), SendError<T>> {
91            TokioMpscUnboundedSender::send(self, t).map_err(|err| SendError::Closed(err.0))
92        }
93    }
94}
95
96//
97mod error_convert {
98    use super::*;
99
100    use crate::error::SendError;
101
102    impl<T> From<TrySendError<T>> for SendError<T> {
103        fn from(err: TrySendError<T>) -> Self {
104            match err {
105                TrySendError::Full(v) => Self::Full(v),
106                TrySendError::Closed(v) => Self::Closed(v),
107            }
108        }
109    }
110}
111
112#[cfg(test)]
113mod multi_producer_impl_tests {
114    use crate::{
115        error::{SendError, SendErrorWithoutFull},
116        multi_producer::{BoundedSender, UnboundedSender},
117    };
118
119    #[tokio::test]
120    async fn test_with_channel() {
121        {
122            let (tx, mut rx) = tokio::sync::mpsc::channel(1);
123            let sender: Box<dyn BoundedSender<usize>> = Box::new(tx);
124            let sender = sender.clone();
125            assert_eq!(sender.send(1).await, Ok(()));
126            assert_eq!(sender.try_send(2), Err(SendError::Full(2)));
127            assert!(
128                tokio::time::timeout(tokio::time::Duration::from_millis(200), sender.send(2))
129                    .await
130                    .is_err()
131            );
132            assert_eq!(rx.recv().await, Some(1));
133            drop(rx);
134            assert_eq!(sender.send(3).await, Err(SendErrorWithoutFull::Closed(3)));
135            assert_eq!(sender.try_send(3), Err(SendError::Closed(3)));
136        }
137    }
138
139    #[tokio::test]
140    async fn test_with_unbounded_channel() {
141        {
142            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
143            let sender: Box<dyn UnboundedSender<usize>> = Box::new(tx);
144            let sender = sender.clone();
145            assert_eq!(sender.send(1), Ok(()));
146            assert_eq!(sender.send(2), Ok(()));
147            assert_eq!(rx.recv().await, Some(1));
148            assert_eq!(rx.recv().await, Some(2));
149            drop(rx);
150            assert_eq!(sender.send(3), Err(SendErrorWithoutFull::Closed(3)));
151        }
152    }
153}
154
155#[cfg(test)]
156mod one_shot_impl_tests {
157    use crate::one_shot::{BoxSender, Sender};
158
159    #[tokio::test]
160    async fn test_with_channel() {
161        {
162            fn send<T>(sender: T, msg: usize)
163            where
164                T: Sender<usize>,
165            {
166                assert_eq!(sender.send(msg), Ok(()));
167            }
168
169            let (tx, rx) = tokio::sync::oneshot::channel();
170            send(tx, 1);
171            assert_eq!(rx.await, Ok(1));
172
173            let (tx, rx) = tokio::sync::oneshot::channel();
174            let sender: Box<dyn BoxSender<usize>> = Box::new(tx);
175            assert_eq!(sender.send(1), Ok(()));
176            assert_eq!(rx.await, Ok(1));
177        }
178    }
179}
180
181#[cfg(test)]
182mod generic_impl_tests {
183    use crate::{
184        error::SendError,
185        generic::{CloneableSender, Sender},
186    };
187
188    #[tokio::test]
189    async fn test_with_channel() {
190        {
191            let (tx, mut rx) = tokio::sync::mpsc::channel(1);
192            let sender: Box<dyn Sender<usize>> = Box::new(tx);
193            assert_eq!(sender.send(1), Ok(()));
194            assert_eq!(sender.send(2), Err(SendError::Full(2)));
195            assert_eq!(rx.recv().await, Some(1));
196            drop(rx);
197            assert_eq!(sender.send(3), Err(SendError::Closed(3)));
198        }
199        {
200            let (tx, mut rx) = tokio::sync::mpsc::channel(1);
201            let sender: Box<dyn CloneableSender<usize>> = Box::new(tx);
202            let sender = sender.clone();
203            assert_eq!(sender.send(1), Ok(()));
204            assert_eq!(sender.send(2), Err(SendError::Full(2)));
205            assert_eq!(rx.recv().await, Some(1));
206            drop(rx);
207            assert_eq!(sender.send(3), Err(SendError::Closed(3)));
208        }
209    }
210
211    #[tokio::test]
212    async fn test_with_unbounded_channel() {
213        {
214            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
215            let sender: Box<dyn Sender<usize>> = Box::new(tx);
216            assert_eq!(sender.send(1), Ok(()));
217            assert_eq!(sender.send(2), Ok(()));
218            assert_eq!(rx.recv().await, Some(1));
219            assert_eq!(rx.recv().await, Some(2));
220            drop(rx);
221            assert_eq!(sender.send(3), Err(SendError::Closed(3)));
222        }
223        {
224            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
225            let sender: Box<dyn CloneableSender<usize>> = Box::new(tx);
226            let sender = sender.clone();
227            assert_eq!(sender.send(1), Ok(()));
228            assert_eq!(sender.send(2), Ok(()));
229            assert_eq!(rx.recv().await, Some(1));
230            assert_eq!(rx.recv().await, Some(2));
231            drop(rx);
232            assert_eq!(sender.send(3), Err(SendError::Closed(3)));
233        }
234    }
235}
channel_sender/impl_tokio.rs

channel_sender/
impl_tokio.rs
