Robin Mueller
de4e6183b3
All checks were successful
Rust/sat-rs/pipeline/pr-main This commit looks good
- Add new shared subcrate satrs-shared to split off some shared components not expected to change very often. - Renmame `satrs-core` to `satrs`. It is expected that sat-rs will remain the primary crate, so the core information is superfluous, and core also implies stability, which will not be the case for some time.
711 lines
27 KiB
Rust
711 lines
27 KiB
Rust
//! Event management and forwarding
|
|
//!
|
|
//! 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. One common use case for satellite systems is to offer a light-weight
|
|
//! publish-subscribe mechanism and IPC mechanism for software and hardware events which are also
|
|
//! packaged as telemetry (TM) or can trigger a system response.
|
|
//!
|
|
//! 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:
|
|
//!
|
|
//! The event manager has a listener table abstracted by the [ListenerTable], which maps
|
|
//! listener groups identified by [ListenerKey]s to a [sender ID][ChannelId].
|
|
//! It also contains a sender table abstracted by the [SenderTable] which maps these sender IDs
|
|
//! to a concrete [SendEventProvider]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 [EventReceiver] implementation. This abstraction allow to use different
|
|
//! message queue backends. A straightforward implementation where dynamic memory allocation is
|
|
//! not a big concern could use [std::sync::mpsc::channel] to do this and is provided in
|
|
//! form of the [MpscEventReceiver].
|
|
//! 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 [EventReceiver]
|
|
//! implementation so all events are routed to the manager.
|
|
//! 4. Create the [send event providers][SendEventProvider]s which allow routing events to
|
|
//! subscribers. You can now use their [sender IDs][SendEventProvider::id] 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-core/tests/pus_events.rs)
|
|
//! for a concrete example using multi-threading where events are routed to
|
|
//! different threads.
|
|
use crate::events::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw};
|
|
use crate::params::{Params, ParamsHeapless};
|
|
#[cfg(feature = "alloc")]
|
|
use alloc::boxed::Box;
|
|
#[cfg(feature = "alloc")]
|
|
use alloc::vec;
|
|
#[cfg(feature = "alloc")]
|
|
use alloc::vec::Vec;
|
|
use core::slice::Iter;
|
|
#[cfg(feature = "alloc")]
|
|
use hashbrown::HashMap;
|
|
|
|
use crate::ChannelId;
|
|
#[cfg(feature = "std")]
|
|
pub use stdmod::*;
|
|
|
|
#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)]
|
|
pub enum ListenerKey {
|
|
Single(LargestEventRaw),
|
|
Group(LargestGroupIdRaw),
|
|
All,
|
|
}
|
|
|
|
pub type EventWithHeaplessAuxData<Event> = (Event, Option<ParamsHeapless>);
|
|
pub type EventU32WithHeaplessAuxData = EventWithHeaplessAuxData<EventU32>;
|
|
pub type EventU16WithHeaplessAuxData = EventWithHeaplessAuxData<EventU16>;
|
|
|
|
pub type EventWithAuxData<Event> = (Event, Option<Params>);
|
|
pub type EventU32WithAuxData = EventWithAuxData<EventU32>;
|
|
pub type EventU16WithAuxData = EventWithAuxData<EventU16>;
|
|
|
|
pub trait SendEventProvider<Provider: GenericEvent, AuxDataProvider = Params> {
|
|
type Error;
|
|
|
|
fn id(&self) -> ChannelId;
|
|
fn send_no_data(&self, event: Provider) -> Result<(), Self::Error> {
|
|
self.send(event, None)
|
|
}
|
|
fn send(&self, event: Provider, aux_data: Option<AuxDataProvider>) -> Result<(), Self::Error>;
|
|
}
|
|
|
|
/// Generic abstraction for an event receiver.
|
|
pub trait EventReceiver<Event: GenericEvent, AuxDataProvider = Params> {
|
|
/// This function has to be provided by any event receiver. A receive call may or may not return
|
|
/// an event.
|
|
///
|
|
/// To allow returning arbitrary additional auxiliary data, a mutable slice is passed to the
|
|
/// [Self::receive] call as well. Receivers can write data to this slice, but care must be taken
|
|
/// to avoid panics due to size missmatches or out of bound writes.
|
|
fn receive(&self) -> Option<(Event, Option<AuxDataProvider>)>;
|
|
}
|
|
|
|
pub trait ListenerTable {
|
|
fn get_listeners(&self) -> Vec<ListenerKey>;
|
|
fn contains_listener(&self, key: &ListenerKey) -> bool;
|
|
fn get_listener_ids(&self, key: &ListenerKey) -> Option<Iter<ChannelId>>;
|
|
fn add_listener(&mut self, key: ListenerKey, sender_id: ChannelId) -> bool;
|
|
fn remove_duplicates(&mut self, key: &ListenerKey);
|
|
}
|
|
|
|
pub trait SenderTable<SendProviderError, Event: GenericEvent = EventU32, AuxDataProvider = Params> {
|
|
fn contains_send_event_provider(&self, id: &ChannelId) -> bool;
|
|
fn get_send_event_provider(
|
|
&self,
|
|
id: &ChannelId,
|
|
) -> Option<&dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>>;
|
|
fn add_send_event_provider(
|
|
&mut self,
|
|
send_provider: Box<
|
|
dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>,
|
|
>,
|
|
) -> bool;
|
|
}
|
|
|
|
/// Generic event manager implementation.
|
|
///
|
|
/// # Generics
|
|
///
|
|
/// * `SendProviderError`: [SendEventProvider] error type
|
|
/// * `Event`: Concrete event provider, currently either [EventU32] or [EventU16]
|
|
/// * `AuxDataProvider`: Concrete auxiliary data provider, currently either [Params] or
|
|
/// [ParamsHeapless]
|
|
pub struct EventManager<SendProviderError, Event: GenericEvent = EventU32, AuxDataProvider = Params>
|
|
{
|
|
listener_table: Box<dyn ListenerTable>,
|
|
sender_table: Box<dyn SenderTable<SendProviderError, Event, AuxDataProvider>>,
|
|
event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>,
|
|
}
|
|
|
|
/// Safety: It is safe to implement [Send] because all fields in the [EventManager] are [Send]
|
|
/// as well
|
|
#[cfg(feature = "std")]
|
|
unsafe impl<E, Event: GenericEvent + Send, AuxDataProvider: Send> Send
|
|
for EventManager<E, Event, AuxDataProvider>
|
|
{
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
pub type EventManagerWithMpscQueue<Event, AuxDataProvider> = EventManager<
|
|
std::sync::mpsc::SendError<(Event, Option<AuxDataProvider>)>,
|
|
Event,
|
|
AuxDataProvider,
|
|
>;
|
|
|
|
#[derive(Debug)]
|
|
pub enum EventRoutingResult<Event: GenericEvent, AuxDataProvider> {
|
|
/// No event was received
|
|
Empty,
|
|
/// An event was received and routed.
|
|
/// The first tuple entry will contain the number of recipients.
|
|
Handled(u32, Event, Option<AuxDataProvider>),
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum EventRoutingError<E> {
|
|
SendError(E),
|
|
NoSendersForKey(ListenerKey),
|
|
NoSenderForId(ChannelId),
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct EventRoutingErrorsWithResult<Event: GenericEvent, AuxDataProvider, E> {
|
|
pub result: EventRoutingResult<Event, AuxDataProvider>,
|
|
pub errors: [Option<EventRoutingError<E>>; 3],
|
|
}
|
|
|
|
impl<E, Event: GenericEvent + Copy> EventManager<E, Event> {
|
|
pub fn remove_duplicates(&mut self, key: &ListenerKey) {
|
|
self.listener_table.remove_duplicates(key)
|
|
}
|
|
|
|
/// Subscribe for a unique event.
|
|
pub fn subscribe_single(&mut self, event: &Event, sender_id: ChannelId) {
|
|
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: ChannelId) {
|
|
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: ChannelId) {
|
|
self.update_listeners(ListenerKey::All, sender_id);
|
|
}
|
|
}
|
|
|
|
impl<E: 'static, Event: GenericEvent + Copy + 'static, AuxDataProvider: Clone + 'static>
|
|
EventManager<E, Event, AuxDataProvider>
|
|
{
|
|
/// Create an event manager where the sender table will be the [DefaultSenderTableProvider]
|
|
/// and the listener table will be the [DefaultListenerTableProvider].
|
|
pub fn new(event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>) -> Self {
|
|
let listener_table: Box<DefaultListenerTableProvider> = Box::default();
|
|
let sender_table: Box<DefaultSenderTableProvider<E, Event, AuxDataProvider>> =
|
|
Box::default();
|
|
Self::new_custom_tables(listener_table, sender_table, event_receiver)
|
|
}
|
|
}
|
|
|
|
impl<E, Event: GenericEvent + Copy, AuxDataProvider: Clone>
|
|
EventManager<E, Event, AuxDataProvider>
|
|
{
|
|
pub fn new_custom_tables(
|
|
listener_table: Box<dyn ListenerTable>,
|
|
sender_table: Box<dyn SenderTable<E, Event, AuxDataProvider>>,
|
|
event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>,
|
|
) -> Self {
|
|
EventManager {
|
|
listener_table,
|
|
sender_table,
|
|
event_receiver,
|
|
}
|
|
}
|
|
|
|
pub fn add_sender(
|
|
&mut self,
|
|
send_provider: impl SendEventProvider<Event, AuxDataProvider, Error = E> + 'static,
|
|
) {
|
|
if !self
|
|
.sender_table
|
|
.contains_send_event_provider(&send_provider.id())
|
|
{
|
|
self.sender_table
|
|
.add_send_event_provider(Box::new(send_provider));
|
|
}
|
|
}
|
|
|
|
fn update_listeners(&mut self, key: ListenerKey, sender_id: ChannelId) {
|
|
self.listener_table.add_listener(key, sender_id);
|
|
}
|
|
|
|
/// 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.
|
|
///
|
|
/// This function will track up to 3 errors returned as part of the
|
|
/// [EventRoutingErrorsWithResult] error struct.
|
|
pub fn try_event_handling(
|
|
&self,
|
|
) -> Result<
|
|
EventRoutingResult<Event, AuxDataProvider>,
|
|
EventRoutingErrorsWithResult<Event, AuxDataProvider, E>,
|
|
> {
|
|
let mut err_idx = 0;
|
|
let mut err_slice = [None, None, None];
|
|
let mut num_recipients = 0;
|
|
let mut add_error = |error: EventRoutingError<E>| {
|
|
if err_idx < 3 {
|
|
err_slice[err_idx] = Some(error);
|
|
err_idx += 1;
|
|
}
|
|
};
|
|
let mut send_handler =
|
|
|key: &ListenerKey, event: Event, aux_data: &Option<AuxDataProvider>| {
|
|
if self.listener_table.contains_listener(key) {
|
|
if let Some(ids) = self.listener_table.get_listener_ids(key) {
|
|
for id in ids {
|
|
if let Some(sender) = self.sender_table.get_send_event_provider(id) {
|
|
if let Err(e) = sender.send(event, aux_data.clone()) {
|
|
add_error(EventRoutingError::SendError(e));
|
|
} else {
|
|
num_recipients += 1;
|
|
}
|
|
} else {
|
|
add_error(EventRoutingError::NoSenderForId(*id));
|
|
}
|
|
}
|
|
} else {
|
|
add_error(EventRoutingError::NoSendersForKey(*key));
|
|
}
|
|
}
|
|
};
|
|
if let Some((event, aux_data)) = self.event_receiver.receive() {
|
|
let single_key = ListenerKey::Single(event.raw_as_largest_type());
|
|
send_handler(&single_key, event, &aux_data);
|
|
let group_key = ListenerKey::Group(event.group_id_as_largest_type());
|
|
send_handler(&group_key, event, &aux_data);
|
|
send_handler(&ListenerKey::All, event, &aux_data);
|
|
if err_idx > 0 {
|
|
return Err(EventRoutingErrorsWithResult {
|
|
result: EventRoutingResult::Handled(num_recipients, event, aux_data),
|
|
errors: err_slice,
|
|
});
|
|
}
|
|
return Ok(EventRoutingResult::Handled(num_recipients, event, aux_data));
|
|
}
|
|
Ok(EventRoutingResult::Empty)
|
|
}
|
|
}
|
|
|
|
#[derive(Default)]
|
|
pub struct DefaultListenerTableProvider {
|
|
listeners: HashMap<ListenerKey, Vec<ChannelId>>,
|
|
}
|
|
|
|
pub struct DefaultSenderTableProvider<
|
|
SendProviderError,
|
|
Event: GenericEvent = EventU32,
|
|
AuxDataProvider = Params,
|
|
> {
|
|
senders: HashMap<
|
|
ChannelId,
|
|
Box<dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>>,
|
|
>,
|
|
}
|
|
|
|
impl<SendProviderError, Event: GenericEvent, AuxDataProvider> Default
|
|
for DefaultSenderTableProvider<SendProviderError, Event, AuxDataProvider>
|
|
{
|
|
fn default() -> Self {
|
|
Self {
|
|
senders: HashMap::new(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl ListenerTable for DefaultListenerTableProvider {
|
|
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<ChannelId>> {
|
|
self.listeners.get(key).map(|vec| vec.iter())
|
|
}
|
|
|
|
fn add_listener(&mut self, key: ListenerKey, sender_id: ChannelId) -> bool {
|
|
if let Some(existing_list) = self.listeners.get_mut(&key) {
|
|
existing_list.push(sender_id);
|
|
} else {
|
|
let new_list = 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();
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<SendProviderError, Event: GenericEvent, AuxDataProvider>
|
|
SenderTable<SendProviderError, Event, AuxDataProvider>
|
|
for DefaultSenderTableProvider<SendProviderError, Event, AuxDataProvider>
|
|
{
|
|
fn contains_send_event_provider(&self, id: &ChannelId) -> bool {
|
|
self.senders.contains_key(id)
|
|
}
|
|
|
|
fn get_send_event_provider(
|
|
&self,
|
|
id: &ChannelId,
|
|
) -> Option<&dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>> {
|
|
self.senders
|
|
.get(id)
|
|
.filter(|sender| sender.id() == *id)
|
|
.map(|v| v.as_ref())
|
|
}
|
|
|
|
fn add_send_event_provider(
|
|
&mut self,
|
|
send_provider: Box<
|
|
dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>,
|
|
>,
|
|
) -> bool {
|
|
let id = send_provider.id();
|
|
if self.senders.contains_key(&id) {
|
|
return false;
|
|
}
|
|
self.senders.insert(id, send_provider).is_none()
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
pub mod stdmod {
|
|
use super::*;
|
|
use crate::event_man::{EventReceiver, EventWithAuxData};
|
|
use crate::events::{EventU16, EventU32, GenericEvent};
|
|
use crate::params::Params;
|
|
use std::sync::mpsc::{Receiver, SendError, Sender};
|
|
|
|
pub struct MpscEventReceiver<Event: GenericEvent + Send = EventU32> {
|
|
mpsc_receiver: Receiver<(Event, Option<Params>)>,
|
|
}
|
|
|
|
impl<Event: GenericEvent + Send> MpscEventReceiver<Event> {
|
|
pub fn new(receiver: Receiver<(Event, Option<Params>)>) -> Self {
|
|
Self {
|
|
mpsc_receiver: receiver,
|
|
}
|
|
}
|
|
}
|
|
impl<Event: GenericEvent + Send> EventReceiver<Event> for MpscEventReceiver<Event> {
|
|
fn receive(&self) -> Option<EventWithAuxData<Event>> {
|
|
if let Ok(event_and_data) = self.mpsc_receiver.try_recv() {
|
|
return Some(event_and_data);
|
|
}
|
|
None
|
|
}
|
|
}
|
|
|
|
pub type MpscEventU32Receiver = MpscEventReceiver<EventU32>;
|
|
pub type MpscEventU16Receiver = MpscEventReceiver<EventU16>;
|
|
|
|
#[derive(Clone)]
|
|
pub struct MpscEventSendProvider<Event: GenericEvent + Send> {
|
|
id: u32,
|
|
sender: Sender<(Event, Option<Params>)>,
|
|
}
|
|
|
|
impl<Event: GenericEvent + Send> MpscEventSendProvider<Event> {
|
|
pub fn new(id: u32, sender: Sender<(Event, Option<Params>)>) -> Self {
|
|
Self { id, sender }
|
|
}
|
|
}
|
|
|
|
impl<Event: GenericEvent + Send> SendEventProvider<Event> for MpscEventSendProvider<Event> {
|
|
type Error = SendError<(Event, Option<Params>)>;
|
|
|
|
fn id(&self) -> u32 {
|
|
self.id
|
|
}
|
|
fn send(&self, event: Event, aux_data: Option<Params>) -> Result<(), Self::Error> {
|
|
self.sender.send((event, aux_data))
|
|
}
|
|
}
|
|
|
|
pub type MpscEventU32SendProvider = MpscEventSendProvider<EventU32>;
|
|
pub type MpscEventU16SendProvider = MpscEventSendProvider<EventU16>;
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::event_man::EventManager;
|
|
use crate::events::{EventU32, GenericEvent, Severity};
|
|
use crate::params::ParamsRaw;
|
|
use alloc::boxed::Box;
|
|
use std::format;
|
|
use std::sync::mpsc::{channel, Receiver, SendError, Sender};
|
|
|
|
#[derive(Clone)]
|
|
struct MpscEventSenderQueue {
|
|
id: u32,
|
|
mpsc_sender: Sender<EventU32WithAuxData>,
|
|
}
|
|
|
|
impl MpscEventSenderQueue {
|
|
fn new(id: u32, mpsc_sender: Sender<EventU32WithAuxData>) -> Self {
|
|
Self { id, mpsc_sender }
|
|
}
|
|
}
|
|
|
|
impl SendEventProvider<EventU32> for MpscEventSenderQueue {
|
|
type Error = SendError<EventU32WithAuxData>;
|
|
|
|
fn id(&self) -> u32 {
|
|
self.id
|
|
}
|
|
fn send(&self, event: EventU32, aux_data: Option<Params>) -> Result<(), Self::Error> {
|
|
self.mpsc_sender.send((event, aux_data))
|
|
}
|
|
}
|
|
|
|
fn check_next_event(
|
|
expected: EventU32,
|
|
receiver: &Receiver<EventU32WithAuxData>,
|
|
) -> Option<Params> {
|
|
if let Ok(event) = receiver.try_recv() {
|
|
assert_eq!(event.0, expected);
|
|
return event.1;
|
|
}
|
|
None
|
|
}
|
|
|
|
fn check_handled_event(
|
|
res: EventRoutingResult<EventU32, Params>,
|
|
expected: EventU32,
|
|
expected_num_sent: u32,
|
|
) {
|
|
assert!(matches!(res, EventRoutingResult::Handled { .. }));
|
|
if let EventRoutingResult::Handled(num_recipients, event, _aux_data) = res {
|
|
assert_eq!(event, expected);
|
|
assert_eq!(num_recipients, expected_num_sent);
|
|
}
|
|
}
|
|
|
|
fn generic_event_man() -> (
|
|
Sender<EventU32WithAuxData>,
|
|
EventManager<SendError<EventU32WithAuxData>>,
|
|
) {
|
|
let (event_sender, manager_queue) = channel();
|
|
let event_man_receiver = MpscEventReceiver::new(manager_queue);
|
|
(
|
|
event_sender,
|
|
EventManager::new(Box::new(event_man_receiver)),
|
|
)
|
|
}
|
|
|
|
#[test]
|
|
fn test_basic() {
|
|
let (event_sender, mut event_man) = generic_event_man();
|
|
let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap();
|
|
let event_grp_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
|
|
let (single_event_sender, single_event_receiver) = channel();
|
|
let single_event_listener = MpscEventSenderQueue::new(0, single_event_sender);
|
|
event_man.subscribe_single(&event_grp_0, single_event_listener.id());
|
|
event_man.add_sender(single_event_listener);
|
|
let (group_event_sender_0, group_event_receiver_0) = channel();
|
|
let group_event_listener = MpscEventSenderQueue {
|
|
id: 1,
|
|
mpsc_sender: group_event_sender_0,
|
|
};
|
|
event_man.subscribe_group(event_grp_1_0.group_id(), group_event_listener.id());
|
|
event_man.add_sender(group_event_listener);
|
|
|
|
// Test event with one listener
|
|
event_sender
|
|
.send((event_grp_0, None))
|
|
.expect("Sending single error failed");
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_grp_0, 1);
|
|
check_next_event(event_grp_0, &single_event_receiver);
|
|
|
|
// Test event which is sent to all group listeners
|
|
event_sender
|
|
.send((event_grp_1_0, None))
|
|
.expect("Sending group error failed");
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_grp_1_0, 1);
|
|
check_next_event(event_grp_1_0, &group_event_receiver_0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_with_basic_aux_data() {
|
|
let (event_sender, mut event_man) = generic_event_man();
|
|
let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap();
|
|
let (single_event_sender, single_event_receiver) = channel();
|
|
let single_event_listener = MpscEventSenderQueue::new(0, single_event_sender);
|
|
event_man.subscribe_single(&event_grp_0, single_event_listener.id());
|
|
event_man.add_sender(single_event_listener);
|
|
event_sender
|
|
.send((event_grp_0, Some(Params::Heapless((2_u32, 3_u32).into()))))
|
|
.expect("Sending group error failed");
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_grp_0, 1);
|
|
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 (event_sender, mut event_man) = generic_event_man();
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
let hres = res.unwrap();
|
|
assert!(matches!(hres, EventRoutingResult::Empty));
|
|
|
|
let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap();
|
|
let event_grp_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
|
|
let (event_grp_0_sender, event_grp_0_receiver) = channel();
|
|
let event_grp_0_and_1_listener = MpscEventSenderQueue {
|
|
id: 0,
|
|
mpsc_sender: event_grp_0_sender,
|
|
};
|
|
event_man.subscribe_group(event_grp_0.group_id(), event_grp_0_and_1_listener.id());
|
|
event_man.subscribe_group(event_grp_1_0.group_id(), event_grp_0_and_1_listener.id());
|
|
event_man.add_sender(event_grp_0_and_1_listener);
|
|
|
|
event_sender
|
|
.send((event_grp_0, None))
|
|
.expect("Sending Event Group 0 failed");
|
|
event_sender
|
|
.send((event_grp_1_0, None))
|
|
.expect("Sendign Event Group 1 failed");
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_grp_0, 1);
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_grp_1_0, 1);
|
|
|
|
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 (event_sender, mut event_man) = generic_event_man();
|
|
let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap();
|
|
let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
|
|
let (event_0_tx_0, event_0_rx_0) = channel();
|
|
let (event_0_tx_1, event_0_rx_1) = channel();
|
|
let event_listener_0 = MpscEventSenderQueue {
|
|
id: 0,
|
|
mpsc_sender: event_0_tx_0,
|
|
};
|
|
let event_listener_1 = MpscEventSenderQueue {
|
|
id: 1,
|
|
mpsc_sender: event_0_tx_1,
|
|
};
|
|
let event_listener_0_sender_id = event_listener_0.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.id();
|
|
event_man.subscribe_single(&event_0, event_listener_1_sender_id);
|
|
event_man.add_sender(event_listener_1);
|
|
event_sender
|
|
.send((event_0, None))
|
|
.expect("Triggering Event 0 failed");
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_0, 2);
|
|
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((event_0, None))
|
|
.expect("Triggering Event 0 failed");
|
|
event_sender
|
|
.send((event_1, None))
|
|
.expect("Triggering Event 1 failed");
|
|
|
|
// 3 Events messages will be sent now
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_0, 2);
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_1, 1);
|
|
// 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((event_1, None))
|
|
.expect("Triggering Event 1 failed");
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_1, 1);
|
|
}
|
|
|
|
#[test]
|
|
fn test_all_events_listener() {
|
|
let (event_sender, manager_queue) = channel();
|
|
let event_man_receiver = MpscEventReceiver::new(manager_queue);
|
|
let mut event_man: EventManager<SendError<EventU32WithAuxData>> =
|
|
EventManager::new(Box::new(event_man_receiver));
|
|
let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap();
|
|
let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
|
|
let (event_0_tx_0, all_events_rx) = channel();
|
|
let all_events_listener = MpscEventSenderQueue {
|
|
id: 0,
|
|
mpsc_sender: event_0_tx_0,
|
|
};
|
|
event_man.subscribe_all(all_events_listener.id());
|
|
event_man.add_sender(all_events_listener);
|
|
event_sender
|
|
.send((event_0, None))
|
|
.expect("Triggering event 0 failed");
|
|
event_sender
|
|
.send((event_1, None))
|
|
.expect("Triggering event 1 failed");
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_0, 1);
|
|
let res = event_man.try_event_handling();
|
|
assert!(res.is_ok());
|
|
check_handled_event(res.unwrap(), event_1, 1);
|
|
check_next_event(event_0, &all_events_rx);
|
|
check_next_event(event_1, &all_events_rx);
|
|
}
|
|
}
|