bombay/routing/
actor_ref.rs

1//! Typed handles to resolved actor mailboxes.
2
3use behavior::{Address, ShutdownEvent, ShutdownRequested, UserEvent};
4
5use crate::runtime::lifecycle::IncarnationReporter;
6use crate::{
7    DeliveryEndpoint, EventSender, LifecycleTransition, MailboxAnchor, MailboxSender, NoLifecycle,
8    RejectedDelivery,
9};
10
11/// A resolved actor endpoint rejected a message because its mailbox retired.
12#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
13#[error("actor mailbox closed")]
14pub struct MailboxDeliveryClosed;
15
16/// Failure to publish a typed graceful-shutdown request.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
18pub enum ShutdownRequestError {
19    /// The composed event protocol declined shutdown construction.
20    #[error("the composed event protocol declined shutdown construction")]
21    Unsupported,
22    /// The actor mailbox consumer has already retired.
23    #[error("the actor mailbox consumer has already retired")]
24    Closed,
25}
26
27/// A typed, directly resolved handle to an actor's user protocol.
28///
29/// References are issued by [`crate::System::spawn`] and may be cloned, but
30/// their concrete mailbox sender cannot be supplied or extracted externally.
31/// The message and event types are associated facts of the sender's event
32/// protocol, not independent choices.
33///
34/// ```compile_fail
35/// use bombay::{ActorRef, behavior::MailAddr};
36///
37/// let _: ActorRef<MailAddr, ()> = ActorRef::new(MailAddr(1), ());
38/// ```
39pub struct ActorRef<A, S, L = NoLifecycle> {
40    address: A,
41    sender: S,
42    lifecycle: L,
43}
44
45impl<A, S> ActorRef<A, S, NoLifecycle> {
46    /// Construct a typed reference from an address and event sender.
47    pub(crate) const fn new(address: A, sender: S) -> Self {
48        Self::with_lifecycle(address, sender, NoLifecycle)
49    }
50}
51
52impl<A, S, L> ActorRef<A, S, L> {
53    pub(crate) const fn with_lifecycle(address: A, sender: S, lifecycle: L) -> Self {
54        Self {
55            address,
56            sender,
57            lifecycle,
58        }
59    }
60
61    /// The destination address.
62    pub const fn address(&self) -> &A {
63        &self.address
64    }
65}
66
67impl<A, E, L> ActorRef<A, MailboxSender<E>, L> {
68    pub(crate) fn sender_anchor(&self) -> MailboxAnchor<E> {
69        self.sender.anchor()
70    }
71}
72
73impl<A, S, L> ActorRef<A, S, L>
74where
75    A: Address + Send + Sync,
76    S: EventSender + Sync,
77    S::Event: UserEvent<Addr = A>,
78    <S::Event as UserEvent>::Message: Send,
79    L: Sync,
80{
81    /// Deliver a message directly, stamped with `from`.
82    ///
83    /// Delivery awaits bounded mailbox admission and retains ownership of the
84    /// message until admission succeeds or the exact event is returned on
85    /// closure. Sequential sends by one producer retain their order. Cloned
86    /// references used by concurrent producers have no pre-admission ordering guarantee;
87    /// their accepted messages may interleave in either order.
88    ///
89    /// # Errors
90    ///
91    /// Returns the event sender's delivery failure.
92    pub async fn send(
93        &self,
94        from: A,
95        message: <S::Event as UserEvent>::Message,
96    ) -> Result<(), S::Error> {
97        self.sender
98            .send(<S::Event as UserEvent>::user(from, message))
99            .await
100    }
101}
102
103impl<A, E, L> ActorRef<A, MailboxSender<E>, L>
104where
105    E: ShutdownEvent,
106    L: IncarnationReporter,
107{
108    /// Publish one priority graceful-shutdown request.
109    ///
110    /// The request does not wait behind bounded user-mailbox backpressure.
111    /// A successful publication does not mean the actor has retired; await
112    /// its [`crate::Handle`] outcome for terminal completion.
113    ///
114    /// # Errors
115    ///
116    /// Returns [`ShutdownRequestError::Unsupported`] when the composed event
117    /// protocol declines construction, or [`ShutdownRequestError::Closed`]
118    /// after mailbox retirement.
119    pub fn request_shutdown(&self) -> Result<(), ShutdownRequestError> {
120        let event =
121            E::shutdown_requested(ShutdownRequested).ok_or(ShutdownRequestError::Unsupported)?;
122        self.sender
123            .send_control(event)
124            .map_err(|_| ShutdownRequestError::Closed)?;
125        self.lifecycle.emit(LifecycleTransition::ShutdownRequested);
126        Ok(())
127    }
128}
129
130impl<A, E, L> DeliveryEndpoint<A, <E as UserEvent>::Message> for ActorRef<A, MailboxAnchor<E>, L>
131where
132    A: Address + Send + Sync,
133    E: UserEvent<Addr = A> + Send,
134    <E as UserEvent>::Message: Send,
135    L: Sync,
136{
137    type Error = MailboxDeliveryClosed;
138
139    async fn deliver(
140        &self,
141        from: A,
142        message: <E as UserEvent>::Message,
143    ) -> Result<(), RejectedDelivery<<E as UserEvent>::Message, Self::Error>> {
144        let event = E::user(from, message);
145        match self.sender.send(event).await {
146            Ok(()) => Ok(()),
147            Err(closed) => {
148                let Ok(user) = closed.0.into_user() else {
149                    unreachable!("UserEvent::user must round-trip through UserEvent::into_user")
150                };
151                Err(RejectedDelivery::new(user.message, MailboxDeliveryClosed))
152            }
153        }
154    }
155}
156
157impl<A: Clone, S: Clone, L: Clone> Clone for ActorRef<A, S, L> {
158    fn clone(&self) -> Self {
159        Self::with_lifecycle(
160            self.address.clone(),
161            self.sender.clone(),
162            self.lifecycle.clone(),
163        )
164    }
165}
166
167#[cfg(test)]
168mod tests {
169    use std::convert::Infallible;
170    use std::sync::{Arc, Mutex};
171
172    use behavior::{MailAddr, ShutdownEvent, ShutdownProtocol, ShutdownRequested, User, UserEvent};
173
174    use super::{ActorRef, MailboxDeliveryClosed};
175    use crate::{
176        DeliveryEndpoint, EventSender, EventSource, MailboxConfig, RejectedDelivery,
177        ShutdownRequestError,
178    };
179
180    struct Message(String);
181
182    #[derive(Clone)]
183    struct Sender<E>(Arc<Mutex<Vec<E>>>);
184
185    impl<E: Send> EventSender for Sender<E> {
186        type Event = E;
187        type Error = Infallible;
188
189        async fn send(&self, event: E) -> Result<(), Self::Error> {
190            self.0.lock().expect("event lock").push(event);
191            Ok(())
192        }
193    }
194
195    #[tokio::test]
196    async fn moves_a_non_clone_message_directly_into_the_actor_event() {
197        let events: Arc<Mutex<Vec<User<MailAddr, Message>>>> = Arc::new(Mutex::new(Vec::new()));
198        let actor_ref = ActorRef::new(MailAddr(9), Sender(events.clone()));
199
200        actor_ref
201            .send(MailAddr(7), Message(String::from("hello")))
202            .await
203            .unwrap();
204
205        assert_eq!(*actor_ref.address(), MailAddr(9));
206        let events = events.lock().expect("event lock");
207        assert_eq!(events[0].from, MailAddr(7));
208        assert_eq!(events[0].message.0, "hello");
209    }
210
211    #[tokio::test]
212    async fn resolved_closed_mailbox_returns_the_exact_non_clone_message() {
213        let (sender, source) = MailboxConfig::bounded(1).create::<User<MailAddr, Message>>();
214        let endpoint = ActorRef::new(MailAddr(9), sender.anchor());
215        drop(source);
216
217        let RejectedDelivery { message, error } =
218            DeliveryEndpoint::deliver(&endpoint, MailAddr(7), Message(String::from("owned")))
219                .await
220                .expect_err("retired mailbox must reject delivery");
221
222        assert_eq!(message.0, "owned");
223        assert_eq!(error, MailboxDeliveryClosed);
224    }
225
226    #[tokio::test]
227    async fn publishes_shutdown_through_the_priority_lane() {
228        let (sender, mut source) =
229            MailboxConfig::bounded(1).create::<ShutdownProtocol<User<MailAddr, Message>>>();
230        let actor_ref = ActorRef::new(MailAddr(9), sender);
231
232        actor_ref.request_shutdown().unwrap();
233
234        assert!(matches!(
235            source.next().await,
236            Some(ShutdownProtocol::ShutdownRequested(ShutdownRequested))
237        ));
238    }
239
240    struct Declined;
241
242    impl ShutdownEvent for Declined {
243        fn shutdown_requested(_event: ShutdownRequested) -> Option<Self> {
244            None
245        }
246    }
247
248    impl UserEvent for Declined {
249        type Addr = MailAddr;
250        type Message = Message;
251
252        fn user(_from: MailAddr, _message: Message) -> Self {
253            Self
254        }
255
256        fn into_user(self) -> Result<User<MailAddr, Message>, Self> {
257            Err(self)
258        }
259    }
260
261    #[test]
262    fn distinguishes_declined_construction_from_closed_delivery() {
263        let (declined, _source) = MailboxConfig::bounded(1).create::<Declined>();
264        let declined = ActorRef::new(MailAddr(1), declined);
265        assert_eq!(
266            declined.request_shutdown(),
267            Err(ShutdownRequestError::Unsupported)
268        );
269
270        let (closed, source) =
271            MailboxConfig::bounded(1).create::<ShutdownProtocol<User<MailAddr, Message>>>();
272        drop(source);
273        let closed = ActorRef::new(MailAddr(2), closed);
274        assert_eq!(closed.request_shutdown(), Err(ShutdownRequestError::Closed));
275    }
276}
bombay/routing/actor_ref.rs

bombay/routing/
actor_ref.rs
