rust/sat-rs
rust/sat-rs
5
1
Fork 0
Code Issues 9 Pull Requests Projects Releases 22 Wiki Activity

improve event handling module
Some checks failed
Rust/sat-rs/pipeline/pr-main There was a failure building this commit
Details

Browse Source
This commit is contained in:
Robin Müller 2024-04-03 11:28:20 +02:00
parent 798987cf55
commit 33b701b3e7
Signed by: muellerr
GPG Key ID: A649FB78196E3849
7 changed files with 333 additions and 237 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
satrs-example/src/events.rs
View File

@ -1,6 +1,8 @@
use std::sync::mpsc::{self}; use std::sync::mpsc::{self};
use crate::pus::create_verification_reporter; use crate::pus::create_verification_reporter;
use satrs::event_man::{EventMessageU32, EventRoutingError};
use satrs::params::WritableToBeBytes;
use satrs::pus::verification::VerificationReporter; use satrs::pus::verification::VerificationReporter;
use satrs::pus::EcssTmSenderCore; use satrs::pus::EcssTmSenderCore;
use satrs::{ use satrs::{
@ -8,8 +10,6 @@ use satrs::{
EventManagerWithBoundedMpsc, EventSendProvider, EventU32SenderMpscBounded, EventManagerWithBoundedMpsc, EventSendProvider, EventU32SenderMpscBounded,
MpscEventReceiver, MpscEventReceiver,
}, },
events::EventU32,
params::Params,
pus::{ pus::{
event_man::{ event_man::{
DefaultPusEventU32Dispatcher, EventReporter, EventRequest, EventRequestWithToken, DefaultPusEventU32Dispatcher, EventReporter, EventRequest, EventRequestWithToken,
@ -28,7 +28,7 @@ use crate::update_time;
pub struct PusEventHandler<TmSender: EcssTmSenderCore> { pub struct PusEventHandler<TmSender: EcssTmSenderCore> {
event_request_rx: mpsc::Receiver<EventRequestWithToken>, event_request_rx: mpsc::Receiver<EventRequestWithToken>,
pus_event_dispatcher: DefaultPusEventU32Dispatcher<()>, pus_event_dispatcher: DefaultPusEventU32Dispatcher<()>,
pus_event_man_rx: mpsc::Receiver<(EventU32, Option<Params>)>, pus_event_man_rx: mpsc::Receiver<EventMessageU32>,
tm_sender: TmSender, tm_sender: TmSender,
time_provider: CdsTime, time_provider: CdsTime,
timestamp: [u8; 7], timestamp: [u8; 7],
@ -108,10 +108,18 @@ impl<TmSender: EcssTmSenderCore> PusEventHandler<TmSender> {
pub fn generate_pus_event_tm(&mut self) { pub fn generate_pus_event_tm(&mut self) {
// Perform the generation of PUS event packets // Perform the generation of PUS event packets
if let Ok((event, _param)) = self.pus_event_man_rx.try_recv() { if let Ok(event_msg) = self.pus_event_man_rx.try_recv() {
update_time(&mut self.time_provider, &mut self.timestamp); update_time(&mut self.time_provider, &mut self.timestamp);
let param_vec = event_msg.params().map_or(Vec::new(), |param| {
param.to_vec().expect("failed to convert params to vec")
});
self.pus_event_dispatcher self.pus_event_dispatcher
.generate_pus_event_tm_generic(&self.tm_sender, &self.timestamp, event, None) .generate_pus_event_tm_generic(
&self.tm_sender,
&self.timestamp,
event_msg.event(),
Some(&param_vec),
)
.expect("Sending TM as event failed"); .expect("Sending TM as event failed");
} }
} }
@ -121,7 +129,7 @@ impl<TmSender: EcssTmSenderCore> PusEventHandler<TmSender> {
/// used to send events to the event manager. /// used to send events to the event manager.
pub struct EventManagerWrapper { pub struct EventManagerWrapper {
event_manager: EventManagerWithBoundedMpsc, event_manager: EventManagerWithBoundedMpsc,
event_sender: mpsc::Sender<(EventU32, Option<Params>)>, event_sender: mpsc::Sender<EventMessageU32>,
} }
impl EventManagerWrapper { impl EventManagerWrapper {
@ -129,7 +137,7 @@ impl EventManagerWrapper {
// The sender handle is the primary sender handle for all components which want to create events. // The sender handle is the primary sender handle for all components which want to create events.
// The event manager will receive the RX handle to receive all the events. // The event manager will receive the RX handle to receive all the events.
let (event_sender, event_man_rx) = mpsc::channel(); let (event_sender, event_man_rx) = mpsc::channel();
let event_recv = MpscEventReceiver::<EventU32>::new(event_man_rx); let event_recv = MpscEventReceiver::new(event_man_rx);
Self { Self {
event_manager: EventManagerWithBoundedMpsc::new(event_recv), event_manager: EventManagerWithBoundedMpsc::new(event_recv),
event_sender, event_sender,
@ -137,7 +145,7 @@ impl EventManagerWrapper {
} }
// Returns a cached event sender to send events to the event manager for routing. // Returns a cached event sender to send events to the event manager for routing.
pub fn clone_event_sender(&self) -> mpsc::Sender<(EventU32, Option<Params>)> { pub fn clone_event_sender(&self) -> mpsc::Sender<EventMessageU32> {
self.event_sender.clone() self.event_sender.clone()
} }
@ -146,10 +154,13 @@ impl EventManagerWrapper {
} }
pub fn try_event_routing(&mut self) { pub fn try_event_routing(&mut self) {
let error_handler = |error| self.routing_error_handler(error);
// Perform the event routing. // Perform the event routing.
self.event_manager self.event_manager.try_event_handling(error_handler);
.try_event_handling() }
.expect("event handling failed");
pub fn routing_error_handler(&self, error: EventRoutingError) {
log::warn!("event routing error: {error:?}");
} }
} }
@ -176,7 +187,7 @@ impl<TmSender: EcssTmSenderCore> EventHandler<TmSender> {
} }
} }
pub fn clone_event_sender(&self) -> mpsc::Sender<(EventU32, Option<Params>)> { pub fn clone_event_sender(&self) -> mpsc::Sender<EventMessageU32> {
self.event_man_wrapper.clone_event_sender() self.event_man_wrapper.clone_event_sender()
} }

12
satrs-example/src/pus/test.rs
View File

@ -1,9 +1,10 @@
use crate::pus::create_verification_reporter; use crate::pus::create_verification_reporter;
use log::{info, warn}; use log::{info, warn};
use satrs::params::Params; use satrs::event_man::{EventMessage, EventMessageU32};
use satrs::pool::SharedStaticMemoryPool; use satrs::pool::SharedStaticMemoryPool;
use satrs::pus::test::PusService17TestHandler; use satrs::pus::test::PusService17TestHandler;
use satrs::pus::verification::{FailParams, VerificationReporter, VerificationReportingProvider}; use satrs::pus::verification::{FailParams, VerificationReporter, VerificationReportingProvider};
use satrs::pus::EcssTcInSharedStoreConverter;
use satrs::pus::{ use satrs::pus::{
EcssTcAndToken, EcssTcInMemConverter, EcssTcInVecConverter, EcssTmSenderCore, MpscTcReceiver, EcssTcAndToken, EcssTcInMemConverter, EcssTcInVecConverter, EcssTmSenderCore, MpscTcReceiver,
MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded, PusPacketHandlerResult, PusServiceHelper, MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded, PusPacketHandlerResult, PusServiceHelper,
@ -13,7 +14,6 @@ use satrs::spacepackets::ecss::tc::PusTcReader;
use satrs::spacepackets::ecss::PusPacket; use satrs::spacepackets::ecss::PusPacket;
use satrs::spacepackets::time::cds::CdsTime; use satrs::spacepackets::time::cds::CdsTime;
use satrs::spacepackets::time::TimeWriter; use satrs::spacepackets::time::TimeWriter;
use satrs::{events::EventU32, pus::EcssTcInSharedStoreConverter};
use satrs_example::config::components::PUS_TEST_SERVICE; use satrs_example::config::components::PUS_TEST_SERVICE;
use satrs_example::config::{tmtc_err, TEST_EVENT}; use satrs_example::config::{tmtc_err, TEST_EVENT};
use std::sync::mpsc; use std::sync::mpsc;
@ -21,7 +21,7 @@ use std::sync::mpsc;
pub fn create_test_service_static( pub fn create_test_service_static(
tm_sender: TmInSharedPoolSender<mpsc::SyncSender<PusTmInPool>>, tm_sender: TmInSharedPoolSender<mpsc::SyncSender<PusTmInPool>>,
tc_pool: SharedStaticMemoryPool, tc_pool: SharedStaticMemoryPool,
event_sender: mpsc::Sender<(EventU32, Option<Params>)>, event_sender: mpsc::Sender<EventMessageU32>,
pus_test_rx: mpsc::Receiver<EcssTcAndToken>, pus_test_rx: mpsc::Receiver<EcssTcAndToken>,
) -> TestCustomServiceWrapper<MpscTmInSharedPoolSenderBounded, EcssTcInSharedStoreConverter> { ) -> TestCustomServiceWrapper<MpscTmInSharedPoolSenderBounded, EcssTcInSharedStoreConverter> {
let pus17_handler = PusService17TestHandler::new(PusServiceHelper::new( let pus17_handler = PusService17TestHandler::new(PusServiceHelper::new(
@ -39,7 +39,7 @@ pub fn create_test_service_static(
pub fn create_test_service_dynamic( pub fn create_test_service_dynamic(
tm_funnel_tx: mpsc::Sender<PusTmAsVec>, tm_funnel_tx: mpsc::Sender<PusTmAsVec>,
event_sender: mpsc::Sender<(EventU32, Option<Params>)>, event_sender: mpsc::Sender<EventMessageU32>,
pus_test_rx: mpsc::Receiver<EcssTcAndToken>, pus_test_rx: mpsc::Receiver<EcssTcAndToken>,
) -> TestCustomServiceWrapper<MpscTmAsVecSender, EcssTcInVecConverter> { ) -> TestCustomServiceWrapper<MpscTmAsVecSender, EcssTcInVecConverter> {
let pus17_handler = PusService17TestHandler::new(PusServiceHelper::new( let pus17_handler = PusService17TestHandler::new(PusServiceHelper::new(
@ -61,7 +61,7 @@ pub struct TestCustomServiceWrapper<
> { > {
pub handler: pub handler:
PusService17TestHandler<MpscTcReceiver, TmSender, TcInMemConverter, VerificationReporter>, PusService17TestHandler<MpscTcReceiver, TmSender, TcInMemConverter, VerificationReporter>,
pub test_srv_event_sender: mpsc::Sender<(EventU32, Option<Params>)>, pub test_srv_event_sender: mpsc::Sender<EventMessageU32>,
} }
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter> impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
@ -101,7 +101,7 @@ impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
if subservice == 128 { if subservice == 128 {
info!("Generating test event"); info!("Generating test event");
self.test_srv_event_sender self.test_srv_event_sender
.send((TEST_EVENT.into(), None)) .send(EventMessage::new(PUS_TEST_SERVICE.id(), TEST_EVENT.into()))
.expect("Sending test event failed"); .expect("Sending test event failed");
let start_token = self let start_token = self
.handler .handler

5
satrs/CHANGELOG.md
View File

@ -16,9 +16,14 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
- Various abstraction and objects for targeted requests. This includes mode request/reply - Various abstraction and objects for targeted requests. This includes mode request/reply
types for actions, HK and modes. types for actions, HK and modes.
- `VerificationReportingProvider::owner_id` method. - `VerificationReportingProvider::owner_id` method.
- Introduced generic `EventMessage` which is generic over the event type and the additional
parameter type. This message also contains the sender ID which can be useful for debugging
or application layer / FDIR logic.
## Changed ## Changed
- `EventManager::try_event_handling` not expects a mutable error handling closure instead of
returning the occured errors.
- Renamed `EventManagerBase` to `EventReportCreator` - Renamed `EventManagerBase` to `EventReportCreator`
- Renamed `VerificationReporterCore` to `VerificationReportCreator`. - Renamed `VerificationReporterCore` to `VerificationReportCreator`.
- Removed `VerificationReporterCore`. The high-level API exposed by `VerificationReporter` and - Removed `VerificationReporterCore`. The high-level API exposed by `VerificationReporter` and

469
satrs/src/event_man.rs
View File

@ -11,7 +11,7 @@
//! about events first: //! about events first:
//! //!
//! The event manager has a listener table abstracted by the [ListenerMapProvider], which maps //! The event manager has a listener table abstracted by the [ListenerMapProvider], which maps
//! listener groups identified by [ListenerKey]s to a [sender ID][ComponentId]. //! 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 //! 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 //! 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 //! for each OBSW thread and then subscribe for all interesting events for a particular thread
@ -28,8 +28,8 @@
//! manager. //! manager.
//! 3. The event manager receives the receiver component as part of a [EventReceiveProvider] //! 3. The event manager receives the receiver component as part of a [EventReceiveProvider]
//! implementation so all events are routed to the manager. //! implementation so all events are routed to the manager.
//! 4. Create the [send event providers][EventSendProvider]s which allow routing events to //! 4. Create the [event sender map][SenderMapProvider]s which allow routing events to
//! subscribers. You can now use their [sender IDs][EventSendProvider::channel_id] to subscribe //! subscribers. You can now use the subscriber component IDs to subscribe
//! for event groups, for example by using the [EventManager::subscribe_single] method. //! 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 //! 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. //! manager can route listener groups to a the send provider.
@ -45,8 +45,9 @@
//! for a concrete example using multi-threading where events are routed to //! for a concrete example using multi-threading where events are routed to
//! different threads. //! different threads.
use crate::events::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw}; use crate::events::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw};
use crate::params::{Params, ParamsHeapless}; use crate::params::Params;
use crate::queue::GenericSendError; use crate::queue::GenericSendError;
use core::fmt::Debug;
use core::marker::PhantomData; use core::marker::PhantomData;
use core::slice::Iter; use core::slice::Iter;
@ -65,29 +66,70 @@ pub enum ListenerKey {
All, All,
} }
pub type EventWithHeaplessAuxData<Event> = (Event, Option<ParamsHeapless>); // pub type EventWithAuxData<Event> = (Event, Option<Params>);
pub type EventU32WithHeaplessAuxData = EventWithHeaplessAuxData<EventU32>; // pub type EventU32WithAuxData = EventWithAuxData<EventU32>;
pub type EventU16WithHeaplessAuxData = EventWithHeaplessAuxData<EventU16>; // pub type EventU16WithAuxData = EventWithAuxData<EventU16>;
pub type EventWithAuxData<Event> = (Event, Option<Params>); #[derive(Debug)]
pub type EventU32WithAuxData = EventWithAuxData<EventU32>; pub struct EventMessage<Event: GenericEvent, ParamProvider: Debug = Params> {
pub type EventU16WithAuxData = EventWithAuxData<EventU16>; sender_id: ComponentId,
event: Event,
params: Option<ParamProvider>,
}
pub trait EventSendProvider<EV: GenericEvent, AuxDataProvider = Params> { impl<Event: GenericEvent, ParamProvider: Debug> EventMessage<Event, ParamProvider> {
fn target_id(&self) -> ComponentId; pub fn new_generic(
sender_id: ComponentId,
fn send_no_data(&self, event: EV) -> Result<(), GenericSendError> { event: Event,
self.send(event, None) params: Option<ParamProvider>,
) -> Self {
Self {
sender_id,
event,
params,
}
} }
fn send(&self, event: EV, aux_data: Option<AuxDataProvider>) -> Result<(), GenericSendError>; 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. /// Generic abstraction for an event receiver.
pub trait EventReceiveProvider<Event: GenericEvent, AuxDataProvider = Params> { 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 /// This function has to be provided by any event receiver. A call may or may not return
/// an event and optional auxiliary data. /// an event and optional auxiliary data.
fn try_recv_event(&self) -> Option<(Event, Option<AuxDataProvider>)>; fn try_recv_event(&self) -> Result<Option<EventMessage<Event, ParamsProvider>>, Self::Error>;
} }
pub trait ListenerMapProvider { pub trait ListenerMapProvider {
@ -96,14 +138,14 @@ pub trait ListenerMapProvider {
fn get_listeners(&self) -> alloc::vec::Vec<ListenerKey>; fn get_listeners(&self) -> alloc::vec::Vec<ListenerKey>;
fn contains_listener(&self, key: &ListenerKey) -> bool; fn contains_listener(&self, key: &ListenerKey) -> bool;
fn get_listener_ids(&self, key: &ListenerKey) -> Option<Iter<ComponentId>>; fn get_listener_ids(&self, key: &ListenerKey) -> Option<Iter<ComponentId>>;
fn add_listener(&mut self, key: ListenerKey, sender_id: ComponentId) -> bool; fn add_listener(&mut self, key: ListenerKey, listener_id: ComponentId) -> bool;
fn remove_duplicates(&mut self, key: &ListenerKey); fn remove_duplicates(&mut self, key: &ListenerKey);
} }
pub trait SenderMapProvider< pub trait SenderMapProvider<
EventSender: EventSendProvider<Ev, Data>, EventSender: EventSendProvider<Event, ParamProvider>,
Ev: GenericEvent = EventU32, Event: GenericEvent = EventU32,
Data = Params, ParamProvider: Debug = Params,
> >
{ {
fn contains_send_event_provider(&self, target_id: &ComponentId) -> bool; fn contains_send_event_provider(&self, target_id: &ComponentId) -> bool;
@ -124,28 +166,27 @@ pub trait SenderMapProvider<
/// and [EventU16] are supported. /// and [EventU16] are supported.
/// * `Data`: Auxiliary data which is sent with the event to provide optional context information /// * `Data`: Auxiliary data which is sent with the event to provide optional context information
pub struct EventManager< pub struct EventManager<
EventReceiver: EventReceiveProvider<Ev, Data>, EventReceiver: EventReceiveProvider<Event, ParamProvider>,
SenderMap: SenderMapProvider<EventSender, Ev, Data>, SenderMap: SenderMapProvider<EventSender, Event, ParamProvider>,
ListenerMap: ListenerMapProvider, ListenerMap: ListenerMapProvider,
EventSender: EventSendProvider<Ev, Data>, EventSender: EventSendProvider<Event, ParamProvider>,
Ev: GenericEvent = EventU32, Event: GenericEvent = EventU32,
Data = Params, ParamProvider: Debug = Params,
> { > {
event_receiver: EventReceiver, event_receiver: EventReceiver,
sender_map: SenderMap, sender_map: SenderMap,
listener_map: ListenerMap, listener_map: ListenerMap,
phantom: core::marker::PhantomData<(EventSender, Ev, Data)>, phantom: core::marker::PhantomData<(EventSender, Event, ParamProvider)>,
} }
#[derive(Debug)] #[derive(Debug)]
pub enum EventRoutingResult<EV: GenericEvent, AUX> { pub enum EventRoutingResult<Event: GenericEvent, ParamProvider: Debug> {
/// No event was received /// No event was received
Empty, Empty,
/// An event was received and routed to listeners. /// An event was received and routed to listeners.
Handled { Handled {
num_recipients: u32, num_recipients: u32,
event: EV, event_msg: EventMessage<Event, ParamProvider>,
aux_data: Option<AUX>,
}, },
} }
@ -156,27 +197,21 @@ pub enum EventRoutingError {
NoSenderForId(ComponentId), NoSenderForId(ComponentId),
} }
#[derive(Debug)]
pub struct EventRoutingErrorsWithResult<Ev: GenericEvent, Data> {
pub result: EventRoutingResult<Ev, Data>,
pub errors: [Option<EventRoutingError>; 3],
}
impl< impl<
EventReceiver: EventReceiveProvider<Ev, Data>, EventReceiver: EventReceiveProvider<Event, ParamProvider>,
SenderMap: SenderMapProvider<EventSender, Ev, Data>, SenderMap: SenderMapProvider<EventSender, Event, ParamProvider>,
ListenerMap: ListenerMapProvider, ListenerMap: ListenerMapProvider,
EventSender: EventSendProvider<Ev, Data>, EventSender: EventSendProvider<Event, ParamProvider>,
Ev: GenericEvent + Copy, Event: GenericEvent + Copy,
Data: Clone, ParamProvider: Debug,
> EventManager<EventReceiver, SenderMap, ListenerMap, EventSender, Ev, Data> > EventManager<EventReceiver, SenderMap, ListenerMap, EventSender, Event, ParamProvider>
{ {
pub fn remove_duplicates(&mut self, key: &ListenerKey) { pub fn remove_duplicates(&mut self, key: &ListenerKey) {
self.listener_map.remove_duplicates(key) self.listener_map.remove_duplicates(key)
} }
/// Subscribe for a unique event. /// Subscribe for a unique event.
pub fn subscribe_single(&mut self, event: &Ev, sender_id: ComponentId) { pub fn subscribe_single(&mut self, event: &Event, sender_id: ComponentId) {
self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), sender_id); self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), sender_id);
} }
@ -193,17 +228,20 @@ impl<
self.update_listeners(ListenerKey::All, sender_id); self.update_listeners(ListenerKey::All, sender_id);
} }
} }
impl< impl<
ERP: EventReceiveProvider<EV, AUX>, EventReceiver: EventReceiveProvider<Event, ParamProvider>,
SMP: SenderMapProvider<SP, EV, AUX>, SenderMap: SenderMapProvider<EventSenderMap, Event, ParamProvider>,
LTR: ListenerMapProvider, ListenerMap: ListenerMapProvider,
SP: EventSendProvider<EV, AUX>, EventSenderMap: EventSendProvider<Event, ParamProvider>,
EV: GenericEvent + Copy, Event: GenericEvent + Copy,
AUX: Clone, ParamProvider: Debug,
> EventManager<ERP, SMP, LTR, SP, EV, AUX> > EventManager<EventReceiver, SenderMap, ListenerMap, EventSenderMap, Event, ParamProvider>
{ {
pub fn new_with_custom_maps(event_receiver: ERP, sender_map: SMP, listener_map: LTR) -> Self { pub fn new_with_custom_maps(
event_receiver: EventReceiver,
sender_map: SenderMap,
listener_map: ListenerMap,
) -> Self {
EventManager { EventManager {
listener_map, listener_map,
sender_map, sender_map,
@ -213,7 +251,7 @@ impl<
} }
/// Add a new sender component which can be used to send events to subscribers. /// Add a new sender component which can be used to send events to subscribers.
pub fn add_sender(&mut self, send_provider: SP) { pub fn add_sender(&mut self, send_provider: EventSenderMap) {
if !self if !self
.sender_map .sender_map
.contains_send_event_provider(&send_provider.target_id()) .contains_send_event_provider(&send_provider.target_id())
@ -226,7 +264,17 @@ impl<
fn update_listeners(&mut self, key: ListenerKey, sender_id: ComponentId) { fn update_listeners(&mut self, key: ListenerKey, sender_id: ComponentId) {
self.listener_map.add_listener(key, sender_id); 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. /// 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 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 /// If this works without any issues, the [EventRoutingResult] will contain context information
@ -234,60 +282,46 @@ impl<
/// ///
/// This function will track up to 3 errors returned as part of the /// This function will track up to 3 errors returned as part of the
/// [EventRoutingErrorsWithResult] error struct. /// [EventRoutingErrorsWithResult] error struct.
pub fn try_event_handling( pub fn try_event_handling<E: FnMut(EventRoutingError)>(
&self, &self,
) -> Result<EventRoutingResult<EV, AUX>, EventRoutingErrorsWithResult<EV, AUX>> { mut error_handler: E,
let mut err_idx = 0; ) -> EventRoutingResult<Event, ParamProvider> {
let mut err_slice = [None, None, None]; // let mut err_idx = 0;
// let mut err_slice = [None, None, None];
let mut num_recipients = 0; let mut num_recipients = 0;
let mut add_error = |error: EventRoutingError| { let mut send_handler = |key: &ListenerKey, event: Event, params: &Option<ParamProvider>| {
if err_idx < 3 {
err_slice[err_idx] = Some(error);
err_idx += 1;
}
};
let mut send_handler = |key: &ListenerKey, event: EV, aux_data: &Option<AUX>| {
if self.listener_map.contains_listener(key) { if self.listener_map.contains_listener(key) {
if let Some(ids) = self.listener_map.get_listener_ids(key) { if let Some(ids) = self.listener_map.get_listener_ids(key) {
for id in ids { for id in ids {
if let Some(sender) = self.sender_map.get_send_event_provider(id) { if let Some(sender) = self.sender_map.get_send_event_provider(id) {
if let Err(e) = sender.send(event, aux_data.clone()) { if let Err(e) =
add_error(EventRoutingError::Send(e)); sender.send(EventMessage::new_generic(*id, event, params.clone()))
{
error_handler(EventRoutingError::Send(e));
} else { } else {
num_recipients += 1; num_recipients += 1;
} }
} else { } else {
add_error(EventRoutingError::NoSenderForId(*id)); error_handler(EventRoutingError::NoSenderForId(*id));
} }
} }
} else { } else {
add_error(EventRoutingError::NoSendersForKey(*key)); error_handler(EventRoutingError::NoSendersForKey(*key));
} }
} }
}; };
if let Some((event, aux_data)) = self.event_receiver.try_recv_event() { if let Ok(Some(event_msg)) = self.event_receiver.try_recv_event() {
let single_key = ListenerKey::Single(event.raw_as_largest_type()); let single_key = ListenerKey::Single(event_msg.event.raw_as_largest_type());
send_handler(&single_key, event, &aux_data); send_handler(&single_key, event_msg.event, &event_msg.params);
let group_key = ListenerKey::Group(event.group_id_as_largest_type()); let group_key = ListenerKey::Group(event_msg.event.group_id_as_largest_type());
send_handler(&group_key, event, &aux_data); send_handler(&group_key, event_msg.event, &event_msg.params);
send_handler(&ListenerKey::All, event, &aux_data); send_handler(&ListenerKey::All, event_msg.event, &event_msg.params);
if err_idx > 0 { return EventRoutingResult::Handled {
return Err(EventRoutingErrorsWithResult {
result: EventRoutingResult::Handled {
num_recipients, num_recipients,
event, event_msg,
aux_data, };
},
errors: err_slice,
});
} }
return Ok(EventRoutingResult::Handled { EventRoutingResult::Empty
num_recipients,
event,
aux_data,
});
}
Ok(EventRoutingResult::Empty)
} }
} }
@ -311,23 +345,31 @@ pub mod alloc_mod {
/// and the [DefaultListenerMap]. It uses /// and the [DefaultListenerMap]. It uses
/// [bounded mpsc senders](https://doc.rust-lang.org/std/sync/mpsc/struct.SyncSender.html) as the /// [bounded mpsc senders](https://doc.rust-lang.org/std/sync/mpsc/struct.SyncSender.html) as the
/// message queue backend. /// message queue backend.
pub type EventManagerWithBoundedMpsc<EV = EventU32, AUX = Params> = EventManager< pub type EventManagerWithBoundedMpsc<Event = EventU32, ParamProvider = Params> = EventManager<
MpscEventReceiver, MpscEventReceiver,
DefaultSenderMap<EventSenderMpscBounded<EV>, EV, AUX>, DefaultSenderMap<EventSenderMpscBounded<Event>, Event, ParamProvider>,
DefaultListenerMap, DefaultListenerMap,
EventSenderMpscBounded<EV>, EventSenderMpscBounded<Event>,
>; >;
impl< impl<
ER: EventReceiveProvider<EV, AUX>, EventReceiver: EventReceiveProvider<Event, ParamProvider>,
SP: EventSendProvider<EV, AUX>, EventSender: EventSendProvider<Event, ParamProvider>,
EV: GenericEvent + Copy, Event: GenericEvent + Copy,
AUX: 'static, ParamProvider: 'static + Debug,
> EventManager<ER, DefaultSenderMap<SP, EV, AUX>, DefaultListenerMap, SP, EV, AUX> >
EventManager<
EventReceiver,
DefaultSenderMap<EventSender, Event, ParamProvider>,
DefaultListenerMap,
EventSender,
Event,
ParamProvider,
>
{ {
/// Create an event manager where the sender table will be the [DefaultSenderMap] /// Create an event manager where the sender table will be the [DefaultSenderMap]
/// and the listener table will be the [DefaultListenerMap]. /// and the listener table will be the [DefaultListenerMap].
pub fn new(event_receiver: ER) -> Self { pub fn new(event_receiver: EventReceiver) -> Self {
Self { Self {
listener_map: DefaultListenerMap::default(), listener_map: DefaultListenerMap::default(),
sender_map: DefaultSenderMap::default(), sender_map: DefaultSenderMap::default(),
@ -384,16 +426,19 @@ pub mod alloc_mod {
/// ///
/// Simple implementation which uses a [HashMap] internally. /// Simple implementation which uses a [HashMap] internally.
pub struct DefaultSenderMap< pub struct DefaultSenderMap<
SP: EventSendProvider<EV, AUX>, EventSender: EventSendProvider<Event, ParamProvider>,
EV: GenericEvent = EventU32, Event: GenericEvent = EventU32,
AUX = Params, ParamProvider: Debug = Params,
> { > {
senders: HashMap<ComponentId, SP>, senders: HashMap<ComponentId, EventSender>,
phantom: PhantomData<(EV, AUX)>, phantom: PhantomData<(Event, ParamProvider)>,
} }
impl<SP: EventSendProvider<EV, AUX>, EV: GenericEvent, AUX> Default impl<
for DefaultSenderMap<SP, EV, AUX> EventSender: EventSendProvider<Event, ParamProvider>,
Event: GenericEvent,
ParamProvider: Debug,
> Default for DefaultSenderMap<EventSender, Event, ParamProvider>
{ {
fn default() -> Self { fn default() -> Self {
Self { Self {
@ -403,20 +448,24 @@ pub mod alloc_mod {
} }
} }
impl<SP: EventSendProvider<EV, AUX>, EV: GenericEvent, AUX> SenderMapProvider<SP, EV, AUX> impl<
for DefaultSenderMap<SP, EV, AUX> 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 { fn contains_send_event_provider(&self, id: &ComponentId) -> bool {
self.senders.contains_key(id) self.senders.contains_key(id)
} }
fn get_send_event_provider(&self, id: &ComponentId) -> Option<&SP> { fn get_send_event_provider(&self, id: &ComponentId) -> Option<&EventSender> {
self.senders self.senders
.get(id) .get(id)
.filter(|sender| sender.target_id() == *id) .filter(|sender| sender.target_id() == *id)
} }
fn add_send_event_provider(&mut self, send_provider: SP) -> bool { fn add_send_event_provider(&mut self, send_provider: EventSender) -> bool {
let id = send_provider.target_id(); let id = send_provider.target_id();
if self.senders.contains_key(&id) { if self.senders.contains_key(&id) {
return false; return false;
@ -428,26 +477,33 @@ pub mod alloc_mod {
#[cfg(feature = "std")] #[cfg(feature = "std")]
pub mod std_mod { pub mod std_mod {
use crate::queue::GenericReceiveError;
use super::*; use super::*;
use std::sync::mpsc; use std::sync::mpsc;
pub struct MpscEventReceiver<Event: GenericEvent + Send = EventU32> { pub struct MpscEventReceiver<Event: GenericEvent + Send = EventU32> {
mpsc_receiver: mpsc::Receiver<(Event, Option<Params>)>, receiver: mpsc::Receiver<EventMessage<Event>>,
} }
impl<Event: GenericEvent + Send> MpscEventReceiver<Event> { impl<Event: GenericEvent + Send> MpscEventReceiver<Event> {
pub fn new(receiver: mpsc::Receiver<(Event, Option<Params>)>) -> Self { pub fn new(receiver: mpsc::Receiver<EventMessage<Event>>) -> Self {
Self { Self { receiver }
mpsc_receiver: receiver,
}
} }
} }
impl<Event: GenericEvent + Send> EventReceiveProvider<Event> for MpscEventReceiver<Event> { impl<Event: GenericEvent + Send> EventReceiveProvider<Event> for MpscEventReceiver<Event> {
fn try_recv_event(&self) -> Option<EventWithAuxData<Event>> { type Error = GenericReceiveError;
if let Ok(event_and_data) = self.mpsc_receiver.try_recv() {
return Some(event_and_data); fn try_recv_event(&self) -> Result<Option<EventMessage<Event>>, Self::Error> {
match self.receiver.try_recv() {
Ok(msg) => Ok(Some(msg)),
Err(e) => match e {
mpsc::TryRecvError::Empty => Ok(None),
mpsc::TryRecvError::Disconnected => {
Err(GenericReceiveError::TxDisconnected(None))
}
},
} }
None
} }
} }
@ -459,22 +515,25 @@ pub mod std_mod {
#[derive(Clone)] #[derive(Clone)]
pub struct EventSenderMpsc<Event: GenericEvent + Send> { pub struct EventSenderMpsc<Event: GenericEvent + Send> {
target_id: ComponentId, target_id: ComponentId,
sender: mpsc::Sender<(Event, Option<Params>)>, sender: mpsc::Sender<EventMessage<Event>>,
} }
impl<Event: GenericEvent + Send> EventSenderMpsc<Event> { impl<Event: GenericEvent + Send> EventSenderMpsc<Event> {
pub fn new(target_id: ComponentId, sender: mpsc::Sender<(Event, Option<Params>)>) -> Self { pub fn new(target_id: ComponentId, sender: mpsc::Sender<EventMessage<Event>>) -> Self {
Self { target_id, sender } Self { target_id, sender }
} }
} }
impl<Event: GenericEvent + Send> EventSendProvider<Event> for EventSenderMpsc<Event> { impl<Event: GenericEvent + Send> EventSendProvider<Event> for EventSenderMpsc<Event> {
type Error = GenericSendError;
fn target_id(&self) -> ComponentId { fn target_id(&self) -> ComponentId {
self.target_id self.target_id
} }
fn send(&self, event: Event, aux_data: Option<Params>) -> Result<(), GenericSendError> {
fn send(&self, event_msg: EventMessage<Event>) -> Result<(), GenericSendError> {
self.sender self.sender
.send((event, aux_data)) .send(event_msg)
.map_err(|_| GenericSendError::RxDisconnected) .map_err(|_| GenericSendError::RxDisconnected)
} }
} }
@ -484,14 +543,14 @@ pub mod std_mod {
#[derive(Clone)] #[derive(Clone)]
pub struct EventSenderMpscBounded<Event: GenericEvent + Send> { pub struct EventSenderMpscBounded<Event: GenericEvent + Send> {
target_id: ComponentId, target_id: ComponentId,
sender: mpsc::SyncSender<(Event, Option<Params>)>, sender: mpsc::SyncSender<EventMessage<Event>>,
capacity: usize, capacity: usize,
} }
impl<Event: GenericEvent + Send> EventSenderMpscBounded<Event> { impl<Event: GenericEvent + Send> EventSenderMpscBounded<Event> {
pub fn new( pub fn new(
target_id: ComponentId, target_id: ComponentId,
sender: mpsc::SyncSender<(Event, Option<Params>)>, sender: mpsc::SyncSender<EventMessage<Event>>,
capacity: usize, capacity: usize,
) -> Self { ) -> Self {
Self { Self {
@ -503,11 +562,14 @@ pub mod std_mod {
} }
impl<Event: GenericEvent + Send> EventSendProvider<Event> for EventSenderMpscBounded<Event> { impl<Event: GenericEvent + Send> EventSendProvider<Event> for EventSenderMpscBounded<Event> {
type Error = GenericSendError;
fn target_id(&self) -> ComponentId { fn target_id(&self) -> ComponentId {
self.target_id self.target_id
} }
fn send(&self, event: Event, aux_data: Option<Params>) -> Result<(), GenericSendError> {
if let Err(e) = self.sender.try_send((event, aux_data)) { fn send(&self, event_msg: EventMessage<Event>) -> Result<(), Self::Error> {
if let Err(e) = self.sender.try_send(event_msg) {
return match e { return match e {
mpsc::TrySendError::Full(_) => { mpsc::TrySendError::Full(_) => {
Err(GenericSendError::QueueFull(Some(self.capacity as u32))) Err(GenericSendError::QueueFull(Some(self.capacity as u32)))
@ -530,19 +592,20 @@ mod tests {
use super::*; use super::*;
use crate::event_man::EventManager; use crate::event_man::EventManager;
use crate::events::{EventU32, GenericEvent, Severity}; use crate::events::{EventU32, GenericEvent, Severity};
use crate::params::ParamsRaw; use crate::params::{ParamsHeapless, ParamsRaw};
use crate::pus::test_util::{TEST_COMPONENT_ID_0, TEST_COMPONENT_ID_1};
use std::format; use std::format;
use std::sync::mpsc::{self, channel, Receiver, Sender}; use std::sync::mpsc::{self};
const TEST_EVENT: EventU32 = EventU32::const_new(Severity::INFO, 0, 5); const TEST_EVENT: EventU32 = EventU32::const_new(Severity::INFO, 0, 5);
fn check_next_event( fn check_next_event(
expected: EventU32, expected: EventU32,
receiver: &Receiver<EventU32WithAuxData>, receiver: &mpsc::Receiver<EventMessageU32>,
) -> Option<Params> { ) -> Option<Params> {
if let Ok(event) = receiver.try_recv() { if let Ok(event_msg) = receiver.try_recv() {
assert_eq!(event.0, expected); assert_eq!(event_msg.event, expected);
return event.1; return event_msg.params;
} }
None None
} }
@ -555,17 +618,16 @@ mod tests {
assert!(matches!(res, EventRoutingResult::Handled { .. })); assert!(matches!(res, EventRoutingResult::Handled { .. }));
if let EventRoutingResult::Handled { if let EventRoutingResult::Handled {
num_recipients, num_recipients,
event, event_msg,
..
} = res } = res
{ {
assert_eq!(event, expected); assert_eq!(event_msg.event, expected);
assert_eq!(num_recipients, expected_num_sent); assert_eq!(num_recipients, expected_num_sent);
} }
} }
fn generic_event_man() -> (Sender<EventU32WithAuxData>, EventManagerWithMpsc) { fn generic_event_man() -> (mpsc::Sender<EventMessageU32>, EventManagerWithMpsc) {
let (event_sender, manager_queue) = channel(); let (event_sender, manager_queue) = mpsc::channel();
let event_man_receiver = MpscEventReceiver::new(manager_queue); let event_man_receiver = MpscEventReceiver::new(manager_queue);
(event_sender, EventManager::new(event_man_receiver)) (event_sender, EventManager::new(event_man_receiver))
} }
@ -575,48 +637,56 @@ mod tests {
let (event_sender, mut event_man) = generic_event_man(); let (event_sender, mut event_man) = generic_event_man();
let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap(); 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_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
let (single_event_sender, single_event_receiver) = channel(); let (single_event_sender, single_event_receiver) = mpsc::channel();
let single_event_listener = EventSenderMpsc::new(0, single_event_sender); let single_event_listener = EventSenderMpsc::new(0, single_event_sender);
event_man.subscribe_single(&event_grp_0, single_event_listener.target_id()); event_man.subscribe_single(&event_grp_0, single_event_listener.target_id());
event_man.add_sender(single_event_listener); event_man.add_sender(single_event_listener);
let (group_event_sender_0, group_event_receiver_0) = channel(); let (group_event_sender_0, group_event_receiver_0) = mpsc::channel();
let group_event_listener = EventU32SenderMpsc::new(1, group_event_sender_0); 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.subscribe_group(event_grp_1_0.group_id(), group_event_listener.target_id());
event_man.add_sender(group_event_listener); event_man.add_sender(group_event_listener);
let error_handler = |e: EventRoutingError| {
panic!("routing error occurred: {:?}", e);
};
// Test event with one listener // Test event with one listener
event_sender event_sender
.send((event_grp_0, None)) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_grp_0))
.expect("Sending single error failed"); .expect("Sending single error failed");
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(&error_handler);
assert!(res.is_ok()); // assert!(res.is_ok());
check_handled_event(res.unwrap(), event_grp_0, 1); check_handled_event(res, event_grp_0, 1);
check_next_event(event_grp_0, &single_event_receiver); check_next_event(event_grp_0, &single_event_receiver);
// Test event which is sent to all group listeners // Test event which is sent to all group listeners
event_sender event_sender
.send((event_grp_1_0, None)) .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_grp_1_0))
.expect("Sending group error failed"); .expect("Sending group error failed");
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(&error_handler);
assert!(res.is_ok()); check_handled_event(res, event_grp_1_0, 1);
check_handled_event(res.unwrap(), event_grp_1_0, 1);
check_next_event(event_grp_1_0, &group_event_receiver_0); check_next_event(event_grp_1_0, &group_event_receiver_0);
} }
#[test] #[test]
fn test_with_basic_aux_data() { fn test_with_basic_params() {
let error_handler = |e: EventRoutingError| {
panic!("routing error occurred: {:?}", e);
};
let (event_sender, mut event_man) = generic_event_man(); let (event_sender, mut event_man) = generic_event_man();
let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap(); let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap();
let (single_event_sender, single_event_receiver) = channel(); let (single_event_sender, single_event_receiver) = mpsc::channel();
let single_event_listener = EventSenderMpsc::new(0, single_event_sender); let single_event_listener = EventSenderMpsc::new(0, single_event_sender);
event_man.subscribe_single(&event_grp_0, single_event_listener.target_id()); event_man.subscribe_single(&event_grp_0, single_event_listener.target_id());
event_man.add_sender(single_event_listener); event_man.add_sender(single_event_listener);
event_sender event_sender
.send((event_grp_0, Some(Params::Heapless((2_u32, 3_u32).into())))) .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"); .expect("Sending group error failed");
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(&error_handler);
assert!(res.is_ok()); check_handled_event(res, event_grp_0, 1);
check_handled_event(res.unwrap(), event_grp_0, 1);
let aux = check_next_event(event_grp_0, &single_event_receiver); let aux = check_next_event(event_grp_0, &single_event_receiver);
assert!(aux.is_some()); assert!(aux.is_some());
let aux = aux.unwrap(); let aux = aux.unwrap();
@ -631,15 +701,16 @@ mod tests {
/// Test listening for multiple groups /// Test listening for multiple groups
#[test] #[test]
fn test_multi_group() { fn test_multi_group() {
let error_handler = |e: EventRoutingError| {
panic!("routing error occurred: {:?}", e);
};
let (event_sender, mut event_man) = generic_event_man(); let (event_sender, mut event_man) = generic_event_man();
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(&error_handler);
assert!(res.is_ok()); assert!(matches!(res, EventRoutingResult::Empty));
let hres = res.unwrap();
assert!(matches!(hres, EventRoutingResult::Empty));
let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap(); 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_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
let (event_grp_0_sender, event_grp_0_receiver) = channel(); 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); let event_grp_0_and_1_listener = EventU32SenderMpsc::new(0, event_grp_0_sender);
event_man.subscribe_group( event_man.subscribe_group(
event_grp_0.group_id(), event_grp_0.group_id(),
@ -652,17 +723,15 @@ mod tests {
event_man.add_sender(event_grp_0_and_1_listener); event_man.add_sender(event_grp_0_and_1_listener);
event_sender event_sender
.send((event_grp_0, None)) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_grp_0))
.expect("Sending Event Group 0 failed"); .expect("Sending Event Group 0 failed");
event_sender event_sender
.send((event_grp_1_0, None)) .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_grp_1_0))
.expect("Sendign Event Group 1 failed"); .expect("Sendign Event Group 1 failed");
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(error_handler);
assert!(res.is_ok()); check_handled_event(res, event_grp_0, 1);
check_handled_event(res.unwrap(), event_grp_0, 1); let res = event_man.try_event_handling(error_handler);
let res = event_man.try_event_handling(); check_handled_event(res, event_grp_1_0, 1);
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_0, &event_grp_0_receiver);
check_next_event(event_grp_1_0, &event_grp_0_receiver); check_next_event(event_grp_1_0, &event_grp_0_receiver);
@ -672,11 +741,14 @@ mod tests {
/// to both group and single events from one listener /// to both group and single events from one listener
#[test] #[test]
fn test_listening_to_same_event_and_multi_type() { fn test_listening_to_same_event_and_multi_type() {
let error_handler = |e: EventRoutingError| {
panic!("routing error occurred: {:?}", e);
};
let (event_sender, mut event_man) = generic_event_man(); let (event_sender, mut event_man) = generic_event_man();
let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap(); let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap();
let event_1 = EventU32::new(Severity::HIGH, 1, 0).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_0, event_0_rx_0) = mpsc::channel();
let (event_0_tx_1, event_0_rx_1) = 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_0 = EventU32SenderMpsc::new(0, event_0_tx_0);
let event_listener_1 = EventU32SenderMpsc::new(1, event_0_tx_1); let event_listener_1 = EventU32SenderMpsc::new(1, event_0_tx_1);
let event_listener_0_sender_id = event_listener_0.target_id(); let event_listener_0_sender_id = event_listener_0.target_id();
@ -686,28 +758,25 @@ mod tests {
event_man.subscribe_single(&event_0, event_listener_1_sender_id); event_man.subscribe_single(&event_0, event_listener_1_sender_id);
event_man.add_sender(event_listener_1); event_man.add_sender(event_listener_1);
event_sender event_sender
.send((event_0, None)) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0))
.expect("Triggering Event 0 failed"); .expect("Triggering Event 0 failed");
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(error_handler);
assert!(res.is_ok()); check_handled_event(res, event_0, 2);
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_0);
check_next_event(event_0, &event_0_rx_1); check_next_event(event_0, &event_0_rx_1);
event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id); event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id);
event_sender event_sender
.send((event_0, None)) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0))
.expect("Triggering Event 0 failed"); .expect("Triggering Event 0 failed");
event_sender event_sender
.send((event_1, None)) .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_1))
.expect("Triggering Event 1 failed"); .expect("Triggering Event 1 failed");
// 3 Events messages will be sent now // 3 Events messages will be sent now
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(error_handler);
assert!(res.is_ok()); check_handled_event(res, event_0, 2);
check_handled_event(res.unwrap(), event_0, 2); let res = event_man.try_event_handling(error_handler);
let res = event_man.try_event_handling(); check_handled_event(res, event_1, 1);
assert!(res.is_ok());
check_handled_event(res.unwrap(), event_1, 1);
// Both the single event and the group event should arrive now // Both the single event and the group event should arrive now
check_next_event(event_0, &event_0_rx_0); check_next_event(event_0, &event_0_rx_0);
check_next_event(event_1, &event_0_rx_0); check_next_event(event_1, &event_0_rx_0);
@ -716,36 +785,36 @@ mod tests {
event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id); event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id);
event_man.remove_duplicates(&ListenerKey::Group(event_1.group_id())); event_man.remove_duplicates(&ListenerKey::Group(event_1.group_id()));
event_sender event_sender
.send((event_1, None)) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_1))
.expect("Triggering Event 1 failed"); .expect("Triggering Event 1 failed");
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(error_handler);
assert!(res.is_ok()); check_handled_event(res, event_1, 1);
check_handled_event(res.unwrap(), event_1, 1);
} }
#[test] #[test]
fn test_all_events_listener() { fn test_all_events_listener() {
let (event_sender, manager_queue) = channel(); let error_handler = |e: EventRoutingError| {
panic!("routing error occurred: {:?}", e);
};
let (event_sender, manager_queue) = mpsc::channel();
let event_man_receiver = MpscEventReceiver::new(manager_queue); let event_man_receiver = MpscEventReceiver::new(manager_queue);
let mut event_man = EventManagerWithMpsc::new(event_man_receiver); let mut event_man = EventManagerWithMpsc::new(event_man_receiver);
let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap(); let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap();
let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap(); let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
let (event_0_tx_0, all_events_rx) = channel(); let (event_0_tx_0, all_events_rx) = mpsc::channel();
let all_events_listener = EventU32SenderMpsc::new(0, event_0_tx_0); let all_events_listener = EventU32SenderMpsc::new(0, event_0_tx_0);
event_man.subscribe_all(all_events_listener.target_id()); event_man.subscribe_all(all_events_listener.target_id());
event_man.add_sender(all_events_listener); event_man.add_sender(all_events_listener);
event_sender event_sender
.send((event_0, None)) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0))
.expect("Triggering event 0 failed"); .expect("Triggering event 0 failed");
event_sender event_sender
.send((event_1, None)) .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_1))
.expect("Triggering event 1 failed"); .expect("Triggering event 1 failed");
let res = event_man.try_event_handling(); let res = event_man.try_event_handling(error_handler);
assert!(res.is_ok()); check_handled_event(res, event_0, 1);
check_handled_event(res.unwrap(), event_0, 1); let res = event_man.try_event_handling(error_handler);
let res = event_man.try_event_handling(); check_handled_event(res, event_1, 1);
assert!(res.is_ok());
check_handled_event(res.unwrap(), event_1, 1);
check_next_event(event_0, &all_events_rx); check_next_event(event_0, &all_events_rx);
check_next_event(event_1, &all_events_rx); check_next_event(event_1, &all_events_rx);
} }
@ -755,15 +824,15 @@ mod tests {
let (event_sender, _event_receiver) = mpsc::sync_channel(3); let (event_sender, _event_receiver) = mpsc::sync_channel(3);
let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3); let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3);
event_sender event_sender
.send_no_data(TEST_EVENT) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT))
.expect("sending test event failed"); .expect("sending test event failed");
event_sender event_sender
.send_no_data(TEST_EVENT) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT))
.expect("sending test event failed"); .expect("sending test event failed");
event_sender event_sender
.send_no_data(TEST_EVENT) .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT))
.expect("sending test event failed"); .expect("sending test event failed");
let error = event_sender.send_no_data(TEST_EVENT); let error = event_sender.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT));
if let Err(e) = error { if let Err(e) = error {
assert!(matches!(e, GenericSendError::QueueFull(Some(3)))); assert!(matches!(e, GenericSendError::QueueFull(Some(3))));
} else { } else {
@ -775,7 +844,7 @@ mod tests {
let (event_sender, event_receiver) = mpsc::sync_channel(3); let (event_sender, event_receiver) = mpsc::sync_channel(3);
let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3); let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3);
drop(event_receiver); drop(event_receiver);
if let Err(e) = event_sender.send_no_data(TEST_EVENT) { if let Err(e) = event_sender.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT)) {
assert!(matches!(e, GenericSendError::RxDisconnected)); assert!(matches!(e, GenericSendError::RxDisconnected));
} else { } else {
panic!("Expected error"); panic!("Expected error");

2
satrs/src/events.rs
View File

@ -80,7 +80,7 @@ impl HasSeverity for SeverityHigh {
const SEVERITY: Severity = Severity::HIGH; const SEVERITY: Severity = Severity::HIGH;
} }
pub trait GenericEvent: EcssEnumeration { pub trait GenericEvent: EcssEnumeration + Copy + Clone {
type Raw; type Raw;
type GroupId; type GroupId;
type UniqueId; type UniqueId;

20
satrs/src/pus/event_man.rs
View File

@ -157,8 +157,8 @@ pub mod alloc_mod {
phantom: PhantomData<(E, EV)>, phantom: PhantomData<(E, EV)>,
} }
impl<B: PusEventMgmtBackendProvider<EV, Error = E>, EV: GenericEvent, E> impl<B: PusEventMgmtBackendProvider<Event, Error = E>, Event: GenericEvent, E>
PusEventDispatcher<B, EV, E> PusEventDispatcher<B, Event, E>
{ {
pub fn new(reporter: EventReporter, backend: B) -> Self { pub fn new(reporter: EventReporter, backend: B) -> Self {
Self { Self {
@ -168,11 +168,11 @@ pub mod alloc_mod {
} }
} }
pub fn enable_tm_for_event(&mut self, event: &EV) -> Result<bool, E> { pub fn enable_tm_for_event(&mut self, event: &Event) -> Result<bool, E> {
self.backend.enable_event_reporting(event) self.backend.enable_event_reporting(event)
} }
pub fn disable_tm_for_event(&mut self, event: &EV) -> Result<bool, E> { pub fn disable_tm_for_event(&mut self, event: &Event) -> Result<bool, E> {
self.backend.disable_event_reporting(event) self.backend.disable_event_reporting(event)
} }
@ -180,8 +180,8 @@ pub mod alloc_mod {
&self, &self,
sender: &(impl EcssTmSenderCore + ?Sized), sender: &(impl EcssTmSenderCore + ?Sized),
time_stamp: &[u8], time_stamp: &[u8],
event: EV, event: Event,
aux_data: Option<&[u8]>, params: Option<&[u8]>,
) -> Result<bool, EventManError> { ) -> Result<bool, EventManError> {
if !self.backend.event_enabled(&event) { if !self.backend.event_enabled(&event) {
return Ok(false); return Ok(false);
@ -189,22 +189,22 @@ pub mod alloc_mod {
match event.severity() { match event.severity() {
Severity::INFO => self Severity::INFO => self
.reporter .reporter
.event_info(sender, time_stamp, event, aux_data) .event_info(sender, time_stamp, event, params)
.map(|_| true) .map(|_| true)
.map_err(|e| e.into()), .map_err(|e| e.into()),
Severity::LOW => self Severity::LOW => self
.reporter .reporter
.event_low_severity(sender, time_stamp, event, aux_data) .event_low_severity(sender, time_stamp, event, params)
.map(|_| true) .map(|_| true)
.map_err(|e| e.into()), .map_err(|e| e.into()),
Severity::MEDIUM => self Severity::MEDIUM => self
.reporter .reporter
.event_medium_severity(sender, time_stamp, event, aux_data) .event_medium_severity(sender, time_stamp, event, params)
.map(|_| true) .map(|_| true)
.map_err(|e| e.into()), .map_err(|e| e.into()),
Severity::HIGH => self Severity::HIGH => self
.reporter .reporter
.event_high_severity(sender, time_stamp, event, aux_data) .event_high_severity(sender, time_stamp, event, params)
.map(|_| true) .map(|_| true)
.map_err(|e| e.into()), .map_err(|e| e.into()),
} }

27
satrs/tests/pus_events.rs
View File

@ -1,10 +1,12 @@
use satrs::event_man::{ use satrs::event_man::{
EventManagerWithMpsc, EventSendProvider, EventU32SenderMpsc, MpscEventU32Receiver, EventManagerWithMpsc, EventMessage, EventRoutingError, EventSendProvider, EventU32SenderMpsc,
MpscEventU32Receiver,
}; };
use satrs::events::{EventU32, EventU32TypedSev, Severity, SeverityInfo}; use satrs::events::{EventU32, EventU32TypedSev, Severity, SeverityInfo};
use satrs::params::U32Pair; use satrs::params::U32Pair;
use satrs::params::{Params, ParamsHeapless, WritableToBeBytes}; use satrs::params::{Params, ParamsHeapless, WritableToBeBytes};
use satrs::pus::event_man::{DefaultPusEventMgmtBackend, EventReporter, PusEventDispatcher}; use satrs::pus::event_man::{DefaultPusEventMgmtBackend, EventReporter, PusEventDispatcher};
use satrs::pus::test_util::TEST_COMPONENT_ID_0;
use satrs::pus::PusTmAsVec; use satrs::pus::PusTmAsVec;
use satrs::request::UniqueApidTargetId; use satrs::request::UniqueApidTargetId;
use spacepackets::ecss::tm::PusTmReader; use spacepackets::ecss::tm::PusTmReader;
@ -39,24 +41,26 @@ fn test_threaded_usage() {
let reporter = let reporter =
EventReporter::new(TEST_ID.raw(), 0x02, 128).expect("Creating event reporter failed"); EventReporter::new(TEST_ID.raw(), 0x02, 128).expect("Creating event reporter failed");
let pus_event_man = PusEventDispatcher::new(reporter, DefaultPusEventMgmtBackend::default()); let pus_event_man = PusEventDispatcher::new(reporter, DefaultPusEventMgmtBackend::default());
let error_handler = |error: EventRoutingError| {
panic!("received routing error {error:?}");
};
// PUS + Generic event manager thread // PUS + Generic event manager thread
let jh0 = thread::spawn(move || { let jh0 = thread::spawn(move || {
let mut event_cnt = 0; let mut event_cnt = 0;
let mut params_array: [u8; 128] = [0; 128]; let mut params_array: [u8; 128] = [0; 128];
loop { loop {
let res = event_man.try_event_handling(); event_man.try_event_handling(error_handler);
assert!(res.is_ok());
match pus_event_man_rx.try_recv() { match pus_event_man_rx.try_recv() {
Ok((event, aux_data)) => { Ok(event_msg) => {
let gen_event = |aux_data| { let gen_event = |aux_data| {
pus_event_man.generate_pus_event_tm_generic( pus_event_man.generate_pus_event_tm_generic(
&event_tx, &event_tx,
&EMPTY_STAMP, &EMPTY_STAMP,
event, event_msg.event(),
aux_data, aux_data,
) )
}; };
let res = if let Some(aux_data) = aux_data { let res = if let Some(aux_data) = event_msg.params() {
match aux_data { match aux_data {
Params::Heapless(heapless) => match heapless { Params::Heapless(heapless) => match heapless {
ParamsHeapless::Raw(raw) => { ParamsHeapless::Raw(raw) => {
@ -97,7 +101,10 @@ fn test_threaded_usage() {
// Event sender and TM checker thread // Event sender and TM checker thread
let jh1 = thread::spawn(move || { let jh1 = thread::spawn(move || {
event_sender event_sender
.send((INFO_EVENT.into(), None)) .send(EventMessage::new(
TEST_COMPONENT_ID_0.id(),
INFO_EVENT.into(),
))
.expect("Sending info event failed"); .expect("Sending info event failed");
loop { loop {
match event_rx.try_recv() { match event_rx.try_recv() {
@ -123,7 +130,11 @@ fn test_threaded_usage() {
} }
} }
event_sender event_sender
.send((LOW_SEV_EVENT, Some(Params::Heapless((2_u32, 3_u32).into())))) .send(EventMessage::new_with_params(
TEST_COMPONENT_ID_0.id(),
LOW_SEV_EVENT,
Params::Heapless((2_u32, 3_u32).into()),
))
.expect("Sending low severity event failed"); .expect("Sending low severity event failed");
loop { loop {
match event_rx.try_recv() { match event_rx.try_recv() {