mm1_core/
envelope.rs

1use std::any::TypeId;
2use std::fmt;
3use std::sync::atomic::{AtomicU64, Ordering};
4
5use mm1_address::address::Address;
6pub use mm1_proc_macros::dispatch;
7use mm1_proto::Message;
8
9use crate::message::AnyMessage;
10use crate::tracing::TraceId;
11
12static ENVELOPE_SEQ_NO: AtomicU64 = AtomicU64::new(0);
13
14const DEFAULT_TTL: u8 = 15;
15
16#[derive(Debug)]
17pub struct EnvelopeHeader {
18    pub to:       Address,
19    pub ttl:      u8,
20    pub priority: bool,
21    pub no:       u64,
22    pub trace_id: TraceId,
23}
24
25pub struct Envelope<M = AnyMessage> {
26    header:  EnvelopeHeader,
27    message: M,
28}
29
30impl EnvelopeHeader {
31    pub fn to_address(to: Address) -> Self {
32        Self {
33            to,
34            no: ENVELOPE_SEQ_NO.fetch_add(1, Ordering::Relaxed),
35            priority: false,
36            ttl: DEFAULT_TTL,
37            trace_id: TraceId::current(),
38        }
39    }
40
41    pub fn with_ttl(self, ttl: u8) -> Self {
42        Self { ttl, ..self }
43    }
44
45    pub fn with_trace_id(self, trace_id: TraceId) -> Self {
46        Self { trace_id, ..self }
47    }
48
49    pub fn with_priority(self, priority: bool) -> Self {
50        Self { priority, ..self }
51    }
52
53    pub fn with_no(self, no: u64) -> Self {
54        Self { no, ..self }
55    }
56
57    pub fn trace_id(&self) -> TraceId {
58        self.trace_id
59    }
60}
61
62impl<M> Envelope<M>
63where
64    M: Message,
65{
66    pub fn into_erased(self) -> Envelope<AnyMessage> {
67        let Self {
68            header: info,
69            message,
70        } = self;
71        let message = AnyMessage::new(message);
72        Envelope {
73            header: info,
74            message,
75        }
76    }
77}
78
79impl<M> Envelope<M> {
80    pub fn new(header: EnvelopeHeader, message: M) -> Self {
81        Self { header, message }
82    }
83
84    pub fn header(&self) -> &EnvelopeHeader {
85        &self.header
86    }
87}
88
89impl Envelope<AnyMessage> {
90    pub fn cast<M>(self) -> Result<Envelope<M>, Self>
91    where
92        M: Message,
93    {
94        let Self {
95            header: info,
96            message,
97        } = self;
98        match message.cast() {
99            Ok(message) => {
100                Ok(Envelope {
101                    header: info,
102                    message,
103                })
104            },
105            Err(message) => {
106                Err(Self {
107                    header: info,
108                    message,
109                })
110            },
111        }
112    }
113
114    pub fn peek<M>(&self) -> Option<&M>
115    where
116        M: Message,
117    {
118        self.message.peek()
119    }
120
121    pub fn is<M>(&self) -> bool
122    where
123        M: Message,
124    {
125        self.message.is::<M>()
126    }
127
128    pub fn tid(&self) -> TypeId {
129        self.message.tid()
130    }
131
132    pub fn message_name(&self) -> &str {
133        self.message.type_name()
134    }
135
136    /// Log an unmatched message and drop it.
137    ///
138    /// The `dispatch!` macro calls this for a message that no arm matched when
139    /// the `match` has no catch-all arm. The message is logged at `WARN` (with
140    /// its header and type name) and dropped — mirroring the fire-and-forget
141    /// semantics elsewhere in the runtime, so an unexpected message from any
142    /// peer no longer panics and kills the actor.
143    ///
144    /// A `dispatch!` used in value position must supply its own catch-all arm,
145    /// because this fallback evaluates to `()`.
146    pub fn log_unhandled(&self) {
147        ::tracing::warn!(
148            header = ?self.header(),
149            message = self.message_name(),
150            "dispatch!: dropping unhandled message"
151        );
152    }
153}
154impl<M> Envelope<M> {
155    pub fn take(self) -> (M, Envelope<()>) {
156        (
157            self.message,
158            Envelope {
159                header:  self.header,
160                message: (),
161            },
162        )
163    }
164}
165
166impl From<Envelope<()>> for EnvelopeHeader {
167    fn from(value: Envelope<()>) -> Self {
168        let Envelope {
169            header,
170            message: (),
171        } = value;
172        header
173    }
174}
175
176impl<M> fmt::Debug for Envelope<M>
177where
178    M: fmt::Debug,
179{
180    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
181        f.debug_struct("Envelope")
182            .field("info", &self.header)
183            .field("message", &self.message)
184            .finish()
185    }
186}
mm1_core/envelope.rs

mm1_core/
envelope.rs
