server/event/
archive.rs

1// SPDX-FileCopyrightText: 2023 Foundation Devices, Inc. <hello@foundation.xyz>
2// SPDX-License-Identifier: GPL-3.0-or-later
3
4use std::{any::type_name, marker::PhantomData};
5
6use rkyv::bytecheck::CheckBytes;
7use whence::WhenceExt;
8
9use crate::{ArchiveCodec, Error, EventSubscriptionMessage, Owned, Server, ServerContext};
10use crate::{XousDeserializer, XousValidator};
11
12/// Handle for a single event subscriber
13pub struct ArchiveEventSubscriber<M>
14where
15    M: ArchiveCodec,
16{
17    pid: xous::PID,
18    cid: xous::CID,
19    msg_id: xous::MessageId,
20    cancel_msg_id: xous::MessageId,
21    _phantom: PhantomData<M>,
22}
23
24impl<M> core::fmt::Debug for ArchiveEventSubscriber<M>
25where
26    M: ArchiveCodec,
27{
28    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        f.debug_struct("ArchiveEventSubscriber").field("pid", &self.pid).finish()
30    }
31}
32
33impl<M> ArchiveEventSubscriber<M>
34where
35    M: ArchiveCodec,
36{
37    /// Send the event to the subscriber.
38    ///
39    /// Warning: Cannot be used in an IRQ handler context.
40    pub fn send(&self, msg: &M) -> Result<xous::Result, xous::Error> {
41        crate::Buffer::into_buf(msg)
42            .map_err(|_| xous::Error::InternalError)?
43            .send(self.cid, self.msg_id as u32)
44    }
45
46    /// Send the event to the subscriber.
47    pub fn send_nowait(&self, msg: &M) -> Result<xous::Result, xous::Error> {
48        crate::Buffer::into_buf(msg)
49            .map_err(|_| xous::Error::InternalError)?
50            .send_nowait(self.cid, self.msg_id as u32)
51    }
52
53    pub fn pid(&self) -> xous::PID { self.pid }
54
55    pub fn cid(&self) -> xous::CID { self.cid }
56}
57
58impl<M> Drop for ArchiveEventSubscriber<M>
59where
60    M: ArchiveCodec,
61{
62    fn drop(&mut self) {
63        if let Err(e) =
64            xous::send_message(self.cid, super::cancellation_message(self.msg_id, self.cancel_msg_id))
65        {
66            log::debug!("Error sending cancellation message {self:?}: {e:?}")
67        }
68        if let Err(e) = xous::disconnect(self.cid) {
69            log::error!("Error disconnecting {self:?}: {e:?}")
70        }
71    }
72}
73
74/// A message which can be serialized and deserialized using rkyv.
75pub trait ArchiveEvent: ArchiveCodec
76where
77    <Self as rkyv::Archive>::Archived: for<'a> CheckBytes<XousValidator<'a>>,
78{
79}
80
81impl<M> ArchiveEvent for M
82where
83    M: ArchiveCodec,
84    <M as rkyv::Archive>::Archived: for<'a> CheckBytes<XousValidator<'a>>,
85{
86}
87
88pub trait ArchiveSubscription
89where
90    Self: crate::MessageId + ArchiveCodec,
91    <Self::Event as rkyv::Archive>::Archived:
92        rkyv::Deserialize<Self::Event, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
93    <Self::Error as rkyv::Archive>::Archived:
94        rkyv::Deserialize<Self::Error, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
95{
96    type Event: ArchiveEvent;
97    type Error: super::SubscriptionError;
98}
99
100pub trait ArchiveEventSubscriptionHandler<M>
101where
102    Self: Server,
103    M: ArchiveSubscription,
104{
105    fn handle(
106        &mut self,
107        msg: M,
108        subscriber: ArchiveEventSubscriber<M::Event>,
109        context: &mut ServerContext<Self>,
110    ) -> Result<(), M::Error>;
111}
112
113/// Handler for an incoming [`ArchiveEvent`]
114pub trait ArchiveEventHandler<M>
115where
116    Self: Server,
117    M: ArchiveEvent,
118    <M as rkyv::Archive>::Archived: for<'a> CheckBytes<XousValidator<'a>>,
119{
120    fn handle(&mut self, msg: Owned<M>, sender: xous::PID, context: &mut ServerContext<Self>);
121}
122
123/// Message handler, used by ServerMessages::messages()
124pub fn handle_archive_subscription<M, S>(
125    handler: &mut S,
126    raw: xous::MessageEnvelope,
127    context: &mut ServerContext<S>,
128) where
129    M: ArchiveSubscription + 'static,
130    S: ArchiveEventSubscriptionHandler<M>,
131    <M as rkyv::Archive>::Archived:
132        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
133{
134    let pid = raw.sender.pid().unwrap();
135    if let Err(e) = try_handle_archive_subscription(pid, handler, raw, context) {
136        log::warn!("archive sub handle error (PID {pid}) for {}: {e}", type_name::<M>());
137    }
138}
139
140/// Message handler, used by ServerMessages::messages()
141fn try_handle_archive_subscription<M, S>(
142    pid: xous::PID,
143    handler: &mut S,
144    mut raw: xous::MessageEnvelope,
145    context: &mut ServerContext<S>,
146) -> whence::Result<(), Error>
147where
148    M: ArchiveSubscription + 'static,
149    S: ArchiveEventSubscriptionHandler<M>,
150    <M as rkyv::Archive>::Archived:
151        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
152{
153    let mem = crate::lend_mut::borrow_mut(&mut raw).whence()?;
154    let msg: EventSubscriptionMessage<M> = crate::Buffer::deserialize(mem).whence()?;
155    let res = handler.handle(
156        msg.msg,
157        ArchiveEventSubscriber::<M::Event> {
158            pid,
159            msg_id: msg.msg_id,
160            cancel_msg_id: msg.cancel_msg_id,
161            cid: msg.cid,
162            _phantom: PhantomData,
163        },
164        context,
165    );
166    crate::Buffer::reply(mem, &res).whence()
167}
168
169pub fn decode_archive_event<M>(raw: xous::MessageEnvelope) -> M
170where
171    M: ArchiveEvent,
172    <M as rkyv::Archive>::Archived:
173        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
174{
175    try_decode_archive_event(raw).unwrap()
176}
177
178pub fn try_decode_archive_event<M>(raw: xous::MessageEnvelope) -> whence::Result<M, Error>
179where
180    M: ArchiveEvent,
181    <M as rkyv::Archive>::Archived:
182        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
183{
184    crate::r#move::decode_move::<M>(&raw).whence()
185}
186
187pub(crate) fn archive_event_handler<M, S>(
188    handler: &mut S,
189    raw: xous::MessageEnvelope,
190    context: &mut ServerContext<S>,
191) where
192    M: ArchiveEvent + 'static,
193    S: ArchiveEventHandler<M>,
194    <M as rkyv::Archive>::Archived: for<'a> CheckBytes<XousValidator<'a>>,
195{
196    if let Err(e) = try_archive_event_handler(handler, raw, context) {
197        log::warn!("failed to handle archive event {e:?}")
198    }
199}
200
201fn try_archive_event_handler<M, S>(
202    handler: &mut S,
203    raw: xous::MessageEnvelope,
204    context: &mut ServerContext<S>,
205) -> whence::Result<(), Error>
206where
207    M: ArchiveEvent + 'static,
208    S: ArchiveEventHandler<M>,
209    <M as rkyv::Archive>::Archived: for<'a> CheckBytes<XousValidator<'a>>,
210{
211    let sender = raw.sender.pid().unwrap();
212    let msg = Owned::new_move(raw).whence()?;
213    handler.handle(msg, sender, context);
214    Ok(())
215}
216
217/// Subscribe to an [`ArchiveEvent`] event.
218///
219/// # Arguments
220///
221/// * `cid` - The connection ID to the event sending server.
222/// * `sid` - The server ID of the event receiving server.
223///
224/// # Returns
225///
226/// A tuple containing two unique message IDs (to this process) for the incoming events:
227/// - The first ID is for the event message.
228/// - The second ID is for the cancellation message.
229pub fn subscribe_archive<M>(cid: xous::CID, msg: M, sid: xous::SID) -> Result<(usize, usize), M::Error>
230where
231    M: ArchiveSubscription + 'static,
232{
233    try_subscribe_archive(cid, msg, sid).unwrap()
234}
235
236pub fn try_subscribe_archive<M>(
237    cid: xous::CID,
238    msg: M,
239    sid: xous::SID,
240) -> whence::Result<Result<(usize, usize), M::Error>, Error>
241where
242    M: ArchiveSubscription + 'static,
243{
244    let msg_id = crate::next_dynamic_message_id();
245    let cancel_msg_id = crate::next_dynamic_message_id();
246    let pid = xous::get_remote_pid(cid).whence()?;
247    let cid_remote = xous::connect_for_process(pid, sid).whence()?;
248    xous::allow_messages_on_connection(pid, cid_remote, msg_id..(cancel_msg_id + 1)).whence()?;
249    let msg = EventSubscriptionMessage { cid: cid_remote, msg_id, cancel_msg_id, msg };
250    let result = msg.send_archive(cid)?;
251    Ok(result.map(|_| (msg_id, cancel_msg_id)))
252}
253
254#[derive(Debug)]
255pub struct ArchiveSubList<T: ArchiveCodec> {
256    inner: Vec<ArchiveEventSubscriber<T>>,
257}
258
259impl<T: ArchiveCodec> Default for ArchiveSubList<T> {
260    fn default() -> Self { Self { inner: Default::default() } }
261}
262
263impl<T: ArchiveCodec> ArchiveSubList<T> {
264    pub fn push(&mut self, sub: ArchiveEventSubscriber<T>) { self.inner.push(sub); }
265
266    pub fn send(&mut self, msg: &T) { self.inner.retain(|sub| sub.send(msg).is_ok()) }
267
268    pub fn send_nowait(&mut self, msg: &T) {
269        self.inner.retain(|sub| match sub.send_nowait(msg) {
270            Ok(_) => true,
271            Err(xous::Error::ServerQueueFull) => {
272                log::warn!("archive event send_nowait error for pid {} {}", sub.pid(), type_name::<T>());
273                true
274            }
275            Err(_) => false,
276        })
277    }
278
279    pub fn remove_cid(&mut self, cid: xous::CID) { self.inner.retain(|s| s.cid() != cid) }
280}
server/event/archive.rs

server/event/
archive.rs
