async_repeater/
handle.rs

1use std::fmt::Display;
2
3use tokio::sync::{
4    mpsc::{self},
5    oneshot::{self, channel},
6};
7
8use crate::entry::RepeaterEntry;
9
10/// Internal messaage for communication between [`RepeaterHandle`](RepeaterHandle) and [`Repeater`](crate::Repeater)
11pub(crate) enum Message<E>
12where
13    E: RepeaterEntry + Clone + Unpin + 'static,
14{
15    Insert(E),
16    Remove(E::Key),
17    Clear,
18    Stop,
19    Len(oneshot::Sender<usize>),
20    IsEmtpy(oneshot::Sender<bool>),
21}
22
23/// Used to communicate with the repeater task
24///
25/// Returned after the call to [`Repeater::run_with_async_callback`](crate::Repeater::run_with_async_callback)
26#[derive(Debug, Clone)]
27pub struct RepeaterHandle<E>
28where
29    E: RepeaterEntry + Clone + Unpin + 'static,
30{
31    tx: mpsc::Sender<Message<E>>,
32}
33
34impl<E> RepeaterHandle<E>
35where
36    E: RepeaterEntry + Clone + Unpin + 'static,
37{
38    pub(crate) fn new(sender: mpsc::Sender<Message<E>>) -> Self {
39        Self { tx: sender }
40    }
41
42    /// Insert entry into [`Repeater`](crate::Repeater)
43    pub async fn insert(&self, e: E) -> Result<(), ConnectionLost> {
44        Ok(self.tx.send(Message::Insert(e)).await?)
45    }
46
47    /// Remove entry from [`Repeater`](crate::Repeater)
48    pub async fn remove(&self, key: E::Key) -> Result<(), ConnectionLost> {
49        Ok(self.tx.send(Message::Remove(key)).await?)
50    }
51
52    /// Remove all entries from [`Repeater`](crate::Repeater)
53    pub async fn clear(&self) -> Result<(), ConnectionLost> {
54        Ok(self.tx.send(Message::Clear).await?)
55    }
56
57    /// Stop a [`Repeater`](crate::Repeater)
58    pub async fn stop(self) -> Result<(), ConnectionLost> {
59        Ok(self.tx.send(Message::Stop).await?)
60    }
61
62    /// Return the number of entries in [`Repeater`](crate::Repeater)
63    pub async fn len(self) -> Result<usize, ConnectionLost> {
64        let (reply_tx, reply_rx) = channel();
65
66        self.tx.send(Message::Len(reply_tx)).await?;
67        Ok(reply_rx.await?)
68    }
69
70    /// Return true if no entries are inside of [`Repeater`](crate::Repeater)
71    pub async fn is_empty(self) -> Result<bool, ConnectionLost> {
72        let (reply_tx, reply_rx) = channel();
73
74        self.tx.send(Message::IsEmtpy(reply_tx)).await?;
75        Ok(reply_rx.await?)
76    }
77}
78
79/// Error type for losing connecting to the [`Repeater`](crate::Repeater)
80#[derive(Debug)]
81pub struct ConnectionLost;
82
83impl Display for ConnectionLost {
84    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85        write!(f, "Connection to repeater lost")
86    }
87}
88
89impl<T> From<mpsc::error::SendError<T>> for ConnectionLost {
90    fn from(_: mpsc::error::SendError<T>) -> Self {
91        Self
92    }
93}
94
95impl From<oneshot::error::RecvError> for ConnectionLost {
96    fn from(_: oneshot::error::RecvError) -> Self {
97        Self
98    }
99}
async_repeater/handle.rs

async_repeater/
handle.rs
