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

allow flexible event manager table backends

Browse Source
This commit is contained in:
Robin Müller
2022-11-19 02:42:11 +01:00
parent 3f84a06fd5
commit ce8babc8a1
5 changed files with 278 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

426
fsrc-core/src/event_man.rs
View File

@ -52,7 +52,7 @@ use hashbrown::HashMap;
#[cfg(feature = "std")]
pub use stdmod::*;
#[derive(PartialEq, Eq, Hash, Copy, Clone)]
#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)]
pub enum ListenerKey {
Single(LargestEventRaw),
Group(LargestGroupIdRaw),
@ -67,7 +67,7 @@ pub type EventWithAuxData<Event> = (Event, Option<Params>);
pub type EventU32WithAuxData = EventWithAuxData<EventU32>;
pub type EventU16WithAuxData = EventWithAuxData<EventU16>;
type SenderId = u32;
pub type SenderId = u32;
pub trait SendEventProvider<Provider: GenericEvent, AuxDataProvider = Params> {
type Error;
@ -83,11 +83,6 @@ pub trait SendEventProvider<Provider: GenericEvent, AuxDataProvider = Params> {
) -> Result<(), Self::Error>;
}
pub struct Listener<E, Event: GenericEvent, AuxDataProvider = Params> {
ltype: ListenerKey,
send_provider: Box<dyn SendEventProvider<Event, AuxDataProvider, Error = E>>,
}
/// 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
@ -99,14 +94,19 @@ pub trait EventReceiver<Event: GenericEvent, AuxDataProvider = Params> {
fn receive(&mut self) -> Option<(Event, Option<AuxDataProvider>)>;
}
pub trait ListenerTable<SendProviderError, Event: GenericEvent = EventU32, AuxDataProvider = Params>
{
fn get_listener_ids(&mut self, key: ListenerKey) -> Option<Iter<SenderId>>;
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<SenderId>>;
fn add_listener(&mut self, key: ListenerKey, sender_id: SenderId) -> bool;
fn remove_duplicates(&mut self, key: &ListenerKey);
}
pub trait SenderTable<SendProviderError, Event: GenericEvent = EventU32, AuxDataProvider = Params> {
fn contains_send_event_provider(&self, id: &SenderId) -> bool;
fn get_send_event_provider(
&mut self,
id: SenderId,
id: &SenderId,
) -> Option<&mut Box<dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>>>;
fn add_send_event_provider(
&mut self,
@ -122,36 +122,218 @@ pub trait ListenerTable<SendProviderError, Event: GenericEvent = EventU32, AuxDa
///
/// * `SendProviderError`: [SendEventProvider] error type
/// * `Event`: Concrete event provider, currently either [EventU32] or [EventU16]
/// * `AuxDataProvider`: Concrete auxiliary data provder, currently either [Params] or
/// * `AuxDataProvider`: Concrete auxiliary data provider, currently either [Params] or
/// [ParamsHeapless]
pub struct EventManager<SendProviderError, Event: GenericEvent = EventU32, AuxDataProvider = Params>
{
listeners: HashMap<ListenerKey, Vec<Listener<SendProviderError, Event, AuxDataProvider>>>,
//listener_table: Box<dyn ListenerTable<SendProviderError, Event, AuxDataProvider>>,
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(SenderId),
}
#[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)
}
pub fn subscribe_single(&mut self, event: &Event, sender_id: SenderId) {
self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), sender_id);
}
pub fn subscribe_group(&mut self, group_id: LargestGroupIdRaw, sender_id: SenderId) {
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: SenderId) {
self.update_listeners(ListenerKey::All, sender_id);
}
}
impl<E: 'static, Event: GenericEvent + Copy + 'static, AuxDataProvider: Clone + 'static>
EventManager<E, Event, AuxDataProvider>
{
pub fn new(event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>) -> Self {
let listener_table = Box::new(DefaultListenerTableProvider::default());
let sender_table =
Box::new(DefaultSenderTableProvider::<E, Event, AuxDataProvider>::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: SenderId) {
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(
&mut 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<SenderId>>,
}
pub struct DefaultSenderTableProvider<
SendProviderError,
Event: GenericEvent = EventU32,
AuxDataProvider = Params,
> {
listeners: HashMap<ListenerKey, Vec<SenderId>>,
senders: HashMap<
SenderId,
Box<dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>>,
>,
}
#[cfg(feature = "std")]
impl<SendProviderError, Event: GenericEvent, AuxDataProvider>
ListenerTable<SendProviderError, Event, AuxDataProvider>
impl<SendProviderError, Event: GenericEvent, AuxDataProvider> Default
for DefaultSenderTableProvider<SendProviderError, Event, AuxDataProvider>
{
fn get_listener_ids(&mut self, key: ListenerKey) -> Option<Iter<SenderId>> {
self.listeners.get(&key).map(|vec| vec.into_iter())
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<SenderId>> {
self.listeners.get(key).map(|vec| vec.iter())
}
fn add_listener(&mut self, key: ListenerKey, sender_id: SenderId) -> bool {
@ -164,12 +346,30 @@ impl<SendProviderError, Event: GenericEvent, AuxDataProvider>
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: &SenderId) -> bool {
self.senders.contains_key(id)
}
fn get_send_event_provider(
&mut self,
id: SenderId,
id: &SenderId,
) -> Option<&mut Box<dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>>>
{
self.senders.get_mut(&id).filter(|sender| sender.id() == id)
self.senders
.get_mut(id)
.filter(|sender| sender.id() == *id)
}
fn add_send_event_provider(
@ -186,145 +386,6 @@ impl<SendProviderError, Event: GenericEvent, 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>
{
}
pub enum HandlerResult<Event: GenericEvent, AuxDataProvider> {
Empty,
Handled(u32, Event, Option<AuxDataProvider>),
}
impl<E, Event: GenericEvent + Copy> EventManager<E, Event> {
pub fn new(event_receiver: Box<dyn EventReceiver<Event>>) -> Self {
EventManager {
listeners: HashMap::new(),
event_receiver,
}
}
pub fn subscribe_single(
&mut self,
event: Event,
dest: impl SendEventProvider<Event, Error = E> + 'static,
) {
self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), dest);
}
pub fn subscribe_group(
&mut self,
group_id: LargestGroupIdRaw,
dest: impl SendEventProvider<Event, Error = E> + 'static,
) {
self.update_listeners(ListenerKey::Group(group_id), dest);
}
/// 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,
send_provider: impl SendEventProvider<Event, Error = E> + 'static,
) {
self.update_listeners(ListenerKey::All, send_provider);
}
/// Helper function which removes single subscriptions for which a group subscription already
/// exists.
pub fn remove_single_subscriptions_for_group(
&mut self,
group_id: LargestGroupIdRaw,
dest: impl SendEventProvider<Event, Error = E> + 'static,
) {
if self.listeners.contains_key(&ListenerKey::Group(group_id)) {
for (ltype, listeners) in &mut self.listeners {
if let ListenerKey::Single(_) = ltype {
listeners.retain(|f| f.send_provider.id() != dest.id());
}
}
}
}
}
impl<E, Event: GenericEvent + Copy, AuxDataProvider: Clone>
EventManager<E, Event, AuxDataProvider>
{
fn update_listeners(
&mut self,
key: ListenerKey,
dest: impl SendEventProvider<Event, AuxDataProvider, Error = E> + 'static,
) {
if !self.listeners.contains_key(&key) {
self.listeners.insert(
key,
vec![Listener {
ltype: key,
send_provider: Box::new(dest),
}],
);
} else {
let vec = self.listeners.get_mut(&key).unwrap();
// To prevent double insertions
for entry in vec.iter() {
if entry.ltype == key && entry.send_provider.id() == dest.id() {
return;
}
}
vec.push(Listener {
ltype: key,
send_provider: Box::new(dest),
});
}
}
pub fn try_event_handling(&mut self) -> Result<HandlerResult<Event, AuxDataProvider>, E> {
let mut err_status = None;
let mut num_recipients = 0;
let mut send_handler =
|event: Event,
aux_data: Option<AuxDataProvider>,
llist: &mut Vec<Listener<E, Event, AuxDataProvider>>| {
for listener in llist.iter_mut() {
if let Err(e) = listener.send_provider.send(event, aux_data.clone()) {
err_status = Some(Err(e));
} else {
num_recipients += 1;
}
}
};
if let Some((event, aux_data)) = self.event_receiver.receive() {
let single_key = ListenerKey::Single(event.raw_as_largest_type());
if self.listeners.contains_key(&single_key) {
send_handler(
event,
aux_data.clone(),
self.listeners.get_mut(&single_key).unwrap(),
);
}
let group_key = ListenerKey::Group(event.group_id_as_largest_type());
if self.listeners.contains_key(&group_key) {
send_handler(
event,
aux_data.clone(),
self.listeners.get_mut(&group_key).unwrap(),
);
}
if let Some(all_receivers) = self.listeners.get_mut(&ListenerKey::All) {
send_handler(event, aux_data.clone(), all_receivers);
}
if let Some(err) = err_status {
return err;
}
return Ok(HandlerResult::Handled(num_recipients, event, aux_data));
}
Ok(HandlerResult::Empty)
}
}
#[cfg(feature = "std")]
pub mod stdmod {
use super::*;
@ -362,9 +423,6 @@ pub mod stdmod {
sender: Sender<(Event, Option<Params>)>,
}
/// Safety: Send is safe to implement because both the ID and the MPSC sender are Send
//unsafe impl<Event: GenericEvent> Send for MpscEventSendProvider<Event> {}
impl<Event: GenericEvent + Send> MpscEventSendProvider<Event> {
pub fn new(id: u32, sender: Sender<(Event, Option<Params>)>) -> Self {
Self { id, sender }
@ -431,12 +489,12 @@ mod tests {
}
fn check_handled_event(
res: HandlerResult<EventU32, Params>,
res: EventRoutingResult<EventU32, Params>,
expected: EventU32,
expected_num_sent: u32,
) {
assert!(matches!(res, HandlerResult::Handled { .. }));
if let HandlerResult::Handled(num_recipients, event, _aux_data) = res {
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);
}
@ -461,13 +519,15 @@ mod tests {
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);
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);
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
@ -494,7 +554,8 @@ mod tests {
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);
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");
@ -519,7 +580,7 @@ mod tests {
let res = event_man.try_event_handling();
assert!(res.is_ok());
let hres = res.unwrap();
assert!(matches!(hres, HandlerResult::Empty));
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();
@ -528,8 +589,9 @@ mod tests {
id: 0,
mpsc_sender: event_grp_0_sender,
};
event_man.subscribe_group(event_grp_0.group_id(), event_grp_0_and_1_listener.clone());
event_man.subscribe_group(event_grp_1_0.group_id(), event_grp_0_and_1_listener);
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))
@ -565,8 +627,12 @@ mod tests {
id: 1,
mpsc_sender: event_0_tx_1,
};
event_man.subscribe_single(event_0, event_listener_0.clone());
event_man.subscribe_single(event_0, event_listener_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");
@ -575,7 +641,7 @@ mod tests {
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.clone());
event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id);
event_sender
.send((event_0, None))
.expect("Triggering Event 0 failed");
@ -594,8 +660,9 @@ mod tests {
check_next_event(event_0, &event_0_rx_0);
check_next_event(event_1, &event_0_rx_0);
// Double insertion should be detected, result should remain the same
event_man.subscribe_group(event_1.group_id(), event_listener_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");
@ -617,7 +684,8 @@ mod tests {
id: 0,
mpsc_sender: event_0_tx_0,
};
event_man.subscribe_all(all_events_listener);
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");

11
fsrc-core/tests/pus_events.rs
View File

@ -1,4 +1,6 @@
use fsrc_core::event_man::{EventManager, MpscEventReceiver, MpscEventU32SendProvider};
use fsrc_core::event_man::{
EventManagerWithMpscQueue, MpscEventU32Receiver, MpscEventU32SendProvider, SendEventProvider,
};
use fsrc_core::events::{EventU32, EventU32TypedSev, Severity, SeverityInfo};
use fsrc_core::params::U32Pair;
use fsrc_core::params::{Params, ParamsHeapless, WritableToBeBytes};
@ -31,12 +33,13 @@ impl EcssTmSender for EventTmSender {
#[test]
fn test_threaded_usage() {
let (event_sender, event_man_receiver) = channel();
let event_receiver = MpscEventReceiver::new(event_man_receiver);
let mut event_man = EventManager::new(Box::new(event_receiver));
let event_receiver = MpscEventU32Receiver::new(event_man_receiver);
let mut event_man = EventManagerWithMpscQueue::new(Box::new(event_receiver));
let (pus_event_man_tx, pus_event_man_rx) = channel();
let pus_event_man_send_provider = MpscEventU32SendProvider::new(1, pus_event_man_tx);
event_man.subscribe_all(pus_event_man_send_provider);
event_man.subscribe_all(pus_event_man_send_provider.id());
event_man.add_sender(pus_event_man_send_provider);
let (event_tx, event_rx) = channel();
let reporter = EventReporter::new(0x02, 128).expect("Creating event reporter failed");
let backend = DefaultPusMgmtBackendProvider::<EventU32>::default();