acktor_ipc/
remote_message.rs

1//! Remote message which can be sent over an IPC channel.
2//!
3
4use std::fmt::{self, Debug};
5
6use bytes::Bytes;
7use thiserror::Error;
8
9use acktor::{Actor, ActorState, Address, Message, Recipient, Sender, channel::oneshot};
10
11use crate::codec::DecodeContext;
12use crate::remote_address::RemoteAddress;
13
14/// A unified remote message used for communication with remote actors.
15///
16/// This is used both for outbound messages (sent by an actor in the current process to an actor
17/// in a remote process through [`Session`][crate::session::Session]) and for inbound messages
18/// (received by a [`Session`][crate::session::Session] and forwarded to an actor in the current
19/// process for handling).
20#[derive(Message)]
21#[result_type(())]
22pub struct RemoteMessage {
23    pub actor_id: u64,
24    pub message_id: u64,
25    pub message: Bytes,
26    pub result_tx: Option<oneshot::Sender<Bytes>>,
27    pub decode_context: Option<DecodeContext>,
28}
29
30impl Debug for RemoteMessage {
31    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
32        f.debug_struct("RemoteMessage")
33            .field("actor_id", &self.actor_id)
34            .field("message_id", &self.message_id)
35            .field("message", &format_args!("Bytes({})", self.message.len()))
36            .field(
37                "result_tx",
38                &match self.result_tx {
39                    Some(_) => format_args!("Send"),
40                    None => format_args!("DoSend"),
41                },
42            )
43            .finish()
44    }
45}
46
47impl RemoteMessage {
48    /// Constructs a new [`RemoteMessage`] which does not expect a response.
49    pub fn do_send(actor_id: u64, message_id: u64, message: Bytes) -> Self {
50        Self {
51            actor_id,
52            message_id,
53            message,
54            result_tx: None,
55            decode_context: None,
56        }
57    }
58
59    /// Constructs a new [`RemoteMessage`] which expects a response.
60    pub fn send(
61        actor_id: u64,
62        message_id: u64,
63        message: Bytes,
64        tx: oneshot::Sender<Bytes>,
65    ) -> Self {
66        Self {
67            actor_id,
68            message_id,
69            message,
70            result_tx: Some(tx),
71            decode_context: None,
72        }
73    }
74
75    /// Sets the decode context on this message.
76    pub fn with_context(mut self, context: DecodeContext) -> Self {
77        self.decode_context = Some(context);
78        self
79    }
80}
81
82#[derive(Debug, Error)]
83pub enum ToRemoteMessageRecipientError {
84    #[error(
85        "actor does not opt-in the `type-erased-recipient-hook` feature, see the docs of \
86         `RemoteActor` for details"
87    )]
88    MissingHook,
89
90    #[error("{0}")]
91    DowncastFailed(String),
92}
93
94pub trait ToRemoteMessageRecipient {
95    fn to_remote_message_recipient(
96        &self,
97    ) -> Result<Recipient<RemoteMessage>, ToRemoteMessageRecipientError>;
98}
99
100impl<A> ToRemoteMessageRecipient for Address<A>
101where
102    A: Actor,
103{
104    fn to_remote_message_recipient(
105        &self,
106    ) -> Result<Recipient<RemoteMessage>, ToRemoteMessageRecipientError> {
107        Ok(*self
108            .type_erased_recipient()
109            .ok_or(ToRemoteMessageRecipientError::MissingHook)?
110            .downcast::<Recipient<RemoteMessage>>()
111            .map_err(|(_, e)| ToRemoteMessageRecipientError::DowncastFailed(e))?)
112    }
113}
114
115impl<M> ToRemoteMessageRecipient for Recipient<M>
116where
117    M: Message,
118{
119    fn to_remote_message_recipient(
120        &self,
121    ) -> Result<Recipient<RemoteMessage>, ToRemoteMessageRecipientError> {
122        Ok(*self
123            .type_erased_recipient()
124            .ok_or(ToRemoteMessageRecipientError::MissingHook)?
125            .downcast::<Recipient<RemoteMessage>>()
126            .map_err(|(_, e)| ToRemoteMessageRecipientError::DowncastFailed(e))?)
127    }
128}
129
130/// A message which is used to report actor status to a supervisor from a remote node.
131#[derive(Debug, Message)]
132#[result_type(())]
133pub enum RemoteSupervisionEvent {
134    /// Warning, the actor could resume by itself.
135    Warn(RemoteAddress, String),
136    /// Actor terminated with or without error.
137    Terminated(RemoteAddress, Option<String>),
138    /// Actor panicked with the given panic info.
139    Panicked(RemoteAddress, String),
140    /// Actor state changed.
141    State(RemoteAddress, ActorState),
142}
acktor_ipc/remote_message.rs

acktor_ipc/
remote_message.rs
