compio_actor\mailbox/
mod.rs

1//! Typed mailboxes, calls, brokers, and delivery errors.
2
3mod call;
4mod error;
5mod name;
6mod receiver;
7
8use std::{
9    any::Any,
10    fmt,
11    num::NonZeroUsize,
12    sync::{
13        Arc,
14        atomic::{AtomicBool, Ordering},
15    },
16};
17
18pub(crate) use call::call_with;
19#[doc(inline)]
20pub use call::{Call, Reply};
21#[doc(inline)]
22pub use error::{CallError, DeliverError};
23use flume::{Sender, TrySendError};
24pub(crate) use name::Name;
25pub(crate) use receiver::{MailboxEvent, Receiver, make_mailbox};
26
27use crate::{Actor, Handler, Message, actor::Delivering};
28
29/// Default number of messages reserved for each mailbox.
30pub const DEFAULT_MAILBOX_CAPACITY: NonZeroUsize = NonZeroUsize::new(64).unwrap();
31
32struct MailboxInner<A: Actor> {
33    name: Option<Name>,
34    messages: Sender<Delivering<A>>,
35    stop: Sender<()>,
36    stopping: AtomicBool,
37    capacity: NonZeroUsize,
38}
39
40impl<A: Actor> MailboxInner<A> {
41    fn send<M>(&self, message: M) -> Result<(), DeliverError<M>>
42    where
43        A: Handler<M>,
44        M: Message,
45    {
46        if self.is_closed() {
47            return Err(DeliverError::Closed(message));
48        }
49
50        self.messages
51            .try_send(Delivering::<A>::from_msg(message))
52            .map_err(|error| match error {
53                TrySendError::Full(message) => DeliverError::Full(message.recover::<M>()),
54                TrySendError::Disconnected(message) => DeliverError::Closed(message.recover::<M>()),
55            })
56    }
57
58    fn stop(&self) -> bool {
59        if self.stopping.swap(true, Ordering::AcqRel) {
60            return false;
61        }
62        self.stop.try_send(()).is_ok()
63    }
64
65    fn begin_stop(&self) {
66        self.stopping.store(true, Ordering::Release);
67    }
68
69    fn is_closed(&self) -> bool {
70        self.stopping.load(Ordering::Acquire)
71            || self.messages.is_disconnected()
72            || self.stop.is_disconnected()
73    }
74}
75
76/// A typed reference to an actor in a cluster.
77pub struct Mailbox<A: Actor> {
78    inner: Arc<MailboxInner<A>>,
79}
80
81impl<A: Actor> Mailbox<A> {
82    pub(crate) fn begin_stop(&self) {
83        self.inner.begin_stop();
84    }
85
86    /// Returns the actor's registered name.
87    pub fn name(&self) -> Option<&str> {
88        self.inner.name.as_ref().map(Name::as_str)
89    }
90
91    /// Enqueues a message handled by this actor without waiting.
92    pub fn send<M>(&self, message: M) -> Result<(), DeliverError<M>>
93    where
94        A: Handler<M>,
95        M: Message,
96    {
97        self.inner.send(message)
98    }
99
100    /// Creates a send-only capability for one message type.
101    pub fn broker<M>(&self) -> Broker<M>
102    where
103        A: Handler<M>,
104        M: Message,
105    {
106        let inner: Arc<dyn BrokerSink<M>> = self.inner.clone();
107        Broker { inner }
108    }
109
110    /// Requests a graceful stop, returning whether this call requested it.
111    pub fn stop(&self) -> bool {
112        self.inner.stop()
113    }
114
115    /// Returns whether the mailbox rejects new messages.
116    pub fn is_closed(&self) -> bool {
117        self.inner.is_closed()
118    }
119
120    /// Returns the fixed mailbox capacity.
121    pub fn capacity(&self) -> NonZeroUsize {
122        self.inner.capacity
123    }
124}
125
126impl<A: Actor> Clone for Mailbox<A> {
127    fn clone(&self) -> Self {
128        Self {
129            inner: self.inner.clone(),
130        }
131    }
132}
133
134impl<A: Actor> fmt::Debug for Mailbox<A> {
135    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136        f.debug_struct("Mailbox")
137            .field("name", &self.name())
138            .field("capacity", &self.inner.capacity)
139            .field("queued", &self.inner.messages.len())
140            .field("closed", &self.is_closed())
141            .finish()
142    }
143}
144
145trait BrokerSink<M: Message>: Send + Sync {
146    fn name(&self) -> Option<&str>;
147    fn send(&self, message: M) -> Result<(), DeliverError<M>>;
148}
149
150impl<A, M> BrokerSink<M> for MailboxInner<A>
151where
152    A: Handler<M>,
153    M: Message,
154{
155    fn name(&self) -> Option<&str> {
156        self.name.as_ref().map(Name::as_str)
157    }
158
159    fn send(&self, message: M) -> Result<(), DeliverError<M>> {
160        MailboxInner::send(self, message)
161    }
162}
163
164/// A send-only capability for messages of type `M`.
165pub struct Broker<M: Message> {
166    inner: Arc<dyn BrokerSink<M>>,
167}
168
169impl<M: Message> Broker<M> {
170    /// Returns the actor's registered name.
171    pub fn name(&self) -> Option<&str> {
172        self.inner.name()
173    }
174
175    /// Enqueues a message without waiting.
176    pub fn send(&self, message: M) -> Result<(), DeliverError<M>> {
177        self.inner.send(message)
178    }
179}
180
181impl<M: Message> Clone for Broker<M> {
182    fn clone(&self) -> Self {
183        Self {
184            inner: self.inner.clone(),
185        }
186    }
187}
188
189impl<M: Message> fmt::Debug for Broker<M> {
190    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
191        f.debug_struct("Broker")
192            .field("name", &self.name())
193            .finish_non_exhaustive()
194    }
195}
196
197pub(crate) type ErasedMailbox = Arc<dyn Any + Send + Sync>;
198
199impl<A: Actor> Mailbox<A> {
200    pub(crate) fn erase(&self) -> ErasedMailbox {
201        self.inner.clone()
202    }
203
204    pub(crate) fn from_erased(inner: ErasedMailbox) -> Option<Self> {
205        Arc::downcast::<MailboxInner<A>>(inner)
206            .ok()
207            .map(|inner| Self { inner })
208    }
209}
compio_actor\mailbox/mod.rs

compio_actor\mailbox/
mod.rs
