server/event/
scalar.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::{Error, EventSubscriptionMessage, ScalarCodec, Server, ServerContext};
10use crate::{XousDeserializer, XousValidator};
11
12/// Handle for a single event subscriber
13pub struct ScalarEventSubscriber<M>
14where
15    M: ScalarEvent,
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 ScalarEventSubscriber<M>
25where
26    M: ScalarEvent,
27{
28    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        f.debug_struct("ScalarEventSubscriber").field("pid", &self.pid).finish()
30    }
31}
32
33impl<M> ScalarEventSubscriber<M>
34where
35    M: ScalarEvent,
36{
37    /// Send the event to the subscriber.
38    /// Can be used in an IRQ handler context.
39    pub fn send(&self, msg: &M) -> Result<xous::Result, xous::Error> {
40        let msg = xous::Message::Scalar(crate::scalar::scalar_to_message(msg, self.msg_id));
41        xous::try_send_message(self.cid, msg)
42    }
43
44    pub fn pid(&self) -> xous::PID { self.pid }
45
46    pub fn cid(&self) -> xous::CID { self.cid }
47}
48
49impl<M> Drop for ScalarEventSubscriber<M>
50where
51    M: ScalarEvent,
52{
53    fn drop(&mut self) {
54        if let Err(e) =
55            xous::send_message(self.cid, super::cancellation_message(self.msg_id, self.cancel_msg_id))
56        {
57            log::debug!("Error sending cancellation message {self:?}: {e:?}")
58        }
59        if let Err(e) = xous::disconnect(self.cid) {
60            log::error!("Error disconnecting {self:?}: {e:?}")
61        }
62    }
63}
64
65/// A message which can be serialized and deserialized using scalar encoding.
66pub trait ScalarEvent: ScalarCodec {}
67
68impl<M> ScalarEvent for M where M: ScalarCodec {}
69
70pub trait ScalarSubscription
71where
72    Self: crate::MessageId + crate::ArchiveCodec,
73    <Self::Error as rkyv::Archive>::Archived:
74        rkyv::Deserialize<Self::Error, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
75    <Result<(), Self::Error> as rkyv::Archive>::Archived:
76        rkyv::Deserialize<Result<(), Self::Error>, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
77{
78    type Event: ScalarEvent;
79    type Error: super::SubscriptionError;
80}
81
82/// A [`ScalarSubscription`] handler.
83pub trait ScalarEventSubscriptionHandler<M>
84where
85    Self: Server,
86    M: ScalarSubscription,
87{
88    /// Handle the subscription.
89    ///
90    /// The `subscriber` parameter can be used to store the subscriber info and send events to them
91    /// later. Once their subscription is not used, the object can be dropped.
92    fn handle(
93        &mut self,
94        msg: M,
95        subscriber: ScalarEventSubscriber<M::Event>,
96        context: &mut ServerContext<Self>,
97    ) -> Result<(), M::Error>;
98}
99
100/// Handler for an incoming [`ScalarEvent`]
101pub trait ScalarEventHandler<M>
102where
103    Self: Server,
104    M: ScalarEvent,
105{
106    fn handle(&mut self, msg: M, sender: xous::PID, context: &mut ServerContext<Self>);
107}
108
109/// Message handler, used by ServerMessages::messages()
110pub fn handle_scalar_subscription<M, S>(
111    handler: &mut S,
112    raw: xous::MessageEnvelope,
113    context: &mut ServerContext<S>,
114) where
115    M: ScalarSubscription + 'static,
116    S: ScalarEventSubscriptionHandler<M>,
117    <M as rkyv::Archive>::Archived:
118        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
119{
120    let pid = raw.sender.pid().unwrap();
121    if let Err(e) = try_handle_scalar_subscription(pid, handler, raw, context) {
122        log::warn!("archive sub handle error (PID {pid}) for {}: {e}", type_name::<M>());
123    }
124}
125
126fn try_handle_scalar_subscription<M, S>(
127    pid: xous::PID,
128    handler: &mut S,
129    mut raw: xous::MessageEnvelope,
130    context: &mut ServerContext<S>,
131) -> whence::Result<(), Error>
132where
133    M: ScalarSubscription + 'static,
134    S: ScalarEventSubscriptionHandler<M>,
135    <M as rkyv::Archive>::Archived:
136        rkyv::Deserialize<M, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
137{
138    let mem = crate::lend_mut::borrow_mut(&mut raw).whence()?;
139    let msg: EventSubscriptionMessage<M> = crate::Buffer::deserialize(mem).whence()?;
140    let res = handler.handle(
141        msg.msg,
142        ScalarEventSubscriber::<M::Event> {
143            pid,
144            msg_id: msg.msg_id,
145            cancel_msg_id: msg.cancel_msg_id,
146            cid: msg.cid,
147            _phantom: PhantomData,
148        },
149        context,
150    );
151    crate::Buffer::reply(mem, &res).whence()
152}
153
154pub fn decode_scalar_event<M>(raw: xous::MessageEnvelope) -> M
155where
156    M: ScalarEvent,
157{
158    try_decode_scalar_event(raw).unwrap()
159}
160
161pub fn try_decode_scalar_event<M>(mut raw: xous::MessageEnvelope) -> whence::Result<M, crate::Error>
162where
163    M: ScalarEvent,
164{
165    let scalar = crate::scalar::extract_scalar_message(&mut raw).whence()?;
166    Ok(M::from_scalar(scalar))
167}
168
169pub(crate) fn scalar_event_handler<M, S>(
170    handler: &mut S,
171    raw: xous::MessageEnvelope,
172    context: &mut ServerContext<S>,
173) where
174    M: ScalarEvent,
175    S: ScalarEventHandler<M>,
176{
177    let sender = raw.sender.pid().unwrap();
178    let msg = decode_scalar_event::<M>(raw);
179    handler.handle(msg, sender, context);
180}
181
182/// Subscribe to a [`ScalarEvent`] event.
183///
184/// # Arguments
185///
186/// * `cid` - The connection ID to the event sending server.
187/// * `sid` - The server ID of the event receiving server.
188///
189/// # Returns
190///
191/// A tuple containing two unique message IDs (to this process) for the incoming events:
192/// - The first ID is for the event message.
193/// - The second ID is for the cancellation message.
194pub fn subscribe_scalar<M>(cid: xous::CID, msg: M, sid: xous::SID) -> Result<(usize, usize), M::Error>
195where
196    M: ScalarSubscription + 'static,
197{
198    try_subscribe_scalar(cid, msg, sid).unwrap()
199}
200
201pub fn try_subscribe_scalar<M>(
202    cid: xous::CID,
203    msg: M,
204    sid: xous::SID,
205) -> whence::Result<Result<(usize, usize), M::Error>, crate::Error>
206where
207    M: ScalarSubscription + 'static,
208{
209    let msg_id = crate::next_dynamic_message_id();
210    let cancel_msg_id = crate::next_dynamic_message_id();
211    let pid = xous::get_remote_pid(cid).whence()?;
212    let cid_remote = xous::connect_for_process(pid, sid).whence()?;
213    xous::allow_messages_on_connection(pid, cid_remote, msg_id..(cancel_msg_id + 1)).whence()?;
214    let msg = EventSubscriptionMessage { cid: cid_remote, msg_id, cancel_msg_id, msg };
215    let result = msg.send_scalar(cid)?;
216    Ok(result.map(|_| (msg_id, cancel_msg_id)))
217}
218
219/// A list of scalar event subscribers.
220pub struct ScalarSubList<T: ScalarCodec> {
221    inner: Vec<ScalarEventSubscriber<T>>,
222}
223
224impl<T: ScalarCodec> Default for ScalarSubList<T> {
225    fn default() -> Self { Self { inner: Default::default() } }
226}
227
228impl<T: ScalarCodec> ScalarSubList<T> {
229    pub fn push(&mut self, sub: ScalarEventSubscriber<T>) { self.inner.push(sub); }
230
231    pub fn send(&mut self, msg: &T) { self.inner.retain(|sub| sub.send(msg).is_ok()) }
232
233    pub fn send_nowait(&mut self, msg: &T) {
234        self.inner.retain(|sub| match sub.send(msg) {
235            Ok(_) => true,
236            Err(xous::Error::ServerQueueFull) => {
237                log::warn!("scalar event send_nowait error for pid {} {}", sub.pid(), type_name::<T>());
238                true
239            }
240            Err(_) => false,
241        })
242    }
243
244    pub fn remove_cid(&mut self, cid: xous::CID) { self.inner.retain(|s| s.cid() != cid) }
245}
server/event/scalar.rs

server/event/
scalar.rs
