commonware_broadcast/buffered/
ingress.rs

1use crate::Broadcaster;
2use commonware_actor::{
3    mailbox::{Overflow, Policy, Sender},
4    Feedback,
5};
6use commonware_codec::Codec;
7use commonware_cryptography::{Digestible, PublicKey};
8use commonware_p2p::Recipients;
9use commonware_utils::channel::oneshot;
10use std::{collections::VecDeque, sync::Arc};
11
12/// Message types that can be sent to the `Mailbox`
13pub(crate) enum Message<P: PublicKey, M: Digestible> {
14    /// Broadcast a [crate::Broadcaster::Message] to the network.
15    Broadcast {
16        recipients: Recipients<P>,
17        message: Arc<M>,
18    },
19
20    /// Subscribe to receive a message by digest.
21    ///
22    /// The responder will be sent the message when it is available; either
23    /// instantly (if cached) or when it is received from the network. The request can be canceled
24    /// by dropping the responder.
25    Subscribe {
26        digest: M::Digest,
27        responder: oneshot::Sender<Arc<M>>,
28    },
29
30    /// Get a message by digest.
31    Get {
32        digest: M::Digest,
33        responder: oneshot::Sender<Option<Arc<M>>>,
34    },
35}
36
37impl<P: PublicKey, M: Digestible> Message<P, M> {
38    fn response_closed(&self) -> bool {
39        match self {
40            Self::Subscribe { responder, .. } => responder.is_closed(),
41            Self::Get { responder, .. } => responder.is_closed(),
42            Self::Broadcast { .. } => false,
43        }
44    }
45}
46
47pub(crate) struct Pending<P: PublicKey, M: Digestible>(VecDeque<Message<P, M>>);
48
49impl<P: PublicKey, M: Digestible> Default for Pending<P, M> {
50    fn default() -> Self {
51        Self(VecDeque::new())
52    }
53}
54
55impl<P: PublicKey, M: Digestible> Overflow<Message<P, M>> for Pending<P, M> {
56    fn is_empty(&self) -> bool {
57        self.0.is_empty()
58    }
59
60    fn drain<F>(&mut self, mut push: F)
61    where
62        F: FnMut(Message<P, M>) -> Option<Message<P, M>>,
63    {
64        while let Some(message) = self.0.pop_front() {
65            if message.response_closed() {
66                continue;
67            }
68
69            if let Some(message) = push(message) {
70                self.0.push_front(message);
71                break;
72            }
73        }
74    }
75}
76
77impl<P: PublicKey, M: Digestible> Policy for Message<P, M> {
78    type Overflow = Pending<P, M>;
79
80    fn handle(overflow: &mut Self::Overflow, message: Self) {
81        if message.response_closed() {
82            return;
83        }
84
85        overflow.0.push_back(message);
86    }
87}
88
89/// Ingress mailbox for [super::Engine].
90#[derive(Clone)]
91pub struct Mailbox<P: PublicKey, M: Digestible + Codec> {
92    sender: Sender<Message<P, M>>,
93}
94
95impl<P: PublicKey, M: Digestible + Codec> Mailbox<P, M> {
96    pub(super) const fn new(sender: Sender<Message<P, M>>) -> Self {
97        Self { sender }
98    }
99
100    /// Subscribe to a message by digest.
101    ///
102    /// The responder will be sent the message when it is available; either
103    /// instantly (if cached) or when it is received from the network. The request can be canceled
104    /// by dropping the responder.
105    ///
106    /// If the engine has shut down, the returned receiver will resolve to `Canceled`.
107    pub fn subscribe(&self, digest: M::Digest) -> oneshot::Receiver<Arc<M>> {
108        let (responder, receiver) = oneshot::channel();
109        let _ = self
110            .sender
111            .enqueue(Message::Subscribe { digest, responder });
112        receiver
113    }
114
115    /// Get a message by digest.
116    ///
117    /// If the engine has shut down, returns `None`.
118    pub async fn get(&self, digest: M::Digest) -> Option<Arc<M>> {
119        let (responder, receiver) = oneshot::channel();
120        let _ = self.sender.enqueue(Message::Get { digest, responder });
121        receiver.await.unwrap_or_default()
122    }
123
124    /// Broadcast a shared message to recipients.
125    ///
126    /// If the engine has shut down, returns [`Feedback::Closed`].
127    pub fn broadcast_shared(&self, recipients: Recipients<P>, message: Arc<M>) -> Feedback {
128        self.sender.enqueue(Message::Broadcast {
129            recipients,
130            message,
131        })
132    }
133}
134
135impl<P: PublicKey, M: Digestible + Codec> Broadcaster for Mailbox<P, M> {
136    type Recipients = Recipients<P>;
137    type Message = M;
138
139    /// Broadcast a message to recipients.
140    ///
141    /// If the engine has shut down, returns [`Feedback::Closed`].
142    fn broadcast(&self, recipients: Self::Recipients, message: Self::Message) -> Feedback {
143        self.broadcast_shared(recipients, Arc::new(message))
144    }
145}
commonware_broadcast/buffered/ingress.rs

commonware_broadcast/buffered/
ingress.rs
