async_transmit/transmit/
tokio.rs

1use async_trait::async_trait;
2use tokio::sync::mpsc::error::SendError;
3use tokio::sync::mpsc::{Sender, UnboundedSender};
4
5use crate::transmit::Transmit;
6
7#[async_trait]
8impl<I> Transmit for Sender<I>
9where
10    I: Send,
11{
12    type Item = I;
13    type Error = SendError<I>;
14
15    async fn transmit(&mut self, item: Self::Item) -> Result<(), Self::Error> {
16        Sender::send(self, item).await?;
17        Ok(())
18    }
19}
20
21#[async_trait]
22impl<I> Transmit for UnboundedSender<I>
23where
24    I: Send,
25{
26    type Item = I;
27    type Error = SendError<I>;
28
29    async fn transmit(&mut self, item: Self::Item) -> Result<(), Self::Error> {
30        UnboundedSender::send(self, item)?;
31        Ok(())
32    }
33}
34
35#[cfg(test)]
36mod tests {
37    use super::super::assert_transmit;
38    use super::*;
39
40    use anyhow::Result;
41    use futures_await_test::async_test;
42
43    #[async_test]
44    async fn tokio_sender_is_transmit() -> Result<()> {
45        let (s, mut r) = tokio::sync::mpsc::channel::<&'static str>(10);
46
47        let mut t = assert_transmit(s);
48        t.transmit("Hello").await?;
49        t.transmit("World").await?;
50        drop(t);
51        assert_eq!(r.recv().await, Some("Hello"));
52        assert_eq!(r.recv().await, Some("World"));
53        assert_eq!(r.recv().await, None);
54
55        Ok(())
56    }
57
58    #[async_test]
59    async fn tokio_unbounded_sender_is_transmit() -> Result<()> {
60        let (s, mut r) = tokio::sync::mpsc::unbounded_channel::<&'static str>();
61
62        let mut t = assert_transmit(s);
63        t.transmit("Hello").await?;
64        t.transmit("World").await?;
65        drop(t);
66        assert_eq!(r.recv().await, Some("Hello"));
67        assert_eq!(r.recv().await, Some("World"));
68        assert_eq!(r.recv().await, None);
69
70        Ok(())
71    }
72}
async_transmit/transmit/tokio.rs

async_transmit/transmit/
tokio.rs
