Robin Mueller
b7ce039406
All checks were successful
Rust/sat-rs/pipeline/head This commit looks good
853 lines
34 KiB
Rust
853 lines
34 KiB
Rust
//! Event management and forwarding
|
|
//!
|
|
//! It is recommended to read the
|
|
//! [sat-rs book chapter](https://absatsw.irs.uni-stuttgart.de/projects/sat-rs/book/events.html)
|
|
//! about events first.
|
|
//!
|
|
//! This module provides components to perform event routing. The most important component for this
|
|
//! task is the [EventManager]. It receives all events and then routes them to event subscribers
|
|
//! where appropriate.
|
|
//!
|
|
//! The event manager has a listener table abstracted by the [ListenerMapProvider], which maps
|
|
//! listener groups identified by [ListenerKey]s to a [listener ID][ComponentId].
|
|
//! It also contains a sender table abstracted by the [SenderMapProvider] which maps these sender
|
|
//! IDs to concrete [EventSendProvider]s. A simple approach would be to use one send event provider
|
|
//! for each OBSW thread and then subscribe for all interesting events for a particular thread
|
|
//! using the send event provider ID.
|
|
//!
|
|
//! This can be done with the [EventManager] like this:
|
|
//!
|
|
//! 1. Provide a concrete [EventReceiveProvider] implementation. This abstraction allow to use different
|
|
//! message queue backends. A straightforward implementation where dynamic memory allocation is
|
|
//! not a big concern would be to use the [std::sync::mpsc::Receiver] handle. The trait is
|
|
//! already implemented for this type.
|
|
//! 2. To set up event creators, create channel pairs using some message queue implementation.
|
|
//! Each event creator gets a (cloned) sender component which allows it to send events to the
|
|
//! manager.
|
|
//! 3. The event manager receives the receiver component as part of a [EventReceiveProvider]
|
|
//! implementation so all events are routed to the manager.
|
|
//! 4. Create the [event sender map][SenderMapProvider]s which allow routing events to
|
|
//! subscribers. You can now use the subscriber component IDs to subscribe
|
|
//! for event groups, for example by using the [EventManager::subscribe_single] method.
|
|
//! 5. Add the send provider as well using the [EventManager::add_sender] call so the event
|
|
//! manager can route listener groups to a the send provider.
|
|
//!
|
|
//! Some components like a PUS Event Service or PUS Event Action Service might require all
|
|
//! events to package them as telemetry or start actions where applicable.
|
|
//! Other components might only be interested in certain events. For example, a thermal system
|
|
//! handler might only be interested in temperature events generated by a thermal sensor component.
|
|
//!
|
|
//! # Examples
|
|
//!
|
|
//! You can check [integration test](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs/tests/pus_events.rs)
|
|
//! for a concrete example using multi-threading where events are routed to
|
|
//! different threads.
|
|
//!
|
|
//! The [satrs-example](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-example)
|
|
//! also contains a full event manager instance and exposes a test event via the PUS test service.
|
|
//! The [PUS event](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-example/src/pus/event.rs)
|
|
//! module and the generic [events module](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-example/src/events.rs)
|
|
//! show how the event management modules can be integrated into a more complex software.
|
|
use crate::events::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw};
|
|
use crate::params::Params;
|
|
use crate::queue::GenericSendError;
|
|
use core::fmt::Debug;
|
|
use core::marker::PhantomData;
|
|
use core::slice::Iter;
|
|
|
|
use crate::ComponentId;
|
|
|
|
#[cfg(feature = "alloc")]
|
|
pub use alloc_mod::*;
|
|
|
|
#[cfg(feature = "std")]
|
|
pub use std_mod::*;
|
|
|
|
#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)]
|
|
pub enum ListenerKey {
|
|
Single(LargestEventRaw),
|
|
Group(LargestGroupIdRaw),
|
|
All,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct EventMessage<Event: GenericEvent, ParamProvider: Debug = Params> {
|
|
sender_id: ComponentId,
|
|
event: Event,
|
|
params: Option<ParamProvider>,
|
|
}
|
|
|
|
impl<Event: GenericEvent, ParamProvider: Debug + Clone> EventMessage<Event, ParamProvider> {
|
|
pub fn new_generic(
|
|
sender_id: ComponentId,
|
|
event: Event,
|
|
params: Option<&ParamProvider>,
|
|
) -> Self {
|
|
Self {
|
|
sender_id,
|
|
event,
|
|
params: params.cloned(),
|
|
}
|
|
}
|
|
|
|
pub fn sender_id(&self) -> ComponentId {
|
|
self.sender_id
|
|
}
|
|
|
|
pub fn event(&self) -> Event {
|
|
self.event
|
|
}
|
|
|
|
pub fn params(&self) -> Option<&ParamProvider> {
|
|
self.params.as_ref()
|
|
}
|
|
|
|
pub fn new(sender_id: ComponentId, event: Event) -> Self {
|
|
Self::new_generic(sender_id, event, None)
|
|
}
|
|
|
|
pub fn new_with_params(sender_id: ComponentId, event: Event, params: &ParamProvider) -> Self {
|
|
Self::new_generic(sender_id, event, Some(params))
|
|
}
|
|
}
|
|
|
|
pub type EventMessageU32 = EventMessage<EventU32, Params>;
|
|
pub type EventMessageU16 = EventMessage<EventU16, Params>;
|
|
|
|
/// Generic abstraction
|
|
pub trait EventSendProvider<Event: GenericEvent, ParamProvider: Debug = Params> {
|
|
type Error;
|
|
|
|
fn target_id(&self) -> ComponentId;
|
|
|
|
fn send(&self, message: EventMessage<Event, ParamProvider>) -> Result<(), Self::Error>;
|
|
}
|
|
|
|
/// Generic abstraction for an event receiver.
|
|
pub trait EventReceiveProvider<Event: GenericEvent, ParamsProvider: Debug = Params> {
|
|
type Error;
|
|
|
|
/// This function has to be provided by any event receiver. A call may or may not return
|
|
/// an event and optional auxiliary data.
|
|
fn try_recv_event(&self) -> Result<Option<EventMessage<Event, ParamsProvider>>, Self::Error>;
|
|
}
|
|
|
|
pub trait ListenerMapProvider {
|
|
#[cfg(feature = "alloc")]
|
|
fn get_listeners(&self) -> alloc::vec::Vec<ListenerKey>;
|
|
fn contains_listener(&self, key: &ListenerKey) -> bool;
|
|
fn get_listener_ids(&self, key: &ListenerKey) -> Option<Iter<ComponentId>>;
|
|
fn add_listener(&mut self, key: ListenerKey, listener_id: ComponentId) -> bool;
|
|
fn remove_duplicates(&mut self, key: &ListenerKey);
|
|
}
|
|
|
|
pub trait SenderMapProvider<
|
|
EventSender: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent = EventU32,
|
|
ParamProvider: Debug = Params,
|
|
>
|
|
{
|
|
fn contains_send_event_provider(&self, target_id: &ComponentId) -> bool;
|
|
|
|
fn get_send_event_provider(&self, target_id: &ComponentId) -> Option<&EventSender>;
|
|
fn add_send_event_provider(&mut self, send_provider: EventSender) -> bool;
|
|
}
|
|
|
|
/// Generic event manager implementation.
|
|
///
|
|
/// # Generics
|
|
///
|
|
/// * `EventReceiver`: [EventReceiveProvider] used to receive all events.
|
|
/// * `SenderMap`: [SenderMapProvider] which maps channel IDs to send providers.
|
|
/// * `ListenerMap`: [ListenerMapProvider] which maps listener keys to channel IDs.
|
|
/// * `EventSender`: [EventSendProvider] contained within the sender map which sends the events.
|
|
/// * `Event`: The event type. This type must implement the [GenericEvent]. Currently only [EventU32]
|
|
/// and [EventU16] are supported.
|
|
/// * `ParamProvider`: Auxiliary data which is sent with the event to provide optional context
|
|
/// information
|
|
pub struct EventManager<
|
|
EventReceiver: EventReceiveProvider<Event, ParamProvider>,
|
|
SenderMap: SenderMapProvider<EventSender, Event, ParamProvider>,
|
|
ListenerMap: ListenerMapProvider,
|
|
EventSender: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent = EventU32,
|
|
ParamProvider: Debug = Params,
|
|
> {
|
|
event_receiver: EventReceiver,
|
|
sender_map: SenderMap,
|
|
listener_map: ListenerMap,
|
|
phantom: core::marker::PhantomData<(EventSender, Event, ParamProvider)>,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum EventRoutingResult<Event: GenericEvent, ParamProvider: Debug> {
|
|
/// No event was received
|
|
Empty,
|
|
/// An event was received and routed to listeners.
|
|
Handled {
|
|
num_recipients: u32,
|
|
event_msg: EventMessage<Event, ParamProvider>,
|
|
},
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum EventRoutingError {
|
|
Send(GenericSendError),
|
|
NoSendersForKey(ListenerKey),
|
|
NoSenderForId(ComponentId),
|
|
}
|
|
|
|
impl<
|
|
EventReceiver: EventReceiveProvider<Event, ParamProvider>,
|
|
SenderMap: SenderMapProvider<EventSender, Event, ParamProvider>,
|
|
ListenerMap: ListenerMapProvider,
|
|
EventSender: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent + Copy,
|
|
ParamProvider: Debug,
|
|
> EventManager<EventReceiver, SenderMap, ListenerMap, EventSender, Event, ParamProvider>
|
|
{
|
|
pub fn remove_duplicates(&mut self, key: &ListenerKey) {
|
|
self.listener_map.remove_duplicates(key)
|
|
}
|
|
|
|
/// Subscribe for a unique event.
|
|
pub fn subscribe_single(&mut self, event: &Event, sender_id: ComponentId) {
|
|
self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), sender_id);
|
|
}
|
|
|
|
/// Subscribe for an event group.
|
|
pub fn subscribe_group(&mut self, group_id: LargestGroupIdRaw, sender_id: ComponentId) {
|
|
self.update_listeners(ListenerKey::Group(group_id), sender_id);
|
|
}
|
|
|
|
/// Subscribe for all events received by the manager.
|
|
///
|
|
/// For example, this can be useful for a handler component which sends every event as
|
|
/// a telemetry packet.
|
|
pub fn subscribe_all(&mut self, sender_id: ComponentId) {
|
|
self.update_listeners(ListenerKey::All, sender_id);
|
|
}
|
|
}
|
|
impl<
|
|
EventReceiver: EventReceiveProvider<Event, ParamProvider>,
|
|
SenderMap: SenderMapProvider<EventSenderMap, Event, ParamProvider>,
|
|
ListenerMap: ListenerMapProvider,
|
|
EventSenderMap: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent + Copy,
|
|
ParamProvider: Debug,
|
|
> EventManager<EventReceiver, SenderMap, ListenerMap, EventSenderMap, Event, ParamProvider>
|
|
{
|
|
pub fn new_with_custom_maps(
|
|
event_receiver: EventReceiver,
|
|
sender_map: SenderMap,
|
|
listener_map: ListenerMap,
|
|
) -> Self {
|
|
EventManager {
|
|
listener_map,
|
|
sender_map,
|
|
event_receiver,
|
|
phantom: PhantomData,
|
|
}
|
|
}
|
|
|
|
/// Add a new sender component which can be used to send events to subscribers.
|
|
pub fn add_sender(&mut self, send_provider: EventSenderMap) {
|
|
if !self
|
|
.sender_map
|
|
.contains_send_event_provider(&send_provider.target_id())
|
|
{
|
|
self.sender_map.add_send_event_provider(send_provider);
|
|
}
|
|
}
|
|
|
|
/// Generic function to update the event subscribers.
|
|
fn update_listeners(&mut self, key: ListenerKey, sender_id: ComponentId) {
|
|
self.listener_map.add_listener(key, sender_id);
|
|
}
|
|
}
|
|
|
|
impl<
|
|
EventReceiver: EventReceiveProvider<Event, ParamProvider>,
|
|
SenderMap: SenderMapProvider<EventSenderMap, Event, ParamProvider>,
|
|
ListenerMap: ListenerMapProvider,
|
|
EventSenderMap: EventSendProvider<Event, ParamProvider, Error = GenericSendError>,
|
|
Event: GenericEvent + Copy,
|
|
ParamProvider: Clone + Debug,
|
|
> EventManager<EventReceiver, SenderMap, ListenerMap, EventSenderMap, Event, ParamProvider>
|
|
{
|
|
/// This function will use the cached event receiver and try to receive one event.
|
|
/// If an event was received, it will try to route that event to all subscribed event listeners.
|
|
/// If this works without any issues, the [EventRoutingResult] will contain context information
|
|
/// about the routed event.
|
|
///
|
|
/// If an error occurs during the routing, the error handler will be called. The error handler
|
|
/// should take a reference to the event message as the first argument, and the routing error
|
|
/// as the second argument.
|
|
pub fn try_event_handling<E: FnMut(&EventMessage<Event, ParamProvider>, EventRoutingError)>(
|
|
&self,
|
|
mut error_handler: E,
|
|
) -> EventRoutingResult<Event, ParamProvider> {
|
|
let mut num_recipients = 0;
|
|
let mut send_handler =
|
|
|key: &ListenerKey, event_msg: &EventMessage<Event, ParamProvider>| {
|
|
if self.listener_map.contains_listener(key) {
|
|
if let Some(ids) = self.listener_map.get_listener_ids(key) {
|
|
for id in ids {
|
|
if let Some(sender) = self.sender_map.get_send_event_provider(id) {
|
|
if let Err(e) = sender.send(EventMessage::new_generic(
|
|
event_msg.sender_id,
|
|
event_msg.event,
|
|
event_msg.params.as_ref(),
|
|
)) {
|
|
error_handler(event_msg, EventRoutingError::Send(e));
|
|
} else {
|
|
num_recipients += 1;
|
|
}
|
|
} else {
|
|
error_handler(event_msg, EventRoutingError::NoSenderForId(*id));
|
|
}
|
|
}
|
|
} else {
|
|
error_handler(event_msg, EventRoutingError::NoSendersForKey(*key));
|
|
}
|
|
}
|
|
};
|
|
if let Ok(Some(event_msg)) = self.event_receiver.try_recv_event() {
|
|
let single_key = ListenerKey::Single(event_msg.event.raw_as_largest_type());
|
|
send_handler(&single_key, &event_msg);
|
|
let group_key = ListenerKey::Group(event_msg.event.group_id_as_largest_type());
|
|
send_handler(&group_key, &event_msg);
|
|
send_handler(&ListenerKey::All, &event_msg);
|
|
return EventRoutingResult::Handled {
|
|
num_recipients,
|
|
event_msg,
|
|
};
|
|
}
|
|
EventRoutingResult::Empty
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "alloc")]
|
|
pub mod alloc_mod {
|
|
use alloc::vec::Vec;
|
|
use hashbrown::HashMap;
|
|
|
|
use super::*;
|
|
|
|
/// Helper type which constrains the sender map and listener map generics to the [DefaultSenderMap]
|
|
/// and the [DefaultListenerMap]. It uses regular mpsc channels as the message queue backend.
|
|
pub type EventManagerWithMpsc<Event = EventU32, ParamProvider = Params> = EventManager<
|
|
EventU32ReceiverMpsc<ParamProvider>,
|
|
DefaultSenderMap<EventSenderMpsc<Event>, Event, ParamProvider>,
|
|
DefaultListenerMap,
|
|
EventSenderMpsc<Event>,
|
|
>;
|
|
|
|
/// Helper type which constrains the sender map and listener map generics to the [DefaultSenderMap]
|
|
/// and the [DefaultListenerMap]. It uses
|
|
/// [bounded mpsc senders](https://doc.rust-lang.org/std/sync/mpsc/struct.SyncSender.html) as the
|
|
/// message queue backend.
|
|
pub type EventManagerWithBoundedMpsc<Event = EventU32, ParamProvider = Params> = EventManager<
|
|
EventU32ReceiverMpsc<ParamProvider>,
|
|
DefaultSenderMap<EventSenderMpscBounded<Event>, Event, ParamProvider>,
|
|
DefaultListenerMap,
|
|
EventSenderMpscBounded<Event>,
|
|
>;
|
|
|
|
impl<
|
|
EventReceiver: EventReceiveProvider<Event, ParamProvider>,
|
|
EventSender: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent + Copy,
|
|
ParamProvider: 'static + Debug,
|
|
>
|
|
EventManager<
|
|
EventReceiver,
|
|
DefaultSenderMap<EventSender, Event, ParamProvider>,
|
|
DefaultListenerMap,
|
|
EventSender,
|
|
Event,
|
|
ParamProvider,
|
|
>
|
|
{
|
|
/// Create an event manager where the sender table will be the [DefaultSenderMap]
|
|
/// and the listener table will be the [DefaultListenerMap].
|
|
pub fn new(event_receiver: EventReceiver) -> Self {
|
|
Self {
|
|
listener_map: DefaultListenerMap::default(),
|
|
sender_map: DefaultSenderMap::default(),
|
|
event_receiver,
|
|
phantom: PhantomData,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Default listener map.
|
|
///
|
|
/// Simple implementation which uses a [HashMap] and a [Vec] internally.
|
|
#[derive(Default)]
|
|
pub struct DefaultListenerMap {
|
|
listeners: HashMap<ListenerKey, Vec<ComponentId>>,
|
|
}
|
|
|
|
impl ListenerMapProvider for DefaultListenerMap {
|
|
fn get_listeners(&self) -> Vec<ListenerKey> {
|
|
let mut key_list = Vec::new();
|
|
for key in self.listeners.keys() {
|
|
key_list.push(*key);
|
|
}
|
|
key_list
|
|
}
|
|
|
|
fn contains_listener(&self, key: &ListenerKey) -> bool {
|
|
self.listeners.contains_key(key)
|
|
}
|
|
|
|
fn get_listener_ids(&self, key: &ListenerKey) -> Option<Iter<ComponentId>> {
|
|
self.listeners.get(key).map(|vec| vec.iter())
|
|
}
|
|
|
|
fn add_listener(&mut self, key: ListenerKey, sender_id: ComponentId) -> bool {
|
|
if let Some(existing_list) = self.listeners.get_mut(&key) {
|
|
existing_list.push(sender_id);
|
|
} else {
|
|
let new_list = alloc::vec![sender_id];
|
|
self.listeners.insert(key, new_list);
|
|
}
|
|
true
|
|
}
|
|
|
|
fn remove_duplicates(&mut self, key: &ListenerKey) {
|
|
if let Some(list) = self.listeners.get_mut(key) {
|
|
list.sort_unstable();
|
|
list.dedup();
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Default sender map.
|
|
///
|
|
/// Simple implementation which uses a [HashMap] internally.
|
|
pub struct DefaultSenderMap<
|
|
EventSender: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent = EventU32,
|
|
ParamProvider: Debug = Params,
|
|
> {
|
|
senders: HashMap<ComponentId, EventSender>,
|
|
phantom: PhantomData<(Event, ParamProvider)>,
|
|
}
|
|
|
|
impl<
|
|
EventSender: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent,
|
|
ParamProvider: Debug,
|
|
> Default for DefaultSenderMap<EventSender, Event, ParamProvider>
|
|
{
|
|
fn default() -> Self {
|
|
Self {
|
|
senders: Default::default(),
|
|
phantom: Default::default(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<
|
|
EventSender: EventSendProvider<Event, ParamProvider>,
|
|
Event: GenericEvent,
|
|
ParamProvider: Debug,
|
|
> SenderMapProvider<EventSender, Event, ParamProvider>
|
|
for DefaultSenderMap<EventSender, Event, ParamProvider>
|
|
{
|
|
fn contains_send_event_provider(&self, id: &ComponentId) -> bool {
|
|
self.senders.contains_key(id)
|
|
}
|
|
|
|
fn get_send_event_provider(&self, id: &ComponentId) -> Option<&EventSender> {
|
|
self.senders
|
|
.get(id)
|
|
.filter(|sender| sender.target_id() == *id)
|
|
}
|
|
|
|
fn add_send_event_provider(&mut self, send_provider: EventSender) -> bool {
|
|
let id = send_provider.target_id();
|
|
if self.senders.contains_key(&id) {
|
|
return false;
|
|
}
|
|
self.senders.insert(id, send_provider).is_none()
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
pub mod std_mod {
|
|
use crate::queue::GenericReceiveError;
|
|
|
|
use super::*;
|
|
use std::sync::mpsc;
|
|
|
|
impl<Event: GenericEvent + Send, ParamProvider: Debug>
|
|
EventReceiveProvider<Event, ParamProvider>
|
|
for mpsc::Receiver<EventMessage<Event, ParamProvider>>
|
|
{
|
|
type Error = GenericReceiveError;
|
|
|
|
fn try_recv_event(
|
|
&self,
|
|
) -> Result<Option<EventMessage<Event, ParamProvider>>, Self::Error> {
|
|
match self.try_recv() {
|
|
Ok(msg) => Ok(Some(msg)),
|
|
Err(e) => match e {
|
|
mpsc::TryRecvError::Empty => Ok(None),
|
|
mpsc::TryRecvError::Disconnected => {
|
|
Err(GenericReceiveError::TxDisconnected(None))
|
|
}
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
pub type EventU32ReceiverMpsc<ParamProvider = Params> =
|
|
mpsc::Receiver<EventMessage<EventU32, ParamProvider>>;
|
|
pub type EventU16ReceiverMpsc<ParamProvider = Params> =
|
|
mpsc::Receiver<EventMessage<EventU16, ParamProvider>>;
|
|
|
|
/// Generic event sender which uses a regular [mpsc::Sender] as the messaging backend to
|
|
/// send events.
|
|
#[derive(Clone)]
|
|
pub struct EventSenderMpsc<Event: GenericEvent + Send> {
|
|
target_id: ComponentId,
|
|
sender: mpsc::Sender<EventMessage<Event>>,
|
|
}
|
|
|
|
impl<Event: GenericEvent + Send> EventSenderMpsc<Event> {
|
|
pub fn new(target_id: ComponentId, sender: mpsc::Sender<EventMessage<Event>>) -> Self {
|
|
Self { target_id, sender }
|
|
}
|
|
}
|
|
|
|
impl<Event: GenericEvent + Send> EventSendProvider<Event> for EventSenderMpsc<Event> {
|
|
type Error = GenericSendError;
|
|
|
|
fn target_id(&self) -> ComponentId {
|
|
self.target_id
|
|
}
|
|
|
|
fn send(&self, event_msg: EventMessage<Event>) -> Result<(), GenericSendError> {
|
|
self.sender
|
|
.send(event_msg)
|
|
.map_err(|_| GenericSendError::RxDisconnected)
|
|
}
|
|
}
|
|
|
|
/// Generic event sender which uses the [mpsc::SyncSender] as the messaging backend to send
|
|
/// events. This has the advantage that the channel is bounded and thus more deterministic.
|
|
#[derive(Clone)]
|
|
pub struct EventSenderMpscBounded<Event: GenericEvent + Send> {
|
|
target_id: ComponentId,
|
|
sender: mpsc::SyncSender<EventMessage<Event>>,
|
|
capacity: usize,
|
|
}
|
|
|
|
impl<Event: GenericEvent + Send> EventSenderMpscBounded<Event> {
|
|
pub fn new(
|
|
target_id: ComponentId,
|
|
sender: mpsc::SyncSender<EventMessage<Event>>,
|
|
capacity: usize,
|
|
) -> Self {
|
|
Self {
|
|
target_id,
|
|
sender,
|
|
capacity,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<Event: GenericEvent + Send> EventSendProvider<Event> for EventSenderMpscBounded<Event> {
|
|
type Error = GenericSendError;
|
|
|
|
fn target_id(&self) -> ComponentId {
|
|
self.target_id
|
|
}
|
|
|
|
fn send(&self, event_msg: EventMessage<Event>) -> Result<(), Self::Error> {
|
|
if let Err(e) = self.sender.try_send(event_msg) {
|
|
return match e {
|
|
mpsc::TrySendError::Full(_) => {
|
|
Err(GenericSendError::QueueFull(Some(self.capacity as u32)))
|
|
}
|
|
mpsc::TrySendError::Disconnected(_) => Err(GenericSendError::RxDisconnected),
|
|
};
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
pub type EventU32SenderMpsc = EventSenderMpsc<EventU32>;
|
|
pub type EventU16SenderMpsc = EventSenderMpsc<EventU16>;
|
|
pub type EventU32SenderMpscBounded = EventSenderMpscBounded<EventU32>;
|
|
pub type EventU16SenderMpscBounded = EventSenderMpscBounded<EventU16>;
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::event_man::EventManager;
|
|
use crate::events::{EventU32, GenericEvent, Severity};
|
|
use crate::params::{ParamsHeapless, ParamsRaw};
|
|
use crate::pus::test_util::{TEST_COMPONENT_ID_0, TEST_COMPONENT_ID_1};
|
|
use std::format;
|
|
use std::sync::mpsc::{self};
|
|
|
|
const TEST_EVENT: EventU32 = EventU32::new(Severity::Info, 0, 5);
|
|
|
|
fn check_next_event(
|
|
expected: EventU32,
|
|
receiver: &mpsc::Receiver<EventMessageU32>,
|
|
) -> Option<Params> {
|
|
if let Ok(event_msg) = receiver.try_recv() {
|
|
assert_eq!(event_msg.event, expected);
|
|
return event_msg.params;
|
|
}
|
|
None
|
|
}
|
|
|
|
fn check_handled_event(
|
|
res: EventRoutingResult<EventU32, Params>,
|
|
expected: EventU32,
|
|
expected_num_sent: u32,
|
|
expected_sender_id: ComponentId,
|
|
) {
|
|
assert!(matches!(res, EventRoutingResult::Handled { .. }));
|
|
if let EventRoutingResult::Handled {
|
|
num_recipients,
|
|
event_msg,
|
|
} = res
|
|
{
|
|
assert_eq!(event_msg.event, expected);
|
|
assert_eq!(event_msg.sender_id, expected_sender_id);
|
|
assert_eq!(num_recipients, expected_num_sent);
|
|
}
|
|
}
|
|
|
|
fn generic_event_man() -> (mpsc::Sender<EventMessageU32>, EventManagerWithMpsc) {
|
|
let (event_sender, event_receiver) = mpsc::channel();
|
|
(event_sender, EventManager::new(event_receiver))
|
|
}
|
|
|
|
#[test]
|
|
fn test_basic() {
|
|
let (event_sender, mut event_man) = generic_event_man();
|
|
let event_grp_0 = EventU32::new(Severity::Info, 0, 0);
|
|
let event_grp_1_0 = EventU32::new(Severity::High, 1, 0);
|
|
let (single_event_sender, single_event_receiver) = mpsc::channel();
|
|
let single_event_listener = EventSenderMpsc::new(0, single_event_sender);
|
|
event_man.subscribe_single(&event_grp_0, single_event_listener.target_id());
|
|
event_man.add_sender(single_event_listener);
|
|
let (group_event_sender_0, group_event_receiver_0) = mpsc::channel();
|
|
let group_event_listener = EventU32SenderMpsc::new(1, group_event_sender_0);
|
|
event_man.subscribe_group(event_grp_1_0.group_id(), group_event_listener.target_id());
|
|
event_man.add_sender(group_event_listener);
|
|
|
|
let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| {
|
|
panic!("routing error occurred for event {:?}: {:?}", event_msg, e);
|
|
};
|
|
// Test event with one listener
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_grp_0))
|
|
.expect("Sending single error failed");
|
|
let res = event_man.try_event_handling(&error_handler);
|
|
check_handled_event(res, event_grp_0, 1, TEST_COMPONENT_ID_0.id());
|
|
check_next_event(event_grp_0, &single_event_receiver);
|
|
|
|
// Test event which is sent to all group listeners
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_grp_1_0))
|
|
.expect("Sending group error failed");
|
|
let res = event_man.try_event_handling(&error_handler);
|
|
check_handled_event(res, event_grp_1_0, 1, TEST_COMPONENT_ID_1.id());
|
|
check_next_event(event_grp_1_0, &group_event_receiver_0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_with_basic_params() {
|
|
let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| {
|
|
panic!("routing error occurred for event {:?}: {:?}", event_msg, e);
|
|
};
|
|
let (event_sender, mut event_man) = generic_event_man();
|
|
let event_grp_0 = EventU32::new(Severity::Info, 0, 0);
|
|
let (single_event_sender, single_event_receiver) = mpsc::channel();
|
|
let single_event_listener = EventSenderMpsc::new(0, single_event_sender);
|
|
event_man.subscribe_single(&event_grp_0, single_event_listener.target_id());
|
|
event_man.add_sender(single_event_listener);
|
|
event_sender
|
|
.send(EventMessage::new_with_params(
|
|
TEST_COMPONENT_ID_0.id(),
|
|
event_grp_0,
|
|
&Params::Heapless((2_u32, 3_u32).into()),
|
|
))
|
|
.expect("Sending group error failed");
|
|
let res = event_man.try_event_handling(&error_handler);
|
|
check_handled_event(res, event_grp_0, 1, TEST_COMPONENT_ID_0.id());
|
|
let aux = check_next_event(event_grp_0, &single_event_receiver);
|
|
assert!(aux.is_some());
|
|
let aux = aux.unwrap();
|
|
if let Params::Heapless(ParamsHeapless::Raw(ParamsRaw::U32Pair(pair))) = aux {
|
|
assert_eq!(pair.0, 2);
|
|
assert_eq!(pair.1, 3);
|
|
} else {
|
|
panic!("{}", format!("Unexpected auxiliary value type {:?}", aux));
|
|
}
|
|
}
|
|
|
|
/// Test listening for multiple groups
|
|
#[test]
|
|
fn test_multi_group() {
|
|
let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| {
|
|
panic!("routing error occurred for event {:?}: {:?}", event_msg, e);
|
|
};
|
|
let (event_sender, mut event_man) = generic_event_man();
|
|
let res = event_man.try_event_handling(error_handler);
|
|
assert!(matches!(res, EventRoutingResult::Empty));
|
|
|
|
let event_grp_0 = EventU32::new(Severity::Info, 0, 0);
|
|
let event_grp_1_0 = EventU32::new(Severity::High, 1, 0);
|
|
let (event_grp_0_sender, event_grp_0_receiver) = mpsc::channel();
|
|
let event_grp_0_and_1_listener = EventU32SenderMpsc::new(0, event_grp_0_sender);
|
|
event_man.subscribe_group(
|
|
event_grp_0.group_id(),
|
|
event_grp_0_and_1_listener.target_id(),
|
|
);
|
|
event_man.subscribe_group(
|
|
event_grp_1_0.group_id(),
|
|
event_grp_0_and_1_listener.target_id(),
|
|
);
|
|
event_man.add_sender(event_grp_0_and_1_listener);
|
|
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_grp_0))
|
|
.expect("Sending Event Group 0 failed");
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_grp_1_0))
|
|
.expect("Sendign Event Group 1 failed");
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_grp_0, 1, TEST_COMPONENT_ID_0.id());
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_grp_1_0, 1, TEST_COMPONENT_ID_1.id());
|
|
|
|
check_next_event(event_grp_0, &event_grp_0_receiver);
|
|
check_next_event(event_grp_1_0, &event_grp_0_receiver);
|
|
}
|
|
|
|
/// Test listening to the same event from multiple listeners. Also test listening
|
|
/// to both group and single events from one listener
|
|
#[test]
|
|
fn test_listening_to_same_event_and_multi_type() {
|
|
let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| {
|
|
panic!("routing error occurred for event {:?}: {:?}", event_msg, e);
|
|
};
|
|
let (event_sender, mut event_man) = generic_event_man();
|
|
let event_0 = EventU32::new(Severity::Info, 0, 5);
|
|
let event_1 = EventU32::new(Severity::High, 1, 0);
|
|
let (event_0_tx_0, event_0_rx_0) = mpsc::channel();
|
|
let (event_0_tx_1, event_0_rx_1) = mpsc::channel();
|
|
let event_listener_0 = EventU32SenderMpsc::new(0, event_0_tx_0);
|
|
let event_listener_1 = EventU32SenderMpsc::new(1, event_0_tx_1);
|
|
let event_listener_0_sender_id = event_listener_0.target_id();
|
|
event_man.subscribe_single(&event_0, event_listener_0_sender_id);
|
|
event_man.add_sender(event_listener_0);
|
|
let event_listener_1_sender_id = event_listener_1.target_id();
|
|
event_man.subscribe_single(&event_0, event_listener_1_sender_id);
|
|
event_man.add_sender(event_listener_1);
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0))
|
|
.expect("Triggering Event 0 failed");
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_0, 2, TEST_COMPONENT_ID_0.id());
|
|
check_next_event(event_0, &event_0_rx_0);
|
|
check_next_event(event_0, &event_0_rx_1);
|
|
event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id);
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0))
|
|
.expect("Triggering Event 0 failed");
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_1))
|
|
.expect("Triggering Event 1 failed");
|
|
|
|
// 3 Events messages will be sent now
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_0, 2, TEST_COMPONENT_ID_0.id());
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_1, 1, TEST_COMPONENT_ID_1.id());
|
|
// Both the single event and the group event should arrive now
|
|
check_next_event(event_0, &event_0_rx_0);
|
|
check_next_event(event_1, &event_0_rx_0);
|
|
|
|
// Do double insertion and then remove duplicates
|
|
event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id);
|
|
event_man.remove_duplicates(&ListenerKey::Group(event_1.group_id()));
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_1))
|
|
.expect("Triggering Event 1 failed");
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_1, 1, TEST_COMPONENT_ID_0.id());
|
|
}
|
|
|
|
#[test]
|
|
fn test_all_events_listener() {
|
|
let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| {
|
|
panic!("routing error occurred for event {:?}: {:?}", event_msg, e);
|
|
};
|
|
let (event_sender, event_receiver) = mpsc::channel();
|
|
let mut event_man = EventManagerWithMpsc::new(event_receiver);
|
|
let event_0 = EventU32::new(Severity::Info, 0, 5);
|
|
let event_1 = EventU32::new(Severity::High, 1, 0);
|
|
let (event_0_tx_0, all_events_rx) = mpsc::channel();
|
|
let all_events_listener = EventU32SenderMpsc::new(0, event_0_tx_0);
|
|
event_man.subscribe_all(all_events_listener.target_id());
|
|
event_man.add_sender(all_events_listener);
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0))
|
|
.expect("Triggering event 0 failed");
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_1))
|
|
.expect("Triggering event 1 failed");
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_0, 1, TEST_COMPONENT_ID_0.id());
|
|
let res = event_man.try_event_handling(error_handler);
|
|
check_handled_event(res, event_1, 1, TEST_COMPONENT_ID_1.id());
|
|
check_next_event(event_0, &all_events_rx);
|
|
check_next_event(event_1, &all_events_rx);
|
|
}
|
|
|
|
#[test]
|
|
fn test_bounded_event_sender_queue_full() {
|
|
let (event_sender, _event_receiver) = mpsc::sync_channel(3);
|
|
let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3);
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT))
|
|
.expect("sending test event failed");
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT))
|
|
.expect("sending test event failed");
|
|
event_sender
|
|
.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT))
|
|
.expect("sending test event failed");
|
|
let error = event_sender.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT));
|
|
if let Err(e) = error {
|
|
assert!(matches!(e, GenericSendError::QueueFull(Some(3))));
|
|
} else {
|
|
panic!("unexpected error {error:?}");
|
|
}
|
|
}
|
|
#[test]
|
|
fn test_bounded_event_sender_rx_dropped() {
|
|
let (event_sender, event_receiver) = mpsc::sync_channel(3);
|
|
let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3);
|
|
drop(event_receiver);
|
|
if let Err(e) = event_sender.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT)) {
|
|
assert!(matches!(e, GenericSendError::RxDisconnected));
|
|
} else {
|
|
panic!("Expected error");
|
|
}
|
|
}
|
|
}
|