acktor_ipc/codec/
control_message.rs

1use std::fmt::Display;
2
3use bytes::{Bytes, BytesMut};
4use prost::Message as _;
5
6use acktor::{
7    Actor, ActorState, Address, Message, SenderId, Signal,
8    cron::CronSignal,
9    observer::Observer,
10    supervisor::{SupervisionEvent, Supervisor},
11};
12use acktor_ipc_proto::control_message as proto;
13
14use super::errors::{DecodeError, EncodeError};
15use super::{Decode, DecodeContext, Encode, EncodeContext};
16use crate::remote_message::RemoteSupervisionEvent;
17
18impl Encode for Signal {
19    const ID: u64 = u64::MAX;
20
21    #[inline]
22    fn encoded_len(&self) -> usize {
23        proto::Signal::new(*self as i32).encoded_len()
24    }
25
26    #[inline]
27    fn encode(&self, buf: &mut BytesMut, _ctx: Option<&EncodeContext>) -> Result<(), EncodeError> {
28        proto::Signal::new(*self as i32)
29            .encode(buf)
30            .map_err(Into::into)
31    }
32}
33
34impl Decode for Signal {
35    const ID: u64 = u64::MAX;
36
37    #[inline]
38    fn decode(buf: Bytes, _ctx: Option<&DecodeContext>) -> Result<Self, DecodeError> {
39        let signal = proto::Signal::decode(buf)?;
40        Signal::try_from(signal.signal as u8)
41            .map_err(|_| "invalid signal value in the `Signal` message".into())
42    }
43}
44
45impl<A> Encode for Supervisor<A>
46where
47    A: Actor,
48{
49    const ID: u64 = u64::MAX - 1;
50
51    fn encoded_len(&self) -> usize {
52        let supervisor = match self {
53            Supervisor::Set(recipient) => proto::Supervisor::set(recipient.index()),
54            Supervisor::Unset => proto::Supervisor::unset(),
55        };
56        supervisor.encoded_len()
57    }
58
59    fn encode(&self, buf: &mut BytesMut, ctx: Option<&EncodeContext>) -> Result<(), EncodeError> {
60        let supervisor = match self {
61            Supervisor::Set(recipient) => {
62                ctx.ok_or(EncodeError::MissingEncodeContext)?
63                    .register(recipient)?;
64                proto::Supervisor::set(recipient.index())
65            }
66
67            Supervisor::Unset => proto::Supervisor::unset(),
68        };
69        supervisor.encode(buf).map_err(Into::into)
70    }
71}
72
73impl<A> Decode for Supervisor<A>
74where
75    A: Actor,
76{
77    const ID: u64 = u64::MAX - 1;
78
79    fn decode(buf: Bytes, ctx: Option<&DecodeContext>) -> Result<Self, DecodeError> {
80        let ctx = ctx.ok_or(DecodeError::MissingDecodeContext)?;
81        let supervisor = proto::Supervisor::decode(buf)?;
82        match supervisor.supervisor {
83            Some(proto::SupervisorType::Set(actor_id)) => {
84                Ok(Supervisor::Set(ctx.create_remote_address(actor_id)?.into()))
85            }
86            Some(proto::SupervisorType::Unset(())) => Ok(Supervisor::Unset),
87            None => Err("missing field `supervisor` in the `Supervisor` message".into()),
88        }
89    }
90}
91
92impl<M> Encode for Observer<M>
93where
94    M: Message + Encode,
95    M::Result: Decode,
96{
97    const ID: u64 = u64::MAX - 2;
98
99    fn encoded_len(&self) -> usize {
100        let observer = match self {
101            Observer::Register(recipient) => proto::Observer::register(recipient.index()),
102            Observer::Unregister(recipient) => proto::Observer::unregister(recipient.index()),
103        };
104        observer.encoded_len()
105    }
106
107    fn encode(&self, buf: &mut BytesMut, ctx: Option<&EncodeContext>) -> Result<(), EncodeError> {
108        let observer = match self {
109            Observer::Register(recipient) => {
110                ctx.ok_or(EncodeError::MissingEncodeContext)?
111                    .register(recipient)?;
112                proto::Observer::register(recipient.index())
113            }
114
115            Observer::Unregister(recipient) => {
116                ctx.ok_or(EncodeError::MissingEncodeContext)?
117                    .register(recipient)?;
118                proto::Observer::unregister(recipient.index())
119            }
120        };
121        observer.encode(buf).map_err(Into::into)
122    }
123}
124
125impl<M> Decode for Observer<M>
126where
127    M: Message + Encode,
128    M::Result: Decode,
129{
130    const ID: u64 = u64::MAX - 2;
131
132    fn decode(buf: Bytes, ctx: Option<&DecodeContext>) -> Result<Self, DecodeError> {
133        let ctx = ctx.ok_or(DecodeError::MissingDecodeContext)?;
134        let observer = proto::Observer::decode(buf)?;
135        match observer.observer {
136            Some(proto::ObserverType::Register(actor_id)) => Ok(Observer::Register(
137                ctx.create_remote_address(actor_id)?.into(),
138            )),
139            Some(proto::ObserverType::Unregister(actor_id)) => Ok(Observer::Unregister(
140                ctx.create_remote_address(actor_id)?.into(),
141            )),
142            None => Err("missing field `observer` in the `Observer` message".into()),
143        }
144    }
145}
146
147fn build_supervision_event_message<A>(
148    event: &SupervisionEvent<A>,
149) -> (proto::SupervisionEvent, &Address<A>)
150where
151    A: Actor,
152    A::Error: Display,
153{
154    match event {
155        SupervisionEvent::Warn(address, error) => (
156            proto::SupervisionEvent::warn(address.index(), error.to_string()),
157            address,
158        ),
159        SupervisionEvent::Terminated(address, error) => (
160            proto::SupervisionEvent::terminated(
161                address.index(),
162                error.as_ref().map(|e| e.to_string()),
163            ),
164            address,
165        ),
166        SupervisionEvent::Panicked(address, info) => (
167            proto::SupervisionEvent::panicked(address.index(), info.to_string()),
168            address,
169        ),
170        SupervisionEvent::State(address, state) => (
171            proto::SupervisionEvent::state(address.index(), *state as i32),
172            address,
173        ),
174    }
175}
176
177impl<A> Encode for SupervisionEvent<A>
178where
179    A: Actor,
180    A::Error: Display,
181{
182    const ID: u64 = u64::MAX - 3;
183
184    fn encoded_len(&self) -> usize {
185        let (event, _) = build_supervision_event_message(self);
186        event.encoded_len()
187    }
188
189    fn encode(&self, buf: &mut BytesMut, ctx: Option<&EncodeContext>) -> Result<(), EncodeError> {
190        let (event, address) = build_supervision_event_message(self);
191        ctx.ok_or(EncodeError::MissingEncodeContext)?
192            .register(address)?;
193        event.encode(buf).map_err(Into::into)
194    }
195
196    fn encode_to_bytes(&self, ctx: Option<&EncodeContext>) -> Result<Bytes, EncodeError> {
197        let (event, address) = build_supervision_event_message(self);
198        ctx.ok_or(EncodeError::MissingEncodeContext)?
199            .register(address)?;
200        let mut buf = BytesMut::with_capacity(event.encoded_len());
201        event.encode(&mut buf)?;
202        Ok(buf.freeze())
203    }
204}
205
206impl Decode for RemoteSupervisionEvent {
207    const ID: u64 = u64::MAX - 3;
208
209    #[inline]
210    fn decode(buf: Bytes, ctx: Option<&DecodeContext>) -> Result<Self, DecodeError> {
211        let ctx = ctx.ok_or(DecodeError::MissingDecodeContext)?;
212        let event = proto::SupervisionEvent::decode(buf)?;
213        match event.event {
214            Some(proto::SupervisionEventType::Warn(warn)) => Ok(RemoteSupervisionEvent::Warn(
215                ctx.create_remote_address(warn.actor_id)?,
216                warn.err,
217            )),
218
219            Some(proto::SupervisionEventType::Terminated(terminated)) => {
220                Ok(RemoteSupervisionEvent::Terminated(
221                    ctx.create_remote_address(terminated.actor_id)?,
222                    terminated.err,
223                ))
224            }
225
226            Some(proto::SupervisionEventType::Panicked(panicked)) => {
227                Ok(RemoteSupervisionEvent::Panicked(
228                    ctx.create_remote_address(panicked.actor_id)?,
229                    panicked.info,
230                ))
231            }
232
233            Some(proto::SupervisionEventType::State(state)) => Ok(RemoteSupervisionEvent::State(
234                ctx.create_remote_address(state.actor_id)?,
235                ActorState::try_from(state.state as u8).map_err(|_| {
236                    DecodeError::from("invalid actor state value in the `SupervisionEvent` message")
237                })?,
238            )),
239
240            None => Err("missing field `event` in the `SupervisionEvent` message".into()),
241        }
242    }
243}
244
245impl Encode for CronSignal {
246    const ID: u64 = u64::MAX - 4;
247
248    #[inline]
249    fn encoded_len(&self) -> usize {
250        proto::CronSignal::new(*self as i32).encoded_len()
251    }
252
253    #[inline]
254    fn encode(&self, buf: &mut BytesMut, _ctx: Option<&EncodeContext>) -> Result<(), EncodeError> {
255        proto::CronSignal::new(*self as i32)
256            .encode(buf)
257            .map_err(Into::into)
258    }
259}
260
261impl Decode for CronSignal {
262    const ID: u64 = u64::MAX - 4;
263
264    #[inline]
265    fn decode(buf: Bytes, _ctx: Option<&DecodeContext>) -> Result<Self, DecodeError> {
266        let cron_signal = proto::CronSignal::decode(buf)?;
267        CronSignal::try_from(cron_signal.signal as u8)
268            .map_err(|_| "invalid signal value in the `CronSignal` message".into())
269    }
270}
acktor_ipc/codec/control_message.rs

acktor_ipc/codec/
control_message.rs
