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

first tests with auxiliary data

Browse Source
This commit is contained in:
Robin Müller 2022-10-29 13:05:29 +02:00
parent 190b6d035a
commit af288a1b2b
No known key found for this signature in database
GPG Key ID: 71B58F8A3CDFA9AC
3 changed files with 225 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -27,7 +27,8 @@
//! 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.
use crate::events::{GenericEvent, LargestEventRaw, LargestGroupIdRaw};
use crate::events::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw};
use crate::util::{AuxData, AuxDataHeapless};
use alloc::boxed::Box;
use alloc::vec;
use alloc::vec::Vec;
@ -43,48 +44,69 @@ enum ListenerType {
All,
}
pub trait SendEventProvider<Provider: GenericEvent> {
pub type EventWithHeaplessAuxData<Event> = (Event, Option<AuxDataHeapless>);
pub type EventU32WithHeaplessAuxData = EventWithHeaplessAuxData<EventU32>;
pub type EventU16WithHeaplessAuxData = EventWithHeaplessAuxData<EventU16>;
pub type EventWithAuxData<Event> = (Event, Option<AuxData>);
pub type EventU32WithAuxData = EventWithAuxData<EventU32>;
pub type EventU16WithAuxData = EventWithAuxData<EventU16>;
pub trait SendEventProvider<Provider: GenericEvent, AuxDataProvider = AuxData> {
type Error;
fn id(&self) -> u32;
fn send_no_data(&mut self, event: Provider) -> Result<(), Self::Error> {
self.send(event, None)
}
fn send(&mut self, event: Provider, aux_data: Option<&[u8]>) -> Result<(), Self::Error>;
fn send(
&mut self,
event: Provider,
aux_data: Option<AuxDataProvider>,
) -> Result<(), Self::Error>;
}
struct Listener<E, Event: GenericEvent> {
struct Listener<E, Event: GenericEvent, AuxDataProvider = AuxData> {
ltype: ListenerType,
send_provider: Box<dyn SendEventProvider<Event, Error = E>>,
send_provider: Box<dyn SendEventProvider<Event, AuxDataProvider, Error = E>>,
}
/// Generic abstraction for an event receiver.
pub trait EventReceiver<Event: GenericEvent> {
pub trait EventReceiver<Event: GenericEvent, AuxDataProvider = AuxData> {
/// 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(&mut self, aux_data: &mut [u8]) -> Option<Event>;
fn receive(&mut self) -> Option<(Event, Option<AuxDataProvider>)>;
}
/// Generic event manager implementation.
pub struct EventManager<E, Event: GenericEvent> {
aux_data_buf: Vec<u8>,
listeners: HashMap<ListenerType, Vec<Listener<E, Event>>>,
event_receiver: Box<dyn EventReceiver<Event>>,
///
/// # Generics
///
/// * `SendProviderError`: [SendEventProvider] error type
/// * `Event`: Concrete event provider, currently either [EventU32] or [EventU16]
/// * `AuxDataProvider`: Concrete auxiliary data provder, currently either [AuxData] or
/// [AuxDataHeapless]
pub struct EventManager<
SendProviderError,
Event: GenericEvent = EventU32,
AuxDataProvider = AuxData,
> {
listeners: HashMap<ListenerType, Vec<Listener<SendProviderError, Event, AuxDataProvider>>>,
event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>,
}
pub enum HandlerResult<Provider: GenericEvent> {
pub enum HandlerResult<Provider: GenericEvent, AuxDataProvider> {
Empty,
Handled(u32, Provider),
Handled(u32, Provider, Option<AuxDataProvider>),
}
impl<E, Event: GenericEvent + Copy> EventManager<E, Event> {
pub fn new(event_receiver: Box<dyn EventReceiver<Event>>, buf_len_aux_data: usize) -> Self {
pub fn new(event_receiver: Box<dyn EventReceiver<Event>>) -> Self {
EventManager {
aux_data_buf: vec![0; buf_len_aux_data],
listeners: HashMap::new(),
event_receiver,
}
@ -105,6 +127,10 @@ impl<E, Event: GenericEvent + Copy> EventManager<E, Event> {
self.update_listeners(ListenerType::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, dest: impl SendEventProvider<Event, Error = E> + 'static) {
self.update_listeners(ListenerType::All, dest);
}
@ -126,11 +152,13 @@ impl<E, Event: GenericEvent + Copy> EventManager<E, Event> {
}
}
impl<E, Provider: GenericEvent + Copy> EventManager<E, Provider> {
impl<E, Event: GenericEvent + Copy, AuxDataProvider: Clone>
EventManager<E, Event, AuxDataProvider>
{
fn update_listeners(
&mut self,
key: ListenerType,
dest: impl SendEventProvider<Provider, Error = E> + 'static,
dest: impl SendEventProvider<Event, AuxDataProvider, Error = E> + 'static,
) {
if !self.listeners.contains_key(&key) {
self.listeners.insert(
@ -155,28 +183,27 @@ impl<E, Provider: GenericEvent + Copy> EventManager<E, Provider> {
}
}
pub fn try_event_handling(&mut self) -> Result<HandlerResult<Provider>, E> {
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: Provider, aux_data: Option<&[u8]>, llist: &mut Vec<Listener<E, Provider>>| {
|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) {
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) = self
.event_receiver
.receive(self.aux_data_buf.as_mut_slice())
{
if let Some((event, aux_data)) = self.event_receiver.receive() {
let single_key = ListenerType::Single(event.raw_as_largest_type());
if self.listeners.contains_key(&single_key) {
send_handler(
event,
Some(self.aux_data_buf.as_slice()),
aux_data.clone(),
self.listeners.get_mut(&single_key).unwrap(),
);
}
@ -184,17 +211,17 @@ impl<E, Provider: GenericEvent + Copy> EventManager<E, Provider> {
if self.listeners.contains_key(&group_key) {
send_handler(
event,
Some(self.aux_data_buf.as_slice()),
aux_data.clone(),
self.listeners.get_mut(&group_key).unwrap(),
);
}
if let Some(all_receivers) = self.listeners.get_mut(&ListenerType::All) {
send_handler(event, Some(self.aux_data_buf.as_slice()), all_receivers);
send_handler(event, aux_data.clone(), all_receivers);
}
if let Some(err) = err_status {
return err;
}
return Ok(HandlerResult::Handled(num_recipients, event));
return Ok(HandlerResult::Handled(num_recipients, event, aux_data));
}
Ok(HandlerResult::Empty)
}
@ -202,31 +229,26 @@ impl<E, Provider: GenericEvent + Copy> EventManager<E, Provider> {
#[cfg(feature = "std")]
pub mod stdmod {
use crate::event_man::EventReceiver;
use crate::event_man::{EventReceiver, EventWithAuxData};
use crate::events::{EventU16, EventU32, GenericEvent};
use crate::util::AuxData;
use std::sync::mpsc::Receiver;
use std::vec::Vec;
pub struct MpscEventReceiver<Event: GenericEvent = EventU32> {
mpsc_receiver: Receiver<(Event, Option<Vec<u8>>)>,
mpsc_receiver: Receiver<(Event, Option<AuxData>)>,
}
impl<Event: GenericEvent> MpscEventReceiver<Event> {
pub fn new(receiver: Receiver<(Event, Option<Vec<u8>>)>) -> Self {
pub fn new(receiver: Receiver<(Event, Option<AuxData>)>) -> Self {
Self {
mpsc_receiver: receiver,
}
}
}
impl<Event: GenericEvent> EventReceiver<Event> for MpscEventReceiver<Event> {
fn receive(&mut self, aux_data: &mut [u8]) -> Option<Event> {
if let Some((event, params)) = self.mpsc_receiver.try_recv().ok() {
if let Some(params) = params {
if params.len() < aux_data.len() {
aux_data[0..params.len()].copy_from_slice(params.as_slice())
}
}
return Some(event);
fn receive(&mut self) -> Option<EventWithAuxData<Event>> {
if let Ok(event_and_data) = self.mpsc_receiver.try_recv() {
return Some(event_and_data);
}
None
}
@ -241,62 +263,76 @@ mod tests {
use super::*;
use crate::event_man::EventManager;
use crate::events::{EventU32, GenericEvent, Severity};
use crate::util::AuxDataRaw;
use alloc::boxed::Box;
use std::format;
use std::sync::mpsc::{channel, Receiver, SendError, Sender};
type EventAndParams<'a> = (EventU32, Option<&'a [u8]>);
#[derive(Clone)]
struct MpscEventSenderQueue<'a> {
struct MpscEventSenderQueue {
id: u32,
mpsc_sender: Sender<EventAndParams<'a>>,
mpsc_sender: Sender<EventU32WithAuxData>,
}
impl<'a> SendEventProvider<EventU32> for MpscEventSenderQueue<'a> {
type Error = SendError<EventAndParams<'a>>;
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(&mut self, event: EventU32, aux_data: Option<&'a [u8]>) -> Result<(), Self::Error> {
fn send(&mut self, event: EventU32, aux_data: Option<AuxData>) -> Result<(), Self::Error> {
self.mpsc_sender.send((event, aux_data))
}
}
fn check_next_event(expected: EventU32, receiver: &Receiver<EventAndParams>) {
for _ in 0..5 {
if let Ok(event) = receiver.try_recv() {
assert_eq!(event.0, expected);
break;
}
fn check_next_event(
expected: EventU32,
receiver: &Receiver<EventU32WithAuxData>,
) -> Option<AuxData> {
if let Ok(event) = receiver.try_recv() {
assert_eq!(event.0, expected);
return event.1;
}
None
}
fn check_handled_event(
res: HandlerResult<EventU32>,
res: HandlerResult<EventU32, AuxData>,
expected: EventU32,
expected_num_sent: u32,
) {
assert!(matches!(res, HandlerResult::Handled { .. }));
if let HandlerResult::Handled(num_recipients, event) = res {
if let HandlerResult::Handled(num_recipients, event, _aux_data) = res {
assert_eq!(event, expected);
assert_eq!(num_recipients, expected_num_sent);
}
}
#[test]
fn test_basic() {
fn generic_event_man() -> (
Sender<EventU32WithAuxData>,
EventManager<SendError<EventU32WithAuxData>>,
) {
let (event_sender, manager_queue) = channel();
let event_man_receiver = MpscEventReceiver::new(manager_queue);
let mut event_man: EventManager<SendError<EventAndParams>, EventU32> =
EventManager::new(Box::new(event_man_receiver), 128);
(
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 {
id: 0,
mpsc_sender: single_event_sender,
};
let single_event_listener = MpscEventSenderQueue::new(0, single_event_sender);
event_man.subscribe_single(event_grp_0, single_event_listener);
let (group_event_sender_0, group_event_receiver_0) = channel();
let group_event_listener = MpscEventSenderQueue {
@ -324,13 +360,34 @@ mod tests {
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);
event_sender
.send((event_grp_0, Some(AuxData::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 AuxData::Heapless(AuxDataHeapless::Raw(AuxDataRaw::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, manager_queue) = channel();
let event_man_receiver = MpscEventReceiver::new(manager_queue);
let mut event_man: EventManager<SendError<EventAndParams>, EventU32> =
EventManager::new(Box::new(event_man_receiver), 128);
let (event_sender, mut event_man) = generic_event_man();
let res = event_man.try_event_handling();
assert!(res.is_ok());
let hres = res.unwrap();
@ -367,10 +424,7 @@ mod tests {
/// to both group and single events from one listener
#[test]
fn test_listening_to_same_event_and_multi_type() {
let (event_sender, manager_queue) = channel();
let event_man_receiver = MpscEventReceiver::new(manager_queue);
let mut event_man: EventManager<SendError<EventAndParams>, EventU32> =
EventManager::new(Box::new(event_man_receiver), 128);
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();
@ -426,8 +480,8 @@ mod tests {
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<EventAndParams>, EventU32> =
EventManager::new(Box::new(event_man_receiver), 128);
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();

1
fsrc-core/src/lib.rs
View File

@ -28,5 +28,6 @@ pub mod objects;
pub mod pool;
pub mod pus;
pub mod tmtc;
pub mod util;
extern crate downcast_rs;

97
fsrc-core/src/util.rs Normal file
View File

@ -0,0 +1,97 @@
use crate::pool::StoreAddr;
#[cfg(feature = "alloc")]
use alloc::string::String;
#[cfg(feature = "alloc")]
use alloc::string::ToString;
#[cfg(feature = "alloc")]
use alloc::vec::Vec;
#[derive(Debug, Copy, Clone)]
pub enum AuxDataRaw {
U8(u8),
U8Pair((u8, u8)),
U8Triplet((u8, u8, u8)),
I8(i8),
I8Pair((i8, i8)),
I8Triplet((i8, i8, i8)),
U16(u16),
U16Pair((u16, u16)),
U16Triplet((u16, u16, u16)),
I16(i16),
I16Pair((i16, i16)),
I16Triplet((i16, i16, i16)),
U32(u32),
U32Pair((u32, u32)),
U32Triplet((u32, u32, u32)),
I32(i32),
I32Tuple((i32, i32)),
I32Triplet((i32, i32, i32)),
F32(f32),
F32Pair((f32, f32)),
F32Triplet((f32, f32, f32)),
U64(u64),
F64(f64),
}
#[derive(Debug, Copy, Clone)]
pub enum AuxDataHeapless {
Raw(AuxDataRaw),
Store(StoreAddr),
}
impl From<StoreAddr> for AuxDataHeapless {
fn from(x: StoreAddr) -> Self {
Self::Store(x)
}
}
impl From<(u32, u32)> for AuxDataRaw {
fn from(val: (u32, u32)) -> Self {
Self::U32Pair(val)
}
}
impl From<(u32, u32)> for AuxDataHeapless {
fn from(val: (u32, u32)) -> Self {
AuxDataHeapless::Raw(val.into())
}
}
#[cfg(feature = "alloc")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "alloc")))]
#[derive(Debug, Clone)]
pub enum AuxData {
Heapless(AuxDataHeapless),
Vec(Vec<u8>),
String(String),
}
impl From<AuxDataHeapless> for AuxData {
fn from(x: AuxDataHeapless) -> Self {
Self::Heapless(x)
}
}
impl From<Vec<u8>> for AuxData {
fn from(val: Vec<u8>) -> Self {
Self::Vec(val)
}
}
impl From<&[u8]> for AuxData {
fn from(val: &[u8]) -> Self {
Self::Vec(val.to_vec())
}
}
impl From<String> for AuxData {
fn from(val: String) -> Self {
Self::String(val)
}
}
impl From<&str> for AuxData {
fn from(val: &str) -> Self {
Self::String(val.to_string())
}
}