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

Merge remote-tracking branch 'origin/main' into mode-tree-init
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-02-26 12:09:16 +01:00
parent 07cd370667 cf9b115e1e
commit b706dec643
27 changed files with 605 additions and 439 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
satrs/CHANGELOG.md
View File

@ -15,6 +15,14 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
- `ListenerTable` -> `ListenerMapProvider`
- `SenderTable` -> `SenderMapProvider`
- There is an `EventManagerWithMpsc` and a `EventManagerWithBoundedMpsc` helper type now.
- Refactored ECSS TM sender abstractions to be generic over different message queue backends.
- Refactored Verification Reporter abstractions and implementation to be generic over the sender
instead of using trait objects.
## Fixed
- Update deprecated API for `PusScheduler::insert_wrapped_tc_cds_short`
and `PusScheduler::insert_wrapped_tc_cds_long`.
# [v0.2.0-rc.0] 2024-02-21

5
satrs/src/cfdp/filestore.rs
View File

@ -7,7 +7,7 @@ use spacepackets::ByteConversionError;
use std::error::Error;
use std::path::Path;
#[cfg(feature = "std")]
pub use stdmod::*;
pub use std_mod::*;
pub const CRC_32: Crc<u32> = Crc::<u32>::new(&CRC_32_CKSUM);
@ -148,12 +148,11 @@ pub trait VirtualFilestore {
}
#[cfg(feature = "std")]
pub mod stdmod {
pub mod std_mod {
use super::*;
use std::{
fs::{self, File, OpenOptions},
io::{BufReader, Read, Seek, SeekFrom, Write},
path::Path,
};
#[derive(Default)]

13
satrs/src/event_man.rs
View File

@ -85,12 +85,8 @@ pub trait EventSendProvider<EV: GenericEvent, AuxDataProvider = Params> {
/// Generic abstraction for an event receiver.
pub trait EventReceiveProvider<Event: GenericEvent, AuxDataProvider = Params> {
/// This function has to be provided by any event receiver. A receive call may or may not return
/// an event.
///
/// To allow returning arbitrary additional auxiliary data, a mutable slice is passed to the
/// [Self::receive] call as well. Receivers can write data to this slice, but care must be taken
/// to avoid panics due to size missmatches or out of bound writes.
/// This function has to be provided by any event receiver. A call may or may not return
/// an event and optional auxiliary data.
fn try_recv_event(&self) -> Option<(Event, Option<AuxDataProvider>)>;
}
@ -433,10 +429,7 @@ pub mod alloc_mod {
#[cfg(feature = "std")]
pub mod std_mod {
use super::*;
use crate::event_man::{EventReceiveProvider, EventWithAuxData};
use crate::events::{EventU16, EventU32, GenericEvent};
use crate::params::Params;
use std::sync::mpsc::{self};
use std::sync::mpsc;
pub struct MpscEventReceiver<Event: GenericEvent + Send = EventU32> {
mpsc_receiver: mpsc::Receiver<(Event, Option<Params>)>,

2
satrs/src/pus/event.rs
View File

@ -269,7 +269,7 @@ mod tests {
}
impl EcssChannel for TestSender {
fn id(&self) -> ChannelId {
fn channel_id(&self) -> ChannelId {
0
}
}

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

@ -41,8 +41,8 @@ pub trait PusEventMgmtBackendProvider<Event: GenericEvent> {
#[cfg(feature = "heapless")]
pub mod heapless_mod {
use super::*;
use crate::events::{GenericEvent, LargestEventRaw};
use std::marker::PhantomData;
use crate::events::LargestEventRaw;
use core::marker::PhantomData;
#[cfg_attr(doc_cfg, doc(cfg(feature = "heapless")))]
// TODO: After a new version of heapless is released which uses hash32 version 0.3, try using
@ -257,9 +257,8 @@ pub mod alloc_mod {
#[cfg(test)]
mod tests {
use super::*;
use crate::events::SeverityInfo;
use crate::pus::MpscTmAsVecSender;
use std::sync::mpsc::{channel, TryRecvError};
use crate::{events::SeverityInfo, pus::TmAsVecSenderWithMpsc};
use std::sync::mpsc::{self, TryRecvError};
const INFO_EVENT: EventU32TypedSev<SeverityInfo> =
EventU32TypedSev::<SeverityInfo>::const_new(1, 0);
@ -279,8 +278,8 @@ mod tests {
#[test]
fn test_basic() {
let mut event_man = create_basic_man_1();
let (event_tx, event_rx) = channel();
let mut sender = MpscTmAsVecSender::new(0, "test_sender", event_tx);
let (event_tx, event_rx) = mpsc::channel();
let mut sender = TmAsVecSenderWithMpsc::new(0, "test_sender", event_tx);
let event_sent = event_man
.generate_pus_event_tm(&mut sender, &EMPTY_STAMP, INFO_EVENT, None)
.expect("Sending info event failed");
@ -293,8 +292,8 @@ mod tests {
#[test]
fn test_disable_event() {
let mut event_man = create_basic_man_2();
let (event_tx, event_rx) = channel();
let mut sender = MpscTmAsVecSender::new(0, "test", event_tx);
let (event_tx, event_rx) = mpsc::channel();
let mut sender = TmAsVecSenderWithMpsc::new(0, "test", event_tx);
let res = event_man.disable_tm_for_event(&LOW_SEV_EVENT);
assert!(res.is_ok());
assert!(res.unwrap());
@ -316,8 +315,8 @@ mod tests {
#[test]
fn test_reenable_event() {
let mut event_man = create_basic_man_1();
let (event_tx, event_rx) = channel();
let mut sender = MpscTmAsVecSender::new(0, "test", event_tx);
let (event_tx, event_rx) = mpsc::channel();
let mut sender = TmAsVecSenderWithMpsc::new(0, "test", event_tx);
let mut res = event_man.disable_tm_for_event_with_sev(&INFO_EVENT);
assert!(res.is_ok());
assert!(res.unwrap());

14
satrs/src/pus/event_srv.rs
View File

@ -23,11 +23,11 @@ impl<
> PusService5EventHandler<TcInMemConverter, VerificationReporter>
{
pub fn new(
service_handler: PusServiceHelper<TcInMemConverter, VerificationReporter>,
service_helper: PusServiceHelper<TcInMemConverter, VerificationReporter>,
event_request_tx: Sender<EventRequestWithToken>,
) -> Self {
Self {
service_helper: service_handler,
service_helper,
event_request_tx,
}
}
@ -138,7 +138,9 @@ mod tests {
use crate::pus::event_man::EventRequest;
use crate::pus::tests::SimplePusPacketHandler;
use crate::pus::verification::{RequestId, VerificationReporterWithSender};
use crate::pus::verification::{
RequestId, VerificationReporterWithSharedPoolMpscBoundedSender,
};
use crate::{
events::EventU32,
pus::{
@ -155,8 +157,10 @@ mod tests {
struct Pus5HandlerWithStoreTester {
common: PusServiceHandlerWithSharedStoreCommon,
handler:
PusService5EventHandler<EcssTcInSharedStoreConverter, VerificationReporterWithSender>,
handler: PusService5EventHandler<
EcssTcInSharedStoreConverter,
VerificationReporterWithSharedPoolMpscBoundedSender,
>,
}
impl Pus5HandlerWithStoreTester {

377
satrs/src/pus/mod.rs
View File

@ -2,6 +2,8 @@
//!
//! This module contains structures to make working with the PUS C standard easier.
//! The satrs-example application contains various usage examples of these components.
use crate::pool::{StoreAddr, StoreError};
use crate::pus::verification::{TcStateAccepted, TcStateToken, VerificationToken};
use crate::queue::{GenericReceiveError, GenericSendError};
use crate::ChannelId;
use core::fmt::{Display, Formatter};
@ -34,8 +36,6 @@ pub mod verification;
#[cfg(feature = "alloc")]
pub use alloc_mod::*;
use crate::pool::{StoreAddr, StoreError};
use crate::pus::verification::{TcStateAccepted, TcStateToken, VerificationToken};
#[cfg(feature = "std")]
pub use std_mod::*;
@ -63,6 +63,7 @@ pub enum EcssTmtcError {
Store(StoreError),
Pus(PusError),
CantSendAddr(StoreAddr),
CantSendDirectTm,
Send(GenericSendError),
Recv(GenericReceiveError),
}
@ -82,6 +83,9 @@ impl Display for EcssTmtcError {
EcssTmtcError::CantSendAddr(addr) => {
write!(f, "can not send address {addr}")
}
EcssTmtcError::CantSendDirectTm => {
write!(f, "can not send TM directly")
}
EcssTmtcError::Send(send_e) => {
write!(f, "send error {send_e}")
}
@ -123,13 +127,14 @@ impl Error for EcssTmtcError {
EcssTmtcError::Store(e) => Some(e),
EcssTmtcError::Pus(e) => Some(e),
EcssTmtcError::Send(e) => Some(e),
EcssTmtcError::Recv(e) => Some(e),
_ => None,
}
}
}
pub trait EcssChannel: Send {
/// Each sender can have an ID associated with it
fn id(&self) -> ChannelId;
fn channel_id(&self) -> ChannelId;
fn name(&self) -> &'static str {
"unset"
}
@ -138,7 +143,7 @@ pub trait EcssChannel: Send {
/// Generic trait for a user supplied sender object.
///
/// This sender object is responsible for sending PUS telemetry to a TM sink.
pub trait EcssTmSenderCore: EcssChannel {
pub trait EcssTmSenderCore: Send {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError>;
}
@ -146,7 +151,7 @@ pub trait EcssTmSenderCore: EcssChannel {
///
/// This sender object is responsible for sending PUS telecommands to a TC recipient. Each
/// telecommand can optionally have a token which contains its verification state.
pub trait EcssTcSenderCore: EcssChannel {
pub trait EcssTcSenderCore {
fn send_tc(&self, tc: PusTcCreator, token: Option<TcStateToken>) -> Result<(), EcssTmtcError>;
}
@ -221,25 +226,25 @@ impl TryFrom<EcssTcAndToken> for AcceptedEcssTcAndToken {
#[derive(Debug, Clone)]
pub enum TryRecvTmtcError {
Error(EcssTmtcError),
Tmtc(EcssTmtcError),
Empty,
}
impl From<EcssTmtcError> for TryRecvTmtcError {
fn from(value: EcssTmtcError) -> Self {
Self::Error(value)
Self::Tmtc(value)
}
}
impl From<PusError> for TryRecvTmtcError {
fn from(value: PusError) -> Self {
Self::Error(value.into())
Self::Tmtc(value.into())
}
}
impl From<StoreError> for TryRecvTmtcError {
fn from(value: StoreError) -> Self {
Self::Error(value.into())
Self::Tmtc(value.into())
}
}
@ -374,10 +379,9 @@ pub mod std_mod {
use crate::{ChannelId, TargetId};
use alloc::boxed::Box;
use alloc::vec::Vec;
use crossbeam_channel as cb;
use spacepackets::ecss::tc::PusTcReader;
use spacepackets::ecss::tm::PusTmCreator;
use spacepackets::ecss::PusError;
use spacepackets::ecss::{PusError, WritablePusPacket};
use spacepackets::time::cds::TimeProvider;
use spacepackets::time::StdTimestampError;
use spacepackets::time::TimeWriter;
@ -386,6 +390,9 @@ pub mod std_mod {
use std::sync::mpsc::TryRecvError;
use thiserror::Error;
#[cfg(feature = "crossbeam")]
pub use cb_mod::*;
use super::verification::VerificationReportingProvider;
use super::{AcceptedEcssTcAndToken, TcInMemory};
@ -395,32 +402,65 @@ pub mod std_mod {
}
}
impl From<cb::SendError<StoreAddr>> for EcssTmtcError {
fn from(_: cb::SendError<StoreAddr>) -> Self {
Self::Send(GenericSendError::RxDisconnected)
impl EcssTmSenderCore for mpsc::Sender<StoreAddr> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => self
.send(addr)
.map_err(|_| GenericSendError::RxDisconnected)?,
PusTmWrapper::Direct(_) => return Err(EcssTmtcError::CantSendDirectTm),
};
Ok(())
}
}
impl From<cb::TrySendError<StoreAddr>> for EcssTmtcError {
fn from(value: cb::TrySendError<StoreAddr>) -> Self {
match value {
cb::TrySendError::Full(_) => Self::Send(GenericSendError::QueueFull(None)),
cb::TrySendError::Disconnected(_) => Self::Send(GenericSendError::RxDisconnected),
}
impl EcssTmSenderCore for mpsc::SyncSender<StoreAddr> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => self
.try_send(addr)
.map_err(|e| EcssTmtcError::Send(e.into()))?,
PusTmWrapper::Direct(_) => return Err(EcssTmtcError::CantSendDirectTm),
};
Ok(())
}
}
impl EcssTmSenderCore for mpsc::Sender<Vec<u8>> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => return Err(EcssTmtcError::CantSendAddr(addr)),
PusTmWrapper::Direct(tm) => self
.send(tm.to_vec()?)
.map_err(|e| EcssTmtcError::Send(e.into()))?,
};
Ok(())
}
}
impl EcssTmSenderCore for mpsc::SyncSender<Vec<u8>> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => return Err(EcssTmtcError::CantSendAddr(addr)),
PusTmWrapper::Direct(tm) => self
.send(tm.to_vec()?)
.map_err(|e| EcssTmtcError::Send(e.into()))?,
};
Ok(())
}
}
#[derive(Clone)]
pub struct MpscTmInSharedPoolSender {
id: ChannelId,
pub struct TmInSharedPoolSenderWithId<Sender: EcssTmSenderCore> {
channel_id: ChannelId,
name: &'static str,
shared_tm_store: SharedTmPool,
sender: mpsc::Sender<StoreAddr>,
sender: Sender,
}
impl EcssChannel for MpscTmInSharedPoolSender {
fn id(&self) -> ChannelId {
self.id
impl<Sender: EcssTmSenderCore> EcssChannel for TmInSharedPoolSenderWithId<Sender> {
fn channel_id(&self) -> ChannelId {
self.channel_id
}
fn name(&self) -> &'static str {
@ -428,36 +468,31 @@ pub mod std_mod {
}
}
impl MpscTmInSharedPoolSender {
impl<Sender: EcssTmSenderCore> TmInSharedPoolSenderWithId<Sender> {
pub fn send_direct_tm(&self, tm: PusTmCreator) -> Result<(), EcssTmtcError> {
let addr = self.shared_tm_store.add_pus_tm(&tm)?;
self.sender
.send(addr)
.map_err(|_| EcssTmtcError::Send(GenericSendError::RxDisconnected))
self.sender.send_tm(PusTmWrapper::InStore(addr))
}
}
impl EcssTmSenderCore for MpscTmInSharedPoolSender {
impl<Sender: EcssTmSenderCore> EcssTmSenderCore for TmInSharedPoolSenderWithId<Sender> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => {
self.sender.send(addr)?;
Ok(())
}
PusTmWrapper::Direct(tm) => self.send_direct_tm(tm),
if let PusTmWrapper::Direct(tm) = tm {
return self.send_direct_tm(tm);
}
self.sender.send_tm(tm)
}
}
impl MpscTmInSharedPoolSender {
impl<Sender: EcssTmSenderCore> TmInSharedPoolSenderWithId<Sender> {
pub fn new(
id: ChannelId,
name: &'static str,
shared_tm_store: SharedTmPool,
sender: mpsc::Sender<StoreAddr>,
sender: Sender,
) -> Self {
Self {
id,
channel_id: id,
name,
shared_tm_store,
sender,
@ -465,6 +500,51 @@ pub mod std_mod {
}
}
pub type TmInSharedPoolSenderWithMpsc = TmInSharedPoolSenderWithId<mpsc::Sender<StoreAddr>>;
pub type TmInSharedPoolSenderWithBoundedMpsc =
TmInSharedPoolSenderWithId<mpsc::SyncSender<StoreAddr>>;
/// This class can be used if frequent heap allocations during run-time are not an issue.
/// PUS TM packets will be sent around as [Vec]s. Please note that the current implementation
/// of this class can not deal with store addresses, so it is assumed that is is always
/// going to be called with direct packets.
#[derive(Clone)]
pub struct TmAsVecSenderWithId<Sender: EcssTmSenderCore> {
id: ChannelId,
name: &'static str,
sender: Sender,
}
impl From<mpsc::SendError<Vec<u8>>> for EcssTmtcError {
fn from(_: mpsc::SendError<Vec<u8>>) -> Self {
Self::Send(GenericSendError::RxDisconnected)
}
}
impl<Sender: EcssTmSenderCore> TmAsVecSenderWithId<Sender> {
pub fn new(id: u32, name: &'static str, sender: Sender) -> Self {
Self { id, sender, name }
}
}
impl<Sender: EcssTmSenderCore> EcssChannel for TmAsVecSenderWithId<Sender> {
fn channel_id(&self) -> ChannelId {
self.id
}
fn name(&self) -> &'static str {
self.name
}
}
impl<Sender: EcssTmSenderCore> EcssTmSenderCore for TmAsVecSenderWithId<Sender> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
self.sender.send_tm(tm)
}
}
pub type TmAsVecSenderWithMpsc = TmAsVecSenderWithId<mpsc::Sender<Vec<u8>>>;
pub type TmAsVecSenderWithBoundedMpsc = TmAsVecSenderWithId<mpsc::SyncSender<Vec<u8>>>;
pub struct MpscTcReceiver {
id: ChannelId,
name: &'static str,
@ -472,7 +552,7 @@ pub mod std_mod {
}
impl EcssChannel for MpscTcReceiver {
fn id(&self) -> ChannelId {
fn channel_id(&self) -> ChannelId {
self.id
}
@ -485,8 +565,8 @@ pub mod std_mod {
fn recv_tc(&self) -> Result<EcssTcAndToken, TryRecvTmtcError> {
self.receiver.try_recv().map_err(|e| match e {
TryRecvError::Empty => TryRecvTmtcError::Empty,
TryRecvError::Disconnected => TryRecvTmtcError::Error(EcssTmtcError::from(
GenericReceiveError::TxDisconnected(Some(self.id())),
TryRecvError::Disconnected => TryRecvTmtcError::Tmtc(EcssTmtcError::from(
GenericReceiveError::TxDisconnected(Some(self.channel_id())),
)),
})
}
@ -502,133 +582,89 @@ pub mod std_mod {
}
}
/// This class can be used if frequent heap allocations during run-time are not an issue.
/// PUS TM packets will be sent around as [Vec]s. Please note that the current implementation
/// of this class can not deal with store addresses, so it is assumed that is is always
/// going to be called with direct packets.
#[derive(Clone)]
pub struct MpscTmAsVecSender {
id: ChannelId,
name: &'static str,
sender: mpsc::Sender<Vec<u8>>,
}
#[cfg(feature = "crossbeam")]
pub mod cb_mod {
use super::*;
use crossbeam_channel as cb;
impl From<mpsc::SendError<Vec<u8>>> for EcssTmtcError {
fn from(_: mpsc::SendError<Vec<u8>>) -> Self {
Self::Send(GenericSendError::RxDisconnected)
}
}
pub type TmInSharedPoolSenderWithCrossbeam =
TmInSharedPoolSenderWithId<cb::Sender<StoreAddr>>;
impl MpscTmAsVecSender {
pub fn new(id: u32, name: &'static str, sender: mpsc::Sender<Vec<u8>>) -> Self {
Self { id, sender, name }
impl From<cb::SendError<StoreAddr>> for EcssTmtcError {
fn from(_: cb::SendError<StoreAddr>) -> Self {
Self::Send(GenericSendError::RxDisconnected)
}
}
}
impl EcssChannel for MpscTmAsVecSender {
fn id(&self) -> ChannelId {
self.id
}
fn name(&self) -> &'static str {
self.name
}
}
impl EcssTmSenderCore for MpscTmAsVecSender {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => Err(EcssTmtcError::CantSendAddr(addr)),
PusTmWrapper::Direct(tm) => {
let mut vec = Vec::new();
tm.append_to_vec(&mut vec).map_err(EcssTmtcError::Pus)?;
self.sender.send(vec)?;
Ok(())
impl From<cb::TrySendError<StoreAddr>> for EcssTmtcError {
fn from(value: cb::TrySendError<StoreAddr>) -> Self {
match value {
cb::TrySendError::Full(_) => Self::Send(GenericSendError::QueueFull(None)),
cb::TrySendError::Disconnected(_) => {
Self::Send(GenericSendError::RxDisconnected)
}
}
}
}
}
#[derive(Clone)]
pub struct CrossbeamTmInStoreSender {
id: ChannelId,
name: &'static str,
shared_tm_store: SharedTmPool,
sender: crossbeam_channel::Sender<StoreAddr>,
}
impl CrossbeamTmInStoreSender {
pub fn new(
id: ChannelId,
name: &'static str,
shared_tm_store: SharedTmPool,
sender: crossbeam_channel::Sender<StoreAddr>,
) -> Self {
Self {
id,
name,
shared_tm_store,
sender,
impl EcssTmSenderCore for cb::Sender<StoreAddr> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => self
.try_send(addr)
.map_err(|e| EcssTmtcError::Send(e.into()))?,
PusTmWrapper::Direct(_) => return Err(EcssTmtcError::CantSendDirectTm),
};
Ok(())
}
}
}
impl EcssChannel for CrossbeamTmInStoreSender {
fn id(&self) -> ChannelId {
self.id
}
fn name(&self) -> &'static str {
self.name
}
}
impl EcssTmSenderCore for CrossbeamTmInStoreSender {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => self.sender.try_send(addr)?,
PusTmWrapper::Direct(tm) => {
let addr = self.shared_tm_store.add_pus_tm(&tm)?;
self.sender.try_send(addr)?;
}
impl EcssTmSenderCore for cb::Sender<Vec<u8>> {
fn send_tm(&self, tm: PusTmWrapper) -> Result<(), EcssTmtcError> {
match tm {
PusTmWrapper::InStore(addr) => return Err(EcssTmtcError::CantSendAddr(addr)),
PusTmWrapper::Direct(tm) => self
.send(tm.to_vec()?)
.map_err(|e| EcssTmtcError::Send(e.into()))?,
};
Ok(())
}
Ok(())
}
}
pub struct CrossbeamTcReceiver {
id: ChannelId,
name: &'static str,
receiver: cb::Receiver<EcssTcAndToken>,
}
impl CrossbeamTcReceiver {
pub fn new(
pub struct CrossbeamTcReceiver {
id: ChannelId,
name: &'static str,
receiver: cb::Receiver<EcssTcAndToken>,
) -> Self {
Self { id, name, receiver }
}
}
impl EcssChannel for CrossbeamTcReceiver {
fn id(&self) -> ChannelId {
self.id
}
fn name(&self) -> &'static str {
self.name
impl CrossbeamTcReceiver {
pub fn new(
id: ChannelId,
name: &'static str,
receiver: cb::Receiver<EcssTcAndToken>,
) -> Self {
Self { id, name, receiver }
}
}
}
impl EcssTcReceiverCore for CrossbeamTcReceiver {
fn recv_tc(&self) -> Result<EcssTcAndToken, TryRecvTmtcError> {
self.receiver.try_recv().map_err(|e| match e {
cb::TryRecvError::Empty => TryRecvTmtcError::Empty,
cb::TryRecvError::Disconnected => TryRecvTmtcError::Error(EcssTmtcError::from(
GenericReceiveError::TxDisconnected(Some(self.id())),
)),
})
impl EcssChannel for CrossbeamTcReceiver {
fn channel_id(&self) -> ChannelId {
self.id
}
fn name(&self) -> &'static str {
self.name
}
}
impl EcssTcReceiverCore for CrossbeamTcReceiver {
fn recv_tc(&self) -> Result<EcssTcAndToken, TryRecvTmtcError> {
self.receiver.try_recv().map_err(|e| match e {
cb::TryRecvError::Empty => TryRecvTmtcError::Empty,
cb::TryRecvError::Disconnected => TryRecvTmtcError::Tmtc(EcssTmtcError::from(
GenericReceiveError::TxDisconnected(Some(self.id())),
)),
})
}
}
}
@ -906,7 +942,7 @@ pub mod std_mod {
}))
}
Err(e) => match e {
TryRecvTmtcError::Error(e) => Err(PusPacketHandlingError::EcssTmtc(e)),
TryRecvTmtcError::Tmtc(e) => Err(PusPacketHandlingError::EcssTmtc(e)),
TryRecvTmtcError::Empty => Ok(None),
},
}
@ -949,6 +985,9 @@ pub mod tests {
use crate::tmtc::tm_helper::SharedTmPool;
use crate::TargetId;
use super::verification::std_mod::{
VerificationReporterWithSharedPoolMpscBoundedSender, VerificationReporterWithVecMpscSender,
};
use super::verification::tests::{SharedVerificationMap, TestVerificationReporter};
use super::verification::{
TcStateAccepted, VerificationReporterCfg, VerificationReporterWithSender,
@ -956,8 +995,9 @@ pub mod tests {
};
use super::{
EcssTcAndToken, EcssTcInSharedStoreConverter, EcssTcInVecConverter, GenericRoutingError,
MpscTcReceiver, MpscTmAsVecSender, MpscTmInSharedPoolSender, PusPacketHandlerResult,
PusPacketHandlingError, PusRoutingErrorHandler, PusServiceHelper, TcInMemory,
MpscTcReceiver, PusPacketHandlerResult, PusPacketHandlingError, PusRoutingErrorHandler,
PusServiceHelper, TcInMemory, TmAsVecSenderWithId, TmInSharedPoolSenderWithBoundedMpsc,
TmInSharedPoolSenderWithId,
};
pub const TEST_APID: u16 = 0x101;
@ -1002,9 +1042,9 @@ pub mod tests {
tm_buf: [u8; 2048],
tc_pool: SharedStaticMemoryPool,
tm_pool: SharedTmPool,
tc_sender: mpsc::Sender<EcssTcAndToken>,
tc_sender: mpsc::SyncSender<EcssTcAndToken>,
tm_receiver: mpsc::Receiver<StoreAddr>,
verification_handler: VerificationReporterWithSender,
verification_handler: VerificationReporterWithSharedPoolMpscBoundedSender,
}
impl PusServiceHandlerWithSharedStoreCommon {
@ -1014,17 +1054,20 @@ pub mod tests {
/// The PUS service handler is instantiated with a [EcssTcInStoreConverter].
pub fn new() -> (
Self,
PusServiceHelper<EcssTcInSharedStoreConverter, VerificationReporterWithSender>,
PusServiceHelper<
EcssTcInSharedStoreConverter,
VerificationReporterWithSharedPoolMpscBoundedSender,
>,
) {
let pool_cfg = StaticPoolConfig::new(alloc::vec![(16, 16), (8, 32), (4, 64)], false);
let tc_pool = StaticMemoryPool::new(pool_cfg.clone());
let tm_pool = StaticMemoryPool::new(pool_cfg);
let shared_tc_pool = SharedStaticMemoryPool::new(RwLock::new(tc_pool));
let shared_tm_pool = SharedTmPool::new(tm_pool);
let (test_srv_tc_tx, test_srv_tc_rx) = mpsc::channel();
let (tm_tx, tm_rx) = mpsc::channel();
let (test_srv_tc_tx, test_srv_tc_rx) = mpsc::sync_channel(10);
let (tm_tx, tm_rx) = mpsc::sync_channel(10);
let verif_sender = MpscTmInSharedPoolSender::new(
let verif_sender = TmInSharedPoolSenderWithBoundedMpsc::new(
0,
"verif_sender",
shared_tm_pool.clone(),
@ -1032,9 +1075,9 @@ pub mod tests {
);
let verif_cfg = VerificationReporterCfg::new(TEST_APID, 1, 2, 8).unwrap();
let verification_handler =
VerificationReporterWithSender::new(&verif_cfg, Box::new(verif_sender));
VerificationReporterWithSharedPoolMpscBoundedSender::new(&verif_cfg, verif_sender);
let test_srv_tm_sender =
MpscTmInSharedPoolSender::new(0, "TEST_SENDER", shared_tm_pool.clone(), tm_tx);
TmInSharedPoolSenderWithId::new(0, "TEST_SENDER", shared_tm_pool.clone(), tm_tx);
let test_srv_tc_receiver = MpscTcReceiver::new(0, "TEST_RECEIVER", test_srv_tc_rx);
let in_store_converter =
EcssTcInSharedStoreConverter::new(shared_tc_pool.clone(), 2048);
@ -1115,20 +1158,20 @@ pub mod tests {
pub verification_handler: VerificationReporter,
}
impl PusServiceHandlerWithVecCommon<VerificationReporterWithSender> {
impl PusServiceHandlerWithVecCommon<VerificationReporterWithVecMpscSender> {
pub fn new_with_standard_verif_reporter() -> (
Self,
PusServiceHelper<EcssTcInVecConverter, VerificationReporterWithSender>,
PusServiceHelper<EcssTcInVecConverter, VerificationReporterWithVecMpscSender>,
) {
let (test_srv_tc_tx, test_srv_tc_rx) = mpsc::channel();
let (tm_tx, tm_rx) = mpsc::channel();
let verif_sender = MpscTmAsVecSender::new(0, "verififcatio-sender", tm_tx.clone());
let verif_sender = TmAsVecSenderWithId::new(0, "verififcatio-sender", tm_tx.clone());
let verif_cfg = VerificationReporterCfg::new(TEST_APID, 1, 2, 8).unwrap();
let verification_handler =
VerificationReporterWithSender::new(&verif_cfg, Box::new(verif_sender));
VerificationReporterWithSender::new(&verif_cfg, verif_sender);
let test_srv_tm_sender = MpscTmAsVecSender::new(0, "test-sender", tm_tx);
let test_srv_tm_sender = TmAsVecSenderWithId::new(0, "test-sender", tm_tx);
let test_srv_tc_receiver = MpscTcReceiver::new(0, "test-receiver", test_srv_tc_rx);
let in_store_converter = EcssTcInVecConverter::default();
(
@ -1157,7 +1200,7 @@ pub mod tests {
let (test_srv_tc_tx, test_srv_tc_rx) = mpsc::channel();
let (tm_tx, tm_rx) = mpsc::channel();
let test_srv_tm_sender = MpscTmAsVecSender::new(0, "test-sender", tm_tx);
let test_srv_tm_sender = TmAsVecSenderWithId::new(0, "test-sender", tm_tx);
let test_srv_tc_receiver = MpscTcReceiver::new(0, "test-receiver", test_srv_tc_rx);
let in_store_converter = EcssTcInVecConverter::default();
let shared_verif_map = SharedVerificationMap::default();

43
satrs/src/pus/scheduler.rs
View File

@ -340,18 +340,18 @@ pub fn generate_insert_telecommand_app_data(
#[cfg(feature = "alloc")]
pub mod alloc_mod {
use alloc::{
collections::{
btree_map::{Entry, Range},
BTreeMap,
},
vec::Vec,
};
use spacepackets::time::cds::{self, DaysLen24Bits};
use crate::pool::StoreAddr;
use super::*;
use crate::pool::{PoolProvider, StoreAddr, StoreError};
use alloc::collections::btree_map::{Entry, Range};
use alloc::collections::BTreeMap;
use alloc::vec;
use alloc::vec::Vec;
use core::time::Duration;
use spacepackets::ecss::scheduling::TimeWindowType;
use spacepackets::ecss::tc::{PusTc, PusTcReader};
use spacepackets::ecss::PusPacket;
use spacepackets::time::cds::DaysLen24Bits;
use spacepackets::time::{cds, CcsdsTimeProvider, UnixTimestamp};
#[cfg(feature = "std")]
use std::time::SystemTimeError;
@ -461,7 +461,7 @@ pub mod alloc_mod {
}
match self.tc_map.entry(time_stamp) {
Entry::Vacant(e) => {
e.insert(vec![info]);
e.insert(alloc::vec![info]);
}
Entry::Occupied(mut v) => {
v.get_mut().push(info);
@ -498,7 +498,7 @@ pub mod alloc_mod {
/// short timestamp with 16-bit length of days field.
pub fn insert_wrapped_tc_cds_short(
&mut self,
pus_tc: &PusTc,
pus_tc: &(impl IsPusTelecommand + PusPacket + GenericPusTcSecondaryHeader),
pool: &mut (impl PoolProvider + ?Sized),
) -> Result<TcInfo, ScheduleError> {
self.insert_wrapped_tc::<cds::TimeProvider>(pus_tc, pool)
@ -508,7 +508,7 @@ pub mod alloc_mod {
/// long timestamp with a 24-bit length of days field.
pub fn insert_wrapped_tc_cds_long(
&mut self,
pus_tc: &PusTc,
pus_tc: &(impl IsPusTelecommand + PusPacket + GenericPusTcSecondaryHeader),
pool: &mut (impl PoolProvider + ?Sized),
) -> Result<TcInfo, ScheduleError> {
self.insert_wrapped_tc::<cds::TimeProvider<DaysLen24Bits>>(pus_tc, pool)
@ -530,7 +530,7 @@ pub mod alloc_mod {
let range = self.retrieve_by_time_filter(time_window);
let mut del_packets = 0;
let mut res_if_fails = None;
let mut keys_to_delete = Vec::new();
let mut keys_to_delete = alloc::vec::Vec::new();
for time_bucket in range {
for tc in time_bucket.1 {
match pool.delete(tc.addr) {
@ -561,7 +561,10 @@ pub mod alloc_mod {
}
/// Retrieve a range over all scheduled commands.
pub fn retrieve_all(&mut self) -> Range<'_, UnixTimestamp, Vec<TcInfo>> {
pub fn retrieve_all(
&mut self,
) -> alloc::collections::btree_map::Range<'_, UnixTimestamp, alloc::vec::Vec<TcInfo>>
{
self.tc_map.range(..)
}
@ -572,7 +575,7 @@ pub mod alloc_mod {
pub fn retrieve_by_time_filter<TimeProvider: CcsdsTimeProvider>(
&mut self,
time_window: TimeWindow<TimeProvider>,
) -> Range<'_, UnixTimestamp, Vec<TcInfo>> {
) -> Range<'_, UnixTimestamp, alloc::vec::Vec<TcInfo>> {
match time_window.time_window_type() {
TimeWindowType::SelectAll => self.tc_map.range(..),
TimeWindowType::TimeTagToTimeTag => {
@ -761,9 +764,9 @@ pub mod alloc_mod {
mut releaser: R,
tc_store: &(impl PoolProvider + ?Sized),
tc_buf: &mut [u8],
) -> Result<Vec<TcInfo>, (Vec<TcInfo>, StoreError)> {
) -> Result<alloc::vec::Vec<TcInfo>, (alloc::vec::Vec<TcInfo>, StoreError)> {
let tcs_to_release = self.telecommands_to_release();
let mut released_tcs = Vec::new();
let mut released_tcs = alloc::vec::Vec::new();
for tc in tcs_to_release {
for info in tc.1 {
tc_store
@ -835,7 +838,7 @@ pub mod alloc_mod {
}
match self.tc_map.entry(time_stamp) {
Entry::Vacant(e) => {
e.insert(vec![info]);
e.insert(alloc::vec![info]);
}
Entry::Occupied(mut v) => {
v.get_mut().push(info);

4
satrs/src/pus/scheduler_srv.rs
View File

@ -174,7 +174,7 @@ impl<
mod tests {
use crate::pool::{StaticMemoryPool, StaticPoolConfig};
use crate::pus::tests::TEST_APID;
use crate::pus::verification::VerificationReporterWithSender;
use crate::pus::verification::VerificationReporterWithSharedPoolMpscBoundedSender;
use crate::pus::{
scheduler::{self, PusSchedulerProvider, TcInfo},
tests::{PusServiceHandlerWithSharedStoreCommon, PusTestHarness},
@ -199,7 +199,7 @@ mod tests {
common: PusServiceHandlerWithSharedStoreCommon,
handler: PusService11SchedHandler<
EcssTcInSharedStoreConverter,
VerificationReporterWithSender,
VerificationReporterWithSharedPoolMpscBoundedSender,
TestScheduler,
>,
sched_tc_pool: StaticMemoryPool,

16
satrs/src/pus/test.rs
View File

@ -104,7 +104,10 @@ mod tests {
PusServiceHandlerWithSharedStoreCommon, PusServiceHandlerWithVecCommon, PusTestHarness,
SimplePusPacketHandler, TEST_APID,
};
use crate::pus::verification::{RequestId, VerificationReporterWithSender};
use crate::pus::verification::std_mod::{
VerificationReporterWithSharedPoolMpscBoundedSender, VerificationReporterWithVecMpscSender,
};
use crate::pus::verification::RequestId;
use crate::pus::verification::{TcStateAccepted, VerificationToken};
use crate::pus::{
EcssTcInSharedStoreConverter, EcssTcInVecConverter, PusPacketHandlerResult,
@ -120,8 +123,10 @@ mod tests {
struct Pus17HandlerWithStoreTester {
common: PusServiceHandlerWithSharedStoreCommon,
handler:
PusService17TestHandler<EcssTcInSharedStoreConverter, VerificationReporterWithSender>,
handler: PusService17TestHandler<
EcssTcInSharedStoreConverter,
VerificationReporterWithSharedPoolMpscBoundedSender,
>,
}
impl Pus17HandlerWithStoreTester {
@ -158,8 +163,9 @@ mod tests {
}
struct Pus17HandlerWithVecTester {
common: PusServiceHandlerWithVecCommon<VerificationReporterWithSender>,
handler: PusService17TestHandler<EcssTcInVecConverter, VerificationReporterWithSender>,
common: PusServiceHandlerWithVecCommon<VerificationReporterWithVecMpscSender>,
handler:
PusService17TestHandler<EcssTcInVecConverter, VerificationReporterWithVecMpscSender>,
}
impl Pus17HandlerWithVecTester {

144
satrs/src/pus/verification.rs
View File

@ -20,7 +20,7 @@
//! VerificationReportingProvider, VerificationReporterCfg, VerificationReporterWithSender
//! };
//! use satrs::seq_count::SeqCountProviderSimple;
//! use satrs::pus::MpscTmInSharedPoolSender;
//! use satrs::pus::TmInSharedPoolSenderWithMpsc;
//! use satrs::tmtc::tm_helper::SharedTmPool;
//! use spacepackets::ecss::PusPacket;
//! use spacepackets::SpHeader;
@ -35,9 +35,9 @@
//! let shared_tm_store = SharedTmPool::new(tm_pool);
//! let tm_store = shared_tm_store.clone_backing_pool();
//! let (verif_tx, verif_rx) = mpsc::channel();
//! let sender = MpscTmInSharedPoolSender::new(0, "Test Sender", shared_tm_store, verif_tx);
//! let sender = TmInSharedPoolSenderWithMpsc::new(0, "Test Sender", shared_tm_store, verif_tx);
//! let cfg = VerificationReporterCfg::new(TEST_APID, 1, 2, 8).unwrap();
//! let mut reporter = VerificationReporterWithSender::new(&cfg , Box::new(sender));
//! let mut reporter = VerificationReporterWithSender::new(&cfg , sender);
//!
//! let mut sph = SpHeader::tc_unseg(TEST_APID, 0, 0).unwrap();
//! let tc_header = PusTcSecondaryHeader::new_simple(17, 1);
@ -95,10 +95,11 @@ pub use crate::seq_count::SeqCountProviderSimple;
pub use spacepackets::ecss::verification::*;
#[cfg(feature = "alloc")]
pub use alloc_mod::{
VerificationReporter, VerificationReporterCfg, VerificationReporterWithSender,
};
#[cfg_attr(feature = "doc_cfg", doc(cfg(feature = "alloc")))]
pub use alloc_mod::*;
#[cfg(feature = "std")]
#[cfg_attr(feature = "doc_cfg", doc(cfg(feature = "std")))]
pub use std_mod::*;
/// This is a request identifier as specified in 5.4.11.2 c. of the PUS standard.
@ -949,15 +950,13 @@ impl VerificationReporterCore {
}
#[cfg(feature = "alloc")]
mod alloc_mod {
pub mod alloc_mod {
use super::*;
use crate::pus::alloc_mod::EcssTmSender;
use crate::seq_count::SequenceCountProvider;
use alloc::boxed::Box;
use alloc::vec;
use alloc::vec::Vec;
use crate::{
pus::{TmAsVecSenderWithId, TmInSharedPoolSenderWithId},
seq_count::SequenceCountProvider,
};
use core::cell::RefCell;
use spacepackets::ecss::tc::IsPusTelecommand;
#[derive(Clone)]
pub struct VerificationReporterCfg {
@ -992,9 +991,9 @@ mod alloc_mod {
/// TM funnel. This helper will always set those fields to 0.
#[derive(Clone)]
pub struct VerificationReporter {
source_data_buf: RefCell<Vec<u8>>,
pub seq_count_provider: Option<Box<dyn SequenceCountProvider<u16> + Send>>,
pub msg_count_provider: Option<Box<dyn SequenceCountProvider<u16> + Send>>,
source_data_buf: RefCell<alloc::vec::Vec<u8>>,
pub seq_count_provider: Option<alloc::boxed::Box<dyn SequenceCountProvider<u16> + Send>>,
pub msg_count_provider: Option<alloc::boxed::Box<dyn SequenceCountProvider<u16> + Send>>,
pub reporter: VerificationReporterCore,
}
@ -1002,7 +1001,7 @@ mod alloc_mod {
pub fn new(cfg: &VerificationReporterCfg) -> Self {
let reporter = VerificationReporterCore::new(cfg.apid).unwrap();
Self {
source_data_buf: RefCell::new(vec![
source_data_buf: RefCell::new(alloc::vec![
0;
RequestId::SIZE_AS_BYTES
+ cfg.step_field_width
@ -1269,21 +1268,18 @@ mod alloc_mod {
/// Helper object which caches the sender passed as a trait object. Provides the same
/// API as [VerificationReporter] but without the explicit sender arguments.
#[derive(Clone)]
pub struct VerificationReporterWithSender {
pub struct VerificationReporterWithSender<Sender: EcssTmSenderCore + Clone> {
pub reporter: VerificationReporter,
pub sender: Box<dyn EcssTmSender>,
pub sender: Sender,
}
impl VerificationReporterWithSender {
pub fn new(cfg: &VerificationReporterCfg, sender: Box<dyn EcssTmSender>) -> Self {
impl<Sender: EcssTmSenderCore + Clone> VerificationReporterWithSender<Sender> {
pub fn new(cfg: &VerificationReporterCfg, sender: Sender) -> Self {
let reporter = VerificationReporter::new(cfg);
Self::new_from_reporter(reporter, sender)
}
pub fn new_from_reporter(
reporter: VerificationReporter,
sender: Box<dyn EcssTmSender>,
) -> Self {
pub fn new_from_reporter(reporter: VerificationReporter, sender: Sender) -> Self {
Self { reporter, sender }
}
@ -1297,7 +1293,9 @@ mod alloc_mod {
}
}
impl VerificationReportingProvider for VerificationReporterWithSender {
impl<Sender: EcssTmSenderCore + Clone> VerificationReportingProvider
for VerificationReporterWithSender<Sender>
{
delegate! {
to self.reporter {
fn add_tc(
@ -1315,7 +1313,7 @@ mod alloc_mod {
) -> Result<VerificationToken<TcStateAccepted>, VerificationOrSendErrorWithToken<TcStateNone>>
{
self.reporter
.acceptance_success(token, self.sender.as_ref(), time_stamp)
.acceptance_success(token, &self.sender, time_stamp)
}
fn acceptance_failure(
@ -1324,7 +1322,7 @@ mod alloc_mod {
params: FailParams,
) -> Result<(), VerificationOrSendErrorWithToken<TcStateNone>> {
self.reporter
.acceptance_failure(token, self.sender.as_ref(), params)
.acceptance_failure(token, &self.sender, params)
}
fn start_success(
@ -1335,8 +1333,7 @@ mod alloc_mod {
VerificationToken<TcStateStarted>,
VerificationOrSendErrorWithToken<TcStateAccepted>,
> {
self.reporter
.start_success(token, self.sender.as_ref(), time_stamp)
self.reporter.start_success(token, &self.sender, time_stamp)
}
fn start_failure(
@ -1344,8 +1341,7 @@ mod alloc_mod {
token: VerificationToken<TcStateAccepted>,
params: FailParams,
) -> Result<(), VerificationOrSendErrorWithToken<TcStateAccepted>> {
self.reporter
.start_failure(token, self.sender.as_ref(), params)
self.reporter.start_failure(token, &self.sender, params)
}
fn step_success(
@ -1355,7 +1351,7 @@ mod alloc_mod {
step: impl EcssEnumeration,
) -> Result<(), EcssTmtcError> {
self.reporter
.step_success(token, self.sender.as_ref(), time_stamp, step)
.step_success(token, &self.sender, time_stamp, step)
}
fn step_failure(
@ -1363,8 +1359,7 @@ mod alloc_mod {
token: VerificationToken<TcStateStarted>,
params: FailParamsWithStep,
) -> Result<(), VerificationOrSendErrorWithToken<TcStateStarted>> {
self.reporter
.step_failure(token, self.sender.as_ref(), params)
self.reporter.step_failure(token, &self.sender, params)
}
fn completion_success<TcState: WasAtLeastAccepted + Copy>(
@ -1373,7 +1368,7 @@ mod alloc_mod {
time_stamp: &[u8],
) -> Result<(), VerificationOrSendErrorWithToken<TcState>> {
self.reporter
.completion_success(token, self.sender.as_ref(), time_stamp)
.completion_success(token, &self.sender, time_stamp)
}
fn completion_failure<TcState: WasAtLeastAccepted + Copy>(
@ -1382,18 +1377,34 @@ mod alloc_mod {
params: FailParams,
) -> Result<(), VerificationOrSendErrorWithToken<TcState>> {
self.reporter
.completion_failure(token, self.sender.as_ref(), params)
.completion_failure(token, &self.sender, params)
}
}
pub type VerificationReporterWithSharedPoolSender<Sender> =
VerificationReporterWithSender<TmInSharedPoolSenderWithId<Sender>>;
pub type VerificationReporterWithVecSender<Sender> =
VerificationReporterWithSender<TmAsVecSenderWithId<Sender>>;
}
#[cfg(feature = "std")]
mod std_mod {
use crate::pus::verification::VerificationReporterWithSender;
use std::sync::{Arc, Mutex};
pub mod std_mod {
use std::sync::mpsc;
pub type StdVerifReporterWithSender = VerificationReporterWithSender;
pub type SharedStdVerifReporterWithSender = Arc<Mutex<StdVerifReporterWithSender>>;
use crate::pool::StoreAddr;
use super::alloc_mod::{
VerificationReporterWithSharedPoolSender, VerificationReporterWithVecSender,
};
pub type VerificationReporterWithSharedPoolMpscSender =
VerificationReporterWithSharedPoolSender<mpsc::Sender<StoreAddr>>;
pub type VerificationReporterWithSharedPoolMpscBoundedSender =
VerificationReporterWithSharedPoolSender<mpsc::SyncSender<StoreAddr>>;
pub type VerificationReporterWithVecMpscSender =
VerificationReporterWithVecSender<mpsc::Sender<alloc::vec::Vec<u8>>>;
pub type VerificationReporterWithVecMpscBoundedSender =
VerificationReporterWithVecSender<mpsc::SyncSender<alloc::vec::Vec<u8>>>;
}
#[cfg(test)]
@ -1405,10 +1416,11 @@ pub mod tests {
VerificationReporter, VerificationReporterCfg, VerificationReporterWithSender,
VerificationToken,
};
use crate::pus::{EcssChannel, MpscTmInSharedPoolSender, PusTmWrapper};
use crate::pus::{
EcssChannel, PusTmWrapper, TmInSharedPoolSenderWithId, TmInSharedPoolSenderWithMpsc,
};
use crate::tmtc::tm_helper::SharedTmPool;
use crate::ChannelId;
use alloc::boxed::Box;
use alloc::format;
use alloc::sync::Arc;
use hashbrown::HashMap;
@ -1637,7 +1649,7 @@ pub mod tests {
}
impl EcssChannel for TestSender {
fn id(&self) -> ChannelId {
fn channel_id(&self) -> ChannelId {
0
}
fn name(&self) -> &'static str {
@ -1688,13 +1700,13 @@ pub mod tests {
&mut self.vr
}
}
struct TestBaseWithHelper<'a> {
helper: VerificationReporterWithSender,
struct TestBaseWithHelper<'a, Sender: EcssTmSenderCore + Clone + 'static> {
helper: VerificationReporterWithSender<Sender>,
#[allow(dead_code)]
tc: PusTcCreator<'a>,
}
impl<'a> TestBaseWithHelper<'a> {
impl<'a, Sender: EcssTmSenderCore + Clone + 'static> TestBaseWithHelper<'a, Sender> {
fn rep(&mut self) -> &mut VerificationReporter {
&mut self.helper.reporter
}
@ -1725,12 +1737,15 @@ pub mod tests {
(TestBase { vr: reporter, tc }, init_tok)
}
fn base_with_helper_init() -> (TestBaseWithHelper<'static>, VerificationToken<TcStateNone>) {
fn base_with_helper_init() -> (
TestBaseWithHelper<'static, TestSender>,
VerificationToken<TcStateNone>,
) {
let mut reporter = base_reporter();
let (tc, _) = base_tc_init(None);
let init_tok = reporter.add_tc(&tc);
let sender = TestSender::default();
let helper = VerificationReporterWithSender::new_from_reporter(reporter, Box::new(sender));
let helper = VerificationReporterWithSender::new_from_reporter(reporter, sender);
(TestBaseWithHelper { helper, tc }, init_tok)
}
@ -1758,7 +1773,7 @@ pub mod tests {
let shared_tm_store = SharedTmPool::new(pool);
let (tx, _) = mpsc::channel();
let mpsc_verif_sender =
MpscTmInSharedPoolSender::new(0, "verif_sender", shared_tm_store, tx);
TmInSharedPoolSenderWithMpsc::new(0, "verif_sender", shared_tm_store, tx);
is_send(&mpsc_verif_sender);
}
@ -1785,8 +1800,7 @@ pub mod tests {
b.helper
.acceptance_success(tok, &EMPTY_STAMP)
.expect("Sending acceptance success failed");
let sender: &mut TestSender = b.helper.sender.downcast_mut().unwrap();
acceptance_check(sender, &tok.req_id);
acceptance_check(&mut b.helper.sender, &tok.req_id);
}
fn acceptance_fail_check(sender: &mut TestSender, req_id: RequestId, stamp_buf: [u8; 7]) {
@ -1830,8 +1844,7 @@ pub mod tests {
b.helper
.acceptance_failure(tok, fail_params)
.expect("Sending acceptance success failed");
let sender: &mut TestSender = b.helper.sender.downcast_mut().unwrap();
acceptance_fail_check(sender, tok.req_id, stamp_buf);
acceptance_fail_check(&mut b.helper.sender, tok.req_id, stamp_buf);
}
#[test]
@ -1961,8 +1974,7 @@ pub mod tests {
b.helper
.start_failure(accepted_token, fail_params)
.expect("Start failure failure");
let sender: &mut TestSender = b.helper.sender.downcast_mut().unwrap();
start_fail_check(sender, tok.req_id, fail_data_raw);
start_fail_check(&mut b.helper.sender, tok.req_id, fail_data_raw);
}
fn step_success_check(sender: &mut TestSender, req_id: RequestId) {
@ -2059,9 +2071,8 @@ pub mod tests {
b.helper
.step_success(&started_token, &EMPTY_STAMP, EcssEnumU8::new(1))
.expect("Sending step 1 success failed");
let sender: &mut TestSender = b.helper.sender.downcast_mut().unwrap();
assert_eq!(sender.service_queue.borrow().len(), 4);
step_success_check(sender, tok.req_id);
assert_eq!(b.helper.sender.service_queue.borrow().len(), 4);
step_success_check(&mut b.helper.sender, tok.req_id);
}
fn check_step_failure(sender: &mut TestSender, req_id: RequestId, fail_data_raw: [u8; 4]) {
@ -2191,8 +2202,7 @@ pub mod tests {
b.helper
.step_failure(started_token, fail_params)
.expect("Step failure failed");
let sender: &mut TestSender = b.helper.sender.downcast_mut().unwrap();
check_step_failure(sender, req_id, fail_data_raw);
check_step_failure(&mut b.helper.sender, req_id, fail_data_raw);
}
fn completion_fail_check(sender: &mut TestSender, req_id: RequestId) {
@ -2278,8 +2288,7 @@ pub mod tests {
b.helper
.completion_failure(started_token, fail_params)
.expect("Completion failure");
let sender: &mut TestSender = b.helper.sender.downcast_mut().unwrap();
completion_fail_check(sender, req_id);
completion_fail_check(&mut b.helper.sender, req_id);
}
fn completion_success_check(sender: &mut TestSender, req_id: RequestId) {
@ -2355,8 +2364,7 @@ pub mod tests {
b.helper
.completion_success(started_token, &EMPTY_STAMP)
.expect("Sending completion success failed");
let sender: &mut TestSender = b.helper.sender.downcast_mut().unwrap();
completion_success_check(sender, tok.req_id);
completion_success_check(&mut b.helper.sender, tok.req_id);
}
#[test]
@ -2368,9 +2376,9 @@ pub mod tests {
let shared_tm_pool = shared_tm_store.clone_backing_pool();
let (verif_tx, verif_rx) = mpsc::channel();
let sender =
MpscTmInSharedPoolSender::new(0, "Verification Sender", shared_tm_store, verif_tx);
TmInSharedPoolSenderWithId::new(0, "Verification Sender", shared_tm_store, verif_tx);
let cfg = VerificationReporterCfg::new(TEST_APID, 1, 2, 8).unwrap();
let mut reporter = VerificationReporterWithSender::new(&cfg, Box::new(sender));
let mut reporter = VerificationReporterWithSender::new(&cfg, sender);
let mut sph = SpHeader::tc_unseg(TEST_APID, 0, 0).unwrap();
let tc_header = PusTcSecondaryHeader::new_simple(17, 1);

41
satrs/src/queue.rs
View File

@ -1,6 +1,8 @@
use core::fmt::{Display, Formatter};
#[cfg(feature = "std")]
use std::error::Error;
#[cfg(feature = "std")]
use std::sync::mpsc;
/// Generic channel ID type.
pub type ChannelId = u32;
@ -10,6 +12,7 @@ pub type ChannelId = u32;
pub enum GenericSendError {
RxDisconnected,
QueueFull(Option<u32>),
TargetDoesNotExist(ChannelId),
}
impl Display for GenericSendError {
@ -21,6 +24,9 @@ impl Display for GenericSendError {
GenericSendError::QueueFull(max_cap) => {
write!(f, "queue with max capacity of {max_cap:?} is full")
}
GenericSendError::TargetDoesNotExist(target) => {
write!(f, "target queue with ID {target} does not exist")
}
}
}
}
@ -53,7 +59,6 @@ impl Error for GenericReceiveError {}
#[derive(Debug, Clone)]
pub enum GenericTargetedMessagingError {
TargetDoesNotExist(ChannelId),
Send(GenericSendError),
Receive(GenericReceiveError),
}
@ -68,3 +73,37 @@ impl From<GenericReceiveError> for GenericTargetedMessagingError {
Self::Receive(value)
}
}
#[cfg(feature = "std")]
impl<T> From<mpsc::SendError<T>> for GenericSendError {
fn from(_: mpsc::SendError<T>) -> Self {
GenericSendError::RxDisconnected
}
}
#[cfg(feature = "std")]
impl<T> From<mpsc::TrySendError<T>> for GenericSendError {
fn from(err: mpsc::TrySendError<T>) -> Self {
match err {
mpsc::TrySendError::Full(_) => GenericSendError::QueueFull(None),
mpsc::TrySendError::Disconnected(_) => GenericSendError::RxDisconnected,
}
}
}
#[cfg(feature = "crossbeam")]
impl<T> From<crossbeam_channel::SendError<T>> for GenericSendError {
fn from(_: crossbeam_channel::SendError<T>) -> Self {
GenericSendError::RxDisconnected
}
}
#[cfg(feature = "crossbeam")]
impl<T> From<crossbeam_channel::TrySendError<T>> for GenericSendError {
fn from(err: crossbeam_channel::TrySendError<T>) -> Self {
match err {
crossbeam_channel::TrySendError::Full(_) => GenericSendError::QueueFull(None),
crossbeam_channel::TrySendError::Disconnected(_) => GenericSendError::RxDisconnected,
}
}
}

4
satrs/src/request.rs
View File

@ -209,9 +209,7 @@ mod std_mod {
.map_err(|_| GenericSendError::RxDisconnected)?;
return Ok(());
}
Err(GenericTargetedMessagingError::TargetDoesNotExist(
target_channel_id,
))
Err(GenericSendError::TargetDoesNotExist(target_channel_id).into())
}
}

12
satrs/tests/pus_events.rs
View File

@ -5,10 +5,10 @@ use satrs::events::{EventU32, EventU32TypedSev, Severity, SeverityInfo};
use satrs::params::U32Pair;
use satrs::params::{Params, ParamsHeapless, WritableToBeBytes};
use satrs::pus::event_man::{DefaultPusEventMgmtBackend, EventReporter, PusEventDispatcher};
use satrs::pus::MpscTmAsVecSender;
use satrs::pus::TmAsVecSenderWithMpsc;
use spacepackets::ecss::tm::PusTmReader;
use spacepackets::ecss::{PusError, PusPacket};
use std::sync::mpsc::{channel, SendError, TryRecvError};
use std::sync::mpsc::{self, SendError, TryRecvError};
use std::thread;
const INFO_EVENT: EventU32TypedSev<SeverityInfo> =
@ -24,21 +24,21 @@ pub enum CustomTmSenderError {
#[test]
fn test_threaded_usage() {
let (event_sender, event_man_receiver) = channel();
let (event_sender, event_man_receiver) = mpsc::channel();
let event_receiver = MpscEventU32Receiver::new(event_man_receiver);
let mut event_man = EventManagerWithMpsc::new(event_receiver);
let (pus_event_man_tx, pus_event_man_rx) = channel();
let (pus_event_man_tx, pus_event_man_rx) = mpsc::channel();
let pus_event_man_send_provider = EventU32SenderMpsc::new(1, pus_event_man_tx);
event_man.subscribe_all(pus_event_man_send_provider.channel_id());
event_man.add_sender(pus_event_man_send_provider);
let (event_tx, event_rx) = channel();
let (event_tx, event_rx) = mpsc::channel();
let reporter = EventReporter::new(0x02, 128).expect("Creating event reporter failed");
let mut pus_event_man =
PusEventDispatcher::new(reporter, DefaultPusEventMgmtBackend::default());
// PUS + Generic event manager thread
let jh0 = thread::spawn(move || {
let mut sender = MpscTmAsVecSender::new(0, "event_sender", event_tx);
let mut sender = TmAsVecSenderWithMpsc::new(0, "event_sender", event_tx);
let mut event_cnt = 0;
let mut params_array: [u8; 128] = [0; 128];
loop {

13
satrs/tests/pus_verification.rs
View File

@ -6,7 +6,7 @@ pub mod crossbeam_test {
FailParams, RequestId, VerificationReporterCfg, VerificationReporterWithSender,
VerificationReportingProvider,
};
use satrs::pus::CrossbeamTmInStoreSender;
use satrs::pus::TmInSharedPoolSenderWithCrossbeam;
use satrs::tmtc::tm_helper::SharedTmPool;
use spacepackets::ecss::tc::{PusTcCreator, PusTcReader, PusTcSecondaryHeader};
use spacepackets::ecss::tm::PusTmReader;
@ -40,10 +40,13 @@ pub mod crossbeam_test {
let shared_tc_pool_0 = Arc::new(RwLock::new(StaticMemoryPool::new(pool_cfg)));
let shared_tc_pool_1 = shared_tc_pool_0.clone();
let (tx, rx) = crossbeam_channel::bounded(10);
let sender =
CrossbeamTmInStoreSender::new(0, "verif_sender", shared_tm_pool.clone(), tx.clone());
let mut reporter_with_sender_0 =
VerificationReporterWithSender::new(&cfg, Box::new(sender));
let sender = TmInSharedPoolSenderWithCrossbeam::new(
0,
"verif_sender",
shared_tm_pool.clone(),
tx.clone(),
);
let mut reporter_with_sender_0 = VerificationReporterWithSender::new(&cfg, sender);
let mut reporter_with_sender_1 = reporter_with_sender_0.clone();
// For test purposes, we retrieve the request ID from the TCs and pass them to the receiver
// tread.