polybox/
tokio_inbox.rs

1use super::*;
2use futures::future::BoxFuture;
3use polybox_core::errors::{SendCheckedError, SendError};
4use std::sync::Arc;
5
6/// A wrapper around a [`tokio::sync::mpsc::Sender`] that acts as a [`PolyBox`].
7pub struct TokioInbox<T> {
8    sender: tokio::sync::mpsc::Sender<T>,
9}
10
11impl<T> TokioInbox<T> {
12    pub fn new(buffer: usize) -> (Self, tokio::sync::mpsc::Receiver<T>) {
13        let (sender, receiver) = tokio::sync::mpsc::channel(buffer);
14        (Self { sender }, receiver)
15    }
16
17    pub fn inner(&self) -> &tokio::sync::mpsc::Sender<T> {
18        &self.sender
19    }
20
21    pub fn into_inner(self) -> tokio::sync::mpsc::Sender<T> {
22        self.sender
23    }
24
25    pub fn from_inner(sender: tokio::sync::mpsc::Sender<T>) -> Self {
26        Self { sender }
27    }
28}
29
30impl<T: Interface> PolyBox for TokioInbox<T> {
31    type Set = T::Set;
32
33    fn into_dyn_unchecked<R>(self) -> DynInbox<R> {
34        DynInbox::new_unchecked(Arc::new(self))
35    }
36}
37
38impl<T: Interface> DynPolyBox for TokioInbox<T> {
39    fn _send_boxed_payload_checked(
40        &self,
41        msg: BoxedPayload,
42    ) -> BoxFuture<'_, Result<(), SendCheckedError<BoxedPayload>>> {
43        Box::pin(async move {
44            let payload = msg
45                .try_into_interface::<T>()
46                .map_err(|payload| SendCheckedError::NotAccepted(payload))?;
47
48            self.send(payload).await.map_err(|SendError(payload)| {
49                SendCheckedError::Closed(T::into_boxed_payload(payload))
50            })
51        })
52    }
53}
54
55impl<T, R> Sends<T> for TokioInbox<R>
56where
57    T: Message,
58    R: TryIntoPayload<T> + FromPayload<T> + Send,
59{
60    async fn send(&self, msg: T) -> Result<Output<T>, SendError<T>> {
61        let (payload, output) = T::build_payload(msg);
62        let interface = R::from_payload(payload);
63
64        match self.sender.send(interface).await {
65            Ok(()) => Ok(output),
66            Err(e) => Err(SendError(T::destroy_payload(
67                e.0.try_into_payload()
68                    .map_err(|_| ())
69                    .expect("Failed to convert payload back"),
70            ))),
71        }
72    }
73}
74
75impl<T> Clone for TokioInbox<T> {
76    fn clone(&self) -> Self {
77        Self {
78            sender: self.sender.clone(),
79        }
80    }
81}
polybox/tokio_inbox.rs

polybox/
tokio_inbox.rs
