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

trying to get rid of more duplication

Browse Source
This commit is contained in:
Robin Müller 2024-04-14 20:45:44 +02:00
parent 2011586478
commit 6c7f9d10fb
Signed by: muellerr
GPG Key ID: A649FB78196E3849
23 changed files with 479 additions and 326 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
satrs-example/src/acs/mgm.rs
View File

@ -11,7 +11,7 @@ use std::sync::{Arc, Mutex};
use satrs::mode::{
ModeAndSubmode, ModeError, ModeProvider, ModeReply, ModeRequest, ModeRequestHandler,
};
use satrs::pus::{EcssTmSenderCore, PusTmVariant};
use satrs::pus::{EcssTmSender, PusTmVariant};
use satrs::request::{GenericMessage, MessageMetadata, UniqueApidTargetId};
use satrs_example::config::components::PUS_MODE_SERVICE;
@ -64,7 +64,7 @@ pub struct MpscModeLeafInterface {
/// Example MGM device handler strongly based on the LIS3MDL MEMS device.
#[derive(new)]
#[allow(clippy::too_many_arguments)]
pub struct MgmHandlerLis3Mdl<ComInterface: SpiInterface, TmSender: EcssTmSenderCore> {
pub struct MgmHandlerLis3Mdl<ComInterface: SpiInterface, TmSender: EcssTmSender> {
id: UniqueApidTargetId,
dev_str: &'static str,
mode_interface: MpscModeLeafInterface,
@ -85,9 +85,7 @@ pub struct MgmHandlerLis3Mdl<ComInterface: SpiInterface, TmSender: EcssTmSenderC
stamp_helper: TimeStampHelper,
}
impl<ComInterface: SpiInterface, TmSender: EcssTmSenderCore>
MgmHandlerLis3Mdl<ComInterface, TmSender>
{
impl<ComInterface: SpiInterface, TmSender: EcssTmSender> MgmHandlerLis3Mdl<ComInterface, TmSender> {
pub fn periodic_operation(&mut self) {
self.stamp_helper.update_from_now();
// Handle requests.
@ -203,7 +201,7 @@ impl<ComInterface: SpiInterface, TmSender: EcssTmSenderCore>
}
}
impl<ComInterface: SpiInterface, TmSender: EcssTmSenderCore> ModeProvider
impl<ComInterface: SpiInterface, TmSender: EcssTmSender> ModeProvider
for MgmHandlerLis3Mdl<ComInterface, TmSender>
{
fn mode_and_submode(&self) -> ModeAndSubmode {
@ -211,7 +209,7 @@ impl<ComInterface: SpiInterface, TmSender: EcssTmSenderCore> ModeProvider
}
}
impl<ComInterface: SpiInterface, TmSender: EcssTmSenderCore> ModeRequestHandler
impl<ComInterface: SpiInterface, TmSender: EcssTmSender> ModeRequestHandler
for MgmHandlerLis3Mdl<ComInterface, TmSender>
{
type Error = ModeError;

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

@ -5,7 +5,7 @@ use satrs::event_man::{EventMessageU32, EventRoutingError};
use satrs::params::WritableToBeBytes;
use satrs::pus::event::EventTmHookProvider;
use satrs::pus::verification::VerificationReporter;
use satrs::pus::EcssTmSenderCore;
use satrs::pus::EcssTmSender;
use satrs::request::UniqueApidTargetId;
use satrs::{
event_man::{
@ -38,7 +38,7 @@ impl EventTmHookProvider for EventApidSetter {
/// The PUS event handler subscribes for all events and converts them into ECSS PUS 5 event
/// packets. It also handles the verification completion of PUS event service requests.
pub struct PusEventHandler<TmSender: EcssTmSenderCore> {
pub struct PusEventHandler<TmSender: EcssTmSender> {
event_request_rx: mpsc::Receiver<EventRequestWithToken>,
pus_event_dispatcher: DefaultPusEventU32Dispatcher<()>,
pus_event_man_rx: mpsc::Receiver<EventMessageU32>,
@ -49,7 +49,7 @@ pub struct PusEventHandler<TmSender: EcssTmSenderCore> {
event_apid_setter: EventApidSetter,
}
impl<TmSender: EcssTmSenderCore> PusEventHandler<TmSender> {
impl<TmSender: EcssTmSender> PusEventHandler<TmSender> {
pub fn new(
tm_sender: TmSender,
verif_handler: VerificationReporter,
@ -177,12 +177,12 @@ impl EventManagerWrapper {
}
}
pub struct EventHandler<TmSender: EcssTmSenderCore> {
pub struct EventHandler<TmSender: EcssTmSender> {
pub event_man_wrapper: EventManagerWrapper,
pub pus_event_handler: PusEventHandler<TmSender>,
}
impl<TmSender: EcssTmSenderCore> EventHandler<TmSender> {
impl<TmSender: EcssTmSender> EventHandler<TmSender> {
pub fn new(
tm_sender: TmSender,
event_request_rx: mpsc::Receiver<EventRequestWithToken>,

9
satrs-example/src/main.rs
View File

@ -17,8 +17,7 @@ use pus::test::create_test_service_dynamic;
use satrs::hal::std::tcp_server::ServerConfig;
use satrs::hal::std::udp_server::UdpTcServer;
use satrs::request::GenericMessage;
use satrs::tmtc::tc_helper::{SharedTcPool, TcSenderSharedPool};
use satrs::tmtc::tm_helper::SharedTmPool;
use satrs::tmtc::tc_helper::{PacketSenderSharedPool, SharedPacketPool};
use satrs_example::config::pool::{create_sched_tc_pool, create_static_pools};
use satrs_example::config::tasks::{
FREQ_MS_AOCS, FREQ_MS_EVENT_HANDLING, FREQ_MS_PUS_STACK, FREQ_MS_UDP_TMTC,
@ -51,8 +50,8 @@ use std::time::Duration;
#[allow(dead_code)]
fn static_tmtc_pool_main() {
let (tm_pool, tc_pool) = create_static_pools();
let shared_tm_pool = SharedTmPool::new(tm_pool);
let shared_tc_pool = SharedTcPool(Arc::new(RwLock::new(tc_pool)));
let shared_tm_pool = (Arc::new(RwLock::new(tm_pool)));
let shared_tc_pool = (Arc::new(RwLock::new(tc_pool)));
let (tc_source_tx, tc_source_rx) = mpsc::sync_channel(50);
let (tm_funnel_tx, tm_funnel_rx) = mpsc::sync_channel(50);
let (tm_server_tx, tm_server_rx) = mpsc::sync_channel(50);
@ -75,7 +74,7 @@ fn static_tmtc_pool_main() {
// This helper structure is used by all telecommand providers which need to send telecommands
// to the TC source.
let tc_source = TcSenderSharedPool::new(tc_source_tx, shared_tc_pool.clone());
let tc_source = PacketSenderSharedPool::new(tc_source_tx, shared_tc_pool.clone());
// Create event handling components
// These sender handles are used to send event requests, for example to enable or disable

15
satrs-example/src/pus/action.rs
View File

@ -11,7 +11,7 @@ use satrs::pus::verification::{
};
use satrs::pus::{
ActiveRequestProvider, EcssTcAndToken, EcssTcInMemConverter, EcssTcInSharedStoreConverter,
EcssTcInVecConverter, EcssTmSenderCore, EcssTmtcError, GenericConversionError, MpscTcReceiver,
EcssTcInVecConverter, EcssTmSender, EcssTmtcError, GenericConversionError, MpscTcReceiver,
MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded, PusPacketHandlerResult, PusReplyHandler,
PusServiceHelper, PusTcToRequestConverter, PusTmAsVec, PusTmInPool, TmInSharedPoolSender,
};
@ -48,7 +48,7 @@ impl PusReplyHandler<ActivePusActionRequestStd, ActionReplyPus> for ActionReplyH
fn handle_unrequested_reply(
&mut self,
reply: &GenericMessage<ActionReplyPus>,
_tm_sender: &impl EcssTmSenderCore,
_tm_sender: &impl EcssTmSender,
) -> Result<(), Self::Error> {
warn!("received unexpected reply for service 8: {reply:?}");
Ok(())
@ -58,7 +58,7 @@ impl PusReplyHandler<ActivePusActionRequestStd, ActionReplyPus> for ActionReplyH
&mut self,
reply: &GenericMessage<ActionReplyPus>,
active_request: &ActivePusActionRequestStd,
tm_sender: &(impl EcssTmSenderCore + ?Sized),
tm_sender: &(impl EcssTmSender + ?Sized),
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<bool, Self::Error> {
@ -121,7 +121,7 @@ impl PusReplyHandler<ActivePusActionRequestStd, ActionReplyPus> for ActionReplyH
fn handle_request_timeout(
&mut self,
active_request: &ActivePusActionRequestStd,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(), Self::Error> {
@ -145,7 +145,7 @@ impl PusTcToRequestConverter<ActivePusActionRequestStd, ActionRequest> for Actio
&mut self,
token: VerificationToken<TcStateAccepted>,
tc: &PusTcReader,
tm_sender: &(impl EcssTmSenderCore + ?Sized),
tm_sender: &(impl EcssTmSender + ?Sized),
verif_reporter: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(ActivePusActionRequestStd, ActionRequest), Self::Error> {
@ -247,8 +247,7 @@ pub fn create_action_service_dynamic(
}
}
pub struct ActionServiceWrapper<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
{
pub struct ActionServiceWrapper<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter> {
pub(crate) service: PusTargetedRequestService<
MpscTcReceiver,
TmSender,
@ -263,7 +262,7 @@ pub struct ActionServiceWrapper<TmSender: EcssTmSenderCore, TcInMemConverter: Ec
>,
}
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter> TargetedPusService
impl<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter> TargetedPusService
for ActionServiceWrapper<TmSender, TcInMemConverter>
{
/// Returns [true] if the packet handling is finished.

6
satrs-example/src/pus/event.rs
View File

@ -8,7 +8,7 @@ use satrs::pus::event_srv::PusEventServiceHandler;
use satrs::pus::verification::VerificationReporter;
use satrs::pus::{
EcssTcAndToken, EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcInVecConverter,
EcssTmSenderCore, MpscTcReceiver, MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded,
EcssTmSender, MpscTcReceiver, MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded,
PusPacketHandlerResult, PusServiceHelper, PusTmAsVec, PusTmInPool, TmInSharedPoolSender,
};
use satrs_example::config::components::PUS_EVENT_MANAGEMENT;
@ -56,12 +56,12 @@ pub fn create_event_service_dynamic(
}
}
pub struct EventServiceWrapper<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter> {
pub struct EventServiceWrapper<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter> {
pub handler:
PusEventServiceHandler<MpscTcReceiver, TmSender, TcInMemConverter, VerificationReporter>,
}
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
impl<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter>
EventServiceWrapper<TmSender, TcInMemConverter>
{
pub fn poll_and_handle_next_tc(&mut self, time_stamp: &[u8]) -> HandlingStatus {

14
satrs-example/src/pus/hk.rs
View File

@ -8,7 +8,7 @@ use satrs::pus::verification::{
};
use satrs::pus::{
ActivePusRequestStd, ActiveRequestProvider, DefaultActiveRequestMap, EcssTcAndToken,
EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcInVecConverter, EcssTmSenderCore,
EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcInVecConverter, EcssTmSender,
EcssTmtcError, GenericConversionError, MpscTcReceiver, MpscTmAsVecSender,
MpscTmInSharedPoolSenderBounded, PusPacketHandlerResult, PusReplyHandler, PusServiceHelper,
PusTcToRequestConverter, PusTmAsVec, PusTmInPool, TmInSharedPoolSender,
@ -46,7 +46,7 @@ impl PusReplyHandler<ActivePusRequestStd, HkReply> for HkReplyHandler {
fn handle_unrequested_reply(
&mut self,
reply: &GenericMessage<HkReply>,
_tm_sender: &impl EcssTmSenderCore,
_tm_sender: &impl EcssTmSender,
) -> Result<(), Self::Error> {
log::warn!("received unexpected reply for service 3: {reply:?}");
Ok(())
@ -56,7 +56,7 @@ impl PusReplyHandler<ActivePusRequestStd, HkReply> for HkReplyHandler {
&mut self,
reply: &GenericMessage<HkReply>,
active_request: &ActivePusRequestStd,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<bool, Self::Error> {
@ -77,7 +77,7 @@ impl PusReplyHandler<ActivePusRequestStd, HkReply> for HkReplyHandler {
fn handle_request_timeout(
&mut self,
active_request: &ActivePusRequestStd,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(), Self::Error> {
@ -111,7 +111,7 @@ impl PusTcToRequestConverter<ActivePusRequestStd, HkRequest> for HkRequestConver
&mut self,
token: VerificationToken<TcStateAccepted>,
tc: &PusTcReader,
tm_sender: &(impl EcssTmSenderCore + ?Sized),
tm_sender: &(impl EcssTmSender + ?Sized),
verif_reporter: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(ActivePusRequestStd, HkRequest), Self::Error> {
@ -282,7 +282,7 @@ pub fn create_hk_service_dynamic(
}
}
pub struct HkServiceWrapper<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter> {
pub struct HkServiceWrapper<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter> {
pub(crate) service: PusTargetedRequestService<
MpscTcReceiver,
TmSender,
@ -297,7 +297,7 @@ pub struct HkServiceWrapper<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTc
>,
}
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
impl<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter>
HkServiceWrapper<TmSender, TcInMemConverter>
{
pub fn poll_and_handle_next_tc(&mut self, time_stamp: &[u8]) -> HandlingStatus {

18
satrs-example/src/pus/mod.rs
View File

@ -6,9 +6,9 @@ use satrs::pus::verification::{
};
use satrs::pus::{
ActiveRequestMapProvider, ActiveRequestProvider, EcssTcAndToken, EcssTcInMemConverter,
EcssTcReceiverCore, EcssTmSenderCore, EcssTmtcError, GenericConversionError,
GenericRoutingError, PusPacketHandlerResult, PusPacketHandlingError, PusReplyHandler,
PusRequestRouter, PusServiceHelper, PusTcToRequestConverter, TcInMemory,
EcssTcReceiverCore, EcssTmSender, EcssTmtcError, GenericConversionError, GenericRoutingError,
PusPacketHandlerResult, PusPacketHandlingError, PusReplyHandler, PusRequestRouter,
PusServiceHelper, PusTcToRequestConverter, TcInMemory,
};
use satrs::queue::{GenericReceiveError, GenericSendError};
use satrs::request::{Apid, GenericMessage, MessageMetadata};
@ -52,7 +52,7 @@ pub struct PusTcMpscRouter {
pub mode_tc_sender: Sender<EcssTcAndToken>,
}
pub struct PusTcDistributor<TmSender: EcssTmSenderCore> {
pub struct PusTcDistributor<TmSender: EcssTmSender> {
pub id: ComponentId,
pub tm_sender: TmSender,
pub verif_reporter: VerificationReporter,
@ -60,7 +60,7 @@ pub struct PusTcDistributor<TmSender: EcssTmSenderCore> {
stamp_helper: TimeStampHelper,
}
impl<TmSender: EcssTmSenderCore> PusTcDistributor<TmSender> {
impl<TmSender: EcssTmSender> PusTcDistributor<TmSender> {
pub fn new(tm_sender: TmSender, pus_router: PusTcMpscRouter) -> Self {
Self {
id: PUS_ROUTING_SERVICE.raw(),
@ -186,7 +186,7 @@ pub trait TargetedPusService {
/// 3. [Self::check_for_request_timeouts] which checks for request timeouts, covering step 7.
pub struct PusTargetedRequestService<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
RequestConverter: PusTcToRequestConverter<ActiveRequestInfo, RequestType, Error = GenericConversionError>,
@ -208,7 +208,7 @@ pub struct PusTargetedRequestService<
impl<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
RequestConverter: PusTcToRequestConverter<ActiveRequestInfo, RequestType, Error = GenericConversionError>,
@ -437,7 +437,7 @@ where
/// Generic timeout handling: Handle the verification failure with a dedicated return code
/// and also log the error.
pub fn generic_pus_request_timeout_handler(
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
active_request: &(impl ActiveRequestProvider + Debug),
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
@ -591,7 +591,7 @@ pub(crate) mod tests {
/// Dummy sender component which does nothing on the [Self::send_tm] call.
///
/// Useful for unit tests.
impl EcssTmSenderCore for DummySender {
impl EcssTmSender for DummySender {
fn send_tm(&self, _source_id: ComponentId, _tm: PusTmVariant) -> Result<(), EcssTmtcError> {
Ok(())
}

14
satrs-example/src/pus/mode.rs
View File

@ -20,7 +20,7 @@ use satrs::{
self, FailParams, TcStateAccepted, TcStateStarted, VerificationReportingProvider,
VerificationToken,
},
ActivePusRequestStd, ActiveRequestProvider, EcssTmSenderCore, EcssTmtcError,
ActivePusRequestStd, ActiveRequestProvider, EcssTmSender, EcssTmtcError,
GenericConversionError, PusReplyHandler, PusTcToRequestConverter, PusTmVariant,
},
request::UniqueApidTargetId,
@ -53,7 +53,7 @@ impl PusReplyHandler<ActivePusRequestStd, ModeReply> for ModeReplyHandler {
fn handle_unrequested_reply(
&mut self,
reply: &GenericMessage<ModeReply>,
_tm_sender: &impl EcssTmSenderCore,
_tm_sender: &impl EcssTmSender,
) -> Result<(), Self::Error> {
log::warn!("received unexpected reply for mode service 5: {reply:?}");
Ok(())
@ -63,7 +63,7 @@ impl PusReplyHandler<ActivePusRequestStd, ModeReply> for ModeReplyHandler {
&mut self,
reply: &GenericMessage<ModeReply>,
active_request: &ActivePusRequestStd,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<bool, Self::Error> {
@ -117,7 +117,7 @@ impl PusReplyHandler<ActivePusRequestStd, ModeReply> for ModeReplyHandler {
fn handle_request_timeout(
&mut self,
active_request: &ActivePusRequestStd,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(), Self::Error> {
@ -142,7 +142,7 @@ impl PusTcToRequestConverter<ActivePusRequestStd, ModeRequest> for ModeRequestCo
&mut self,
token: VerificationToken<TcStateAccepted>,
tc: &PusTcReader,
tm_sender: &(impl EcssTmSenderCore + ?Sized),
tm_sender: &(impl EcssTmSender + ?Sized),
verif_reporter: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(ActivePusRequestStd, ModeRequest), Self::Error> {
@ -253,7 +253,7 @@ pub fn create_mode_service_dynamic(
}
}
pub struct ModeServiceWrapper<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter> {
pub struct ModeServiceWrapper<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter> {
pub(crate) service: PusTargetedRequestService<
MpscTcReceiver,
TmSender,
@ -268,7 +268,7 @@ pub struct ModeServiceWrapper<TmSender: EcssTmSenderCore, TcInMemConverter: Ecss
>,
}
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter> TargetedPusService
impl<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter> TargetedPusService
for ModeServiceWrapper<TmSender, TcInMemConverter>
{
/// Returns [true] if the packet handling is finished.

18
satrs-example/src/pus/scheduler.rs
View File

@ -9,10 +9,10 @@ use satrs::pus::scheduler_srv::PusSchedServiceHandler;
use satrs::pus::verification::VerificationReporter;
use satrs::pus::{
EcssTcAndToken, EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcInVecConverter,
EcssTmSenderCore, MpscTcReceiver, MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded,
EcssTmSender, MpscTcReceiver, MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded,
PusPacketHandlerResult, PusServiceHelper, PusTmAsVec, PusTmInPool, TmInSharedPoolSender,
};
use satrs::tmtc::tc_helper::TcSenderSharedPool;
use satrs::tmtc::tc_helper::{PacketSenderSharedPool, SharedPacketPool};
use satrs_example::config::components::PUS_SCHED_SERVICE;
use super::HandlingStatus;
@ -21,7 +21,7 @@ pub trait TcReleaser {
fn release(&mut self, enabled: bool, info: &TcInfo, tc: &[u8]) -> bool;
}
impl TcReleaser for TcSenderSharedPool {
impl TcReleaser for PacketSenderSharedPool<SharedPacketPool> {
fn release(&mut self, enabled: bool, _info: &TcInfo, tc: &[u8]) -> bool {
if enabled {
let shared_pool = self.shared_pool.get_mut();
@ -51,10 +51,8 @@ impl TcReleaser for mpsc::Sender<Vec<u8>> {
}
}
pub struct SchedulingServiceWrapper<
TmSender: EcssTmSenderCore,
TcInMemConverter: EcssTcInMemConverter,
> {
pub struct SchedulingServiceWrapper<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter>
{
pub pus_11_handler: PusSchedServiceHandler<
MpscTcReceiver,
TmSender,
@ -67,7 +65,7 @@ pub struct SchedulingServiceWrapper<
pub tc_releaser: Box<dyn TcReleaser + Send>,
}
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
impl<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter>
SchedulingServiceWrapper<TmSender, TcInMemConverter>
{
pub fn release_tcs(&mut self) {
@ -121,7 +119,7 @@ impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
pub fn create_scheduler_service_static(
tm_sender: TmInSharedPoolSender<mpsc::SyncSender<PusTmInPool>>,
tc_releaser: TcSenderSharedPool,
tc_releaser: PacketSenderSharedPool<SharedPacketPool>,
pus_sched_rx: mpsc::Receiver<EcssTcAndToken>,
sched_tc_pool: StaticMemoryPool,
) -> SchedulingServiceWrapper<MpscTmInSharedPoolSenderBounded, EcssTcInSharedStoreConverter> {
@ -133,7 +131,7 @@ pub fn create_scheduler_service_static(
pus_sched_rx,
tm_sender,
create_verification_reporter(PUS_SCHED_SERVICE.id(), PUS_SCHED_SERVICE.apid),
EcssTcInSharedStoreConverter::new(tc_releaser.shared_pool(), 2048),
EcssTcInSharedStoreConverter::new(tc_releaser.shared_packet_store().0.clone(), 2048),
),
scheduler,
);

6
satrs-example/src/pus/stack.rs
View File

@ -1,7 +1,7 @@
use crate::pus::mode::ModeServiceWrapper;
use derive_new::new;
use satrs::{
pus::{EcssTcInMemConverter, EcssTmSenderCore},
pus::{EcssTcInMemConverter, EcssTmSender},
spacepackets::time::{cds, TimeWriter},
};
@ -12,7 +12,7 @@ use super::{
};
#[derive(new)]
pub struct PusStack<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter> {
pub struct PusStack<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter> {
test_srv: TestCustomServiceWrapper<TmSender, TcInMemConverter>,
hk_srv_wrapper: HkServiceWrapper<TmSender, TcInMemConverter>,
event_srv: EventServiceWrapper<TmSender, TcInMemConverter>,
@ -21,7 +21,7 @@ pub struct PusStack<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemCon
mode_srv: ModeServiceWrapper<TmSender, TcInMemConverter>,
}
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
impl<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter>
PusStack<TmSender, TcInMemConverter>
{
pub fn periodic_operation(&mut self) {

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

@ -6,7 +6,7 @@ use satrs::pus::test::PusService17TestHandler;
use satrs::pus::verification::{FailParams, VerificationReporter, VerificationReportingProvider};
use satrs::pus::EcssTcInSharedStoreConverter;
use satrs::pus::{
EcssTcAndToken, EcssTcInMemConverter, EcssTcInVecConverter, EcssTmSenderCore, MpscTcReceiver,
EcssTcAndToken, EcssTcInMemConverter, EcssTcInVecConverter, EcssTmSender, MpscTcReceiver,
MpscTmAsVecSender, MpscTmInSharedPoolSenderBounded, PusPacketHandlerResult, PusServiceHelper,
PusTmAsVec, PusTmInPool, TmInSharedPoolSender,
};
@ -57,16 +57,14 @@ pub fn create_test_service_dynamic(
}
}
pub struct TestCustomServiceWrapper<
TmSender: EcssTmSenderCore,
TcInMemConverter: EcssTcInMemConverter,
> {
pub struct TestCustomServiceWrapper<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter>
{
pub handler:
PusService17TestHandler<MpscTcReceiver, TmSender, TcInMemConverter, VerificationReporter>,
pub test_srv_event_sender: mpsc::Sender<EventMessageU32>,
}
impl<TmSender: EcssTmSenderCore, TcInMemConverter: EcssTcInMemConverter>
impl<TmSender: EcssTmSender, TcInMemConverter: EcssTcInMemConverter>
TestCustomServiceWrapper<TmSender, TcInMemConverter>
{
pub fn poll_and_handle_next_packet(&mut self, time_stamp: &[u8]) -> HandlingStatus {

4
satrs-example/src/requests.rs
View File

@ -8,7 +8,7 @@ use satrs::mode::ModeRequest;
use satrs::pus::verification::{
FailParams, TcStateAccepted, VerificationReportingProvider, VerificationToken,
};
use satrs::pus::{ActiveRequestProvider, EcssTmSenderCore, GenericRoutingError, PusRequestRouter};
use satrs::pus::{ActiveRequestProvider, EcssTmSender, GenericRoutingError, PusRequestRouter};
use satrs::queue::GenericSendError;
use satrs::request::{GenericMessage, MessageMetadata, UniqueApidTargetId};
use satrs::spacepackets::ecss::tc::PusTcReader;
@ -47,7 +47,7 @@ impl GenericRequestRouter {
active_request: &impl ActiveRequestProvider,
tc: &PusTcReader,
error: GenericRoutingError,
tm_sender: &(impl EcssTmSenderCore + ?Sized),
tm_sender: &(impl EcssTmSender + ?Sized),
verif_reporter: &impl VerificationReportingProvider,
time_stamp: &[u8],
) {

6
satrs-example/src/tmtc/tc_source.rs
View File

@ -1,4 +1,4 @@
use satrs::{pool::PoolProvider, tmtc::tc_helper::SharedTcPool};
use satrs::{pool::PoolProvider, tmtc::tc_helper::SharedPacketPool};
use std::sync::mpsc::{self, TryRecvError};
use satrs::{
@ -11,7 +11,7 @@ use crate::pus::PusTcDistributor;
// TC source components where static pools are the backing memory of the received telecommands.
pub struct TcSourceTaskStatic {
shared_tc_pool: SharedTcPool,
shared_tc_pool: SharedPacketPool,
tc_receiver: mpsc::Receiver<StoreAddr>,
tc_buf: [u8; 4096],
pus_receiver: PusTcDistributor<MpscTmInSharedPoolSenderBounded>,
@ -19,7 +19,7 @@ pub struct TcSourceTaskStatic {
impl TcSourceTaskStatic {
pub fn new(
shared_tc_pool: SharedTcPool,
shared_tc_pool: SharedPacketPool,
tc_receiver: mpsc::Receiver<StoreAddr>,
pus_receiver: PusTcDistributor<MpscTmInSharedPoolSenderBounded>,
) -> Self {

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

@ -132,7 +132,7 @@ impl EventReportCreator {
#[cfg(feature = "alloc")]
mod alloc_mod {
use super::*;
use crate::pus::{EcssTmSenderCore, EcssTmtcError};
use crate::pus::{EcssTmSender, EcssTmtcError};
use crate::ComponentId;
use alloc::vec;
use alloc::vec::Vec;
@ -194,7 +194,7 @@ mod alloc_mod {
pub fn event_info(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event_id: impl EcssEnumeration,
params: Option<&[u8]>,
@ -211,7 +211,7 @@ mod alloc_mod {
pub fn event_low_severity(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event_id: impl EcssEnumeration,
params: Option<&[u8]>,
@ -228,7 +228,7 @@ mod alloc_mod {
pub fn event_medium_severity(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event_id: impl EcssEnumeration,
params: Option<&[u8]>,
@ -245,7 +245,7 @@ mod alloc_mod {
pub fn event_high_severity(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event_id: impl EcssEnumeration,
params: Option<&[u8]>,
@ -268,7 +268,7 @@ mod tests {
use crate::events::{EventU32, Severity};
use crate::pus::test_util::TEST_COMPONENT_ID_0;
use crate::pus::tests::CommonTmInfo;
use crate::pus::{ChannelWithId, EcssTmSenderCore, EcssTmtcError, PusTmVariant};
use crate::pus::{ChannelWithId, EcssTmSender, EcssTmtcError, PusTmVariant};
use crate::ComponentId;
use spacepackets::ecss::PusError;
use spacepackets::ByteConversionError;
@ -301,7 +301,7 @@ mod tests {
}
}
impl EcssTmSenderCore for TestSender {
impl EcssTmSender for TestSender {
fn send_tm(&self, sender_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError> {
match tm {
PusTmVariant::InStore(_) => {

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

@ -10,7 +10,7 @@ use hashbrown::HashSet;
pub use crate::pus::event::EventReporter;
use crate::pus::verification::TcStateToken;
#[cfg(feature = "alloc")]
use crate::pus::EcssTmSenderCore;
use crate::pus::EcssTmSender;
use crate::pus::EcssTmtcError;
#[cfg(feature = "alloc")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "alloc")))]
@ -178,7 +178,7 @@ pub mod alloc_mod {
pub fn generate_pus_event_tm_generic(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event: Event,
params: Option<&[u8]>,
@ -240,7 +240,7 @@ pub mod alloc_mod {
pub fn generate_pus_event_tm<Severity: HasSeverity>(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event: EventU32TypedSev<Severity>,
aux_data: Option<&[u8]>,

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

@ -9,13 +9,13 @@ use std::sync::mpsc::Sender;
use super::verification::VerificationReportingProvider;
use super::{
EcssTcInMemConverter, EcssTcReceiverCore, EcssTmSenderCore, GenericConversionError,
EcssTcInMemConverter, EcssTcReceiverCore, EcssTmSender, GenericConversionError,
GenericRoutingError, PusServiceHelper,
};
pub struct PusEventServiceHandler<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
> {
@ -26,7 +26,7 @@ pub struct PusEventServiceHandler<
impl<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
> PusEventServiceHandler<TcReceiver, TmSender, TcInMemConverter, VerificationReporter>

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

@ -153,7 +153,7 @@ pub trait ChannelWithId: 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: Send {
pub trait EcssTmSender: Send {
fn send_tm(&self, source_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError>;
}
@ -161,7 +161,7 @@ pub trait EcssTmSenderCore: Send {
///
/// 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 {
pub trait EcssTcSender {
fn send_tc(&self, tc: PusTcCreator, token: Option<TcStateToken>) -> Result<(), EcssTmtcError>;
}
@ -169,7 +169,7 @@ pub trait EcssTcSenderCore {
#[derive(Default)]
pub struct EcssTmDummySender {}
impl EcssTmSenderCore for EcssTmDummySender {
impl EcssTmSender for EcssTmDummySender {
fn send_tm(&self, _source_id: ComponentId, _tm: PusTmVariant) -> Result<(), EcssTmtcError> {
Ok(())
}
@ -322,7 +322,7 @@ pub trait PusReplyHandler<ActiveRequestInfo: ActiveRequestProvider, ReplyType> {
&mut self,
reply: &GenericMessage<ReplyType>,
active_request: &ActiveRequestInfo,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<bool, Self::Error>;
@ -330,14 +330,14 @@ pub trait PusReplyHandler<ActiveRequestInfo: ActiveRequestProvider, ReplyType> {
fn handle_unrequested_reply(
&mut self,
reply: &GenericMessage<ReplyType>,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
) -> Result<(), Self::Error>;
/// Handle the timeout of an active request.
fn handle_request_timeout(
&mut self,
active_request: &ActiveRequestInfo,
tm_sender: &impl EcssTmSenderCore,
tm_sender: &impl EcssTmSender,
verification_handler: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(), Self::Error>;
@ -361,34 +361,34 @@ pub mod alloc_mod {
/// [Clone].
#[cfg(feature = "alloc")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "alloc")))]
pub trait EcssTmSender: EcssTmSenderCore + Downcast + DynClone {
pub trait EcssTmSenderExt: EcssTmSender + Downcast + DynClone {
// Remove this once trait upcasting coercion has been implemented.
// Tracking issue: https://github.com/rust-lang/rust/issues/65991
fn upcast(&self) -> &dyn EcssTmSenderCore;
fn upcast(&self) -> &dyn EcssTmSender;
// Remove this once trait upcasting coercion has been implemented.
// Tracking issue: https://github.com/rust-lang/rust/issues/65991
fn upcast_mut(&mut self) -> &mut dyn EcssTmSenderCore;
fn upcast_mut(&mut self) -> &mut dyn EcssTmSender;
}
/// Blanket implementation for all types which implement [EcssTmSenderCore] and are clonable.
impl<T> EcssTmSender for T
impl<T> EcssTmSenderExt for T
where
T: EcssTmSenderCore + Clone + 'static,
T: EcssTmSender + Clone + 'static,
{
// Remove this once trait upcasting coercion has been implemented.
// Tracking issue: https://github.com/rust-lang/rust/issues/65991
fn upcast(&self) -> &dyn EcssTmSenderCore {
fn upcast(&self) -> &dyn EcssTmSender {
self
}
// Remove this once trait upcasting coercion has been implemented.
// Tracking issue: https://github.com/rust-lang/rust/issues/65991
fn upcast_mut(&mut self) -> &mut dyn EcssTmSenderCore {
fn upcast_mut(&mut self) -> &mut dyn EcssTmSender {
self
}
}
dyn_clone::clone_trait_object!(EcssTmSender);
impl_downcast!(EcssTmSender);
dyn_clone::clone_trait_object!(EcssTmSenderExt);
impl_downcast!(EcssTmSenderExt);
/// Extension trait for [EcssTcSenderCore].
///
@ -402,13 +402,13 @@ pub mod alloc_mod {
/// [Clone].
#[cfg(feature = "alloc")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "alloc")))]
pub trait EcssTcSender: EcssTcSenderCore + Downcast + DynClone {}
pub trait EcssTcSenderExt: EcssTcSender + Downcast + DynClone {}
/// Blanket implementation for all types which implement [EcssTcSenderCore] and are clonable.
impl<T> EcssTcSender for T where T: EcssTcSenderCore + Clone + 'static {}
impl<T> EcssTcSenderExt for T where T: EcssTcSender + Clone + 'static {}
dyn_clone::clone_trait_object!(EcssTcSender);
impl_downcast!(EcssTcSender);
dyn_clone::clone_trait_object!(EcssTcSenderExt);
impl_downcast!(EcssTcSenderExt);
/// Extension trait for [EcssTcReceiverCore].
///
@ -451,7 +451,7 @@ pub mod alloc_mod {
&mut self,
token: VerificationToken<TcStateAccepted>,
tc: &PusTcReader,
tm_sender: &(impl EcssTmSenderCore + ?Sized),
tm_sender: &(impl EcssTmSender + ?Sized),
verif_reporter: &impl VerificationReportingProvider,
time_stamp: &[u8],
) -> Result<(ActiveRequestInfo, Request), Self::Error>;
@ -652,10 +652,10 @@ pub mod std_mod {
};
use crate::pus::verification::{TcStateAccepted, VerificationToken};
use crate::pus::{
EcssTcAndToken, EcssTcReceiverCore, EcssTmSenderCore, EcssTmtcError, GenericReceiveError,
EcssTcAndToken, EcssTcReceiverCore, EcssTmSender, EcssTmtcError, GenericReceiveError,
GenericSendError, PusTmVariant, TryRecvTmtcError,
};
use crate::tmtc::tm_helper::SharedTmPool;
use crate::tmtc::{PacketSenderSharedPool, PusTmPool, SharedPacketPool};
use crate::ComponentId;
use alloc::vec::Vec;
use core::time::Duration;
@ -687,7 +687,7 @@ pub mod std_mod {
}
}
impl EcssTmSenderCore for mpsc::Sender<PusTmInPool> {
impl EcssTmSender for mpsc::Sender<PusTmInPool> {
fn send_tm(&self, source_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError> {
match tm {
PusTmVariant::InStore(store_addr) => self
@ -702,7 +702,7 @@ pub mod std_mod {
}
}
impl EcssTmSenderCore for mpsc::SyncSender<PusTmInPool> {
impl EcssTmSender for mpsc::SyncSender<PusTmInPool> {
fn send_tm(&self, source_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError> {
match tm {
PusTmVariant::InStore(store_addr) => self
@ -725,7 +725,7 @@ pub mod std_mod {
pub type MpscTmAsVecSender = mpsc::Sender<PusTmAsVec>;
impl EcssTmSenderCore for MpscTmAsVecSender {
impl EcssTmSender for MpscTmAsVecSender {
fn send_tm(&self, source_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError> {
match tm {
PusTmVariant::InStore(addr) => return Err(EcssTmtcError::CantSendAddr(addr)),
@ -742,7 +742,7 @@ pub mod std_mod {
pub type MpscTmAsVecSenderBounded = mpsc::SyncSender<PusTmAsVec>;
impl EcssTmSenderCore for MpscTmAsVecSenderBounded {
impl EcssTmSender for MpscTmAsVecSenderBounded {
fn send_tm(&self, source_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError> {
match tm {
PusTmVariant::InStore(addr) => return Err(EcssTmtcError::CantSendAddr(addr)),
@ -757,9 +757,11 @@ pub mod std_mod {
}
}
// TODO: This is a duplication of an existing shared store packet sender. Try to remove it..
/*
#[derive(Clone)]
pub struct TmInSharedPoolSender<Sender: EcssTmSenderCore> {
shared_tm_store: SharedTmPool,
shared_tm_store: SharedPacketPool,
sender: Sender,
}
@ -769,7 +771,7 @@ pub mod std_mod {
source_id: ComponentId,
tm: PusTmCreator,
) -> Result<(), EcssTmtcError> {
let addr = self.shared_tm_store.add_pus_tm(&tm)?;
let addr = self.shared_tm_store.add_pus_tm_from_creator(&tm)?;
self.sender.send_tm(source_id, PusTmVariant::InStore(addr))
}
}
@ -784,7 +786,7 @@ pub mod std_mod {
}
impl<Sender: EcssTmSenderCore> TmInSharedPoolSender<Sender> {
pub fn new(shared_tm_store: SharedTmPool, sender: Sender) -> Self {
pub fn new(shared_tm_store: SharedPacketPool, sender: Sender) -> Self {
Self {
shared_tm_store,
sender,
@ -794,6 +796,7 @@ pub mod std_mod {
pub type MpscTmInSharedPoolSender = TmInSharedPoolSender<mpsc::Sender<PusTmInPool>>;
pub type MpscTmInSharedPoolSenderBounded = TmInSharedPoolSender<mpsc::SyncSender<PusTmInPool>>;
*/
pub type MpscTcReceiver = mpsc::Receiver<EcssTcAndToken>;
@ -832,7 +835,7 @@ pub mod std_mod {
}
}
impl EcssTmSenderCore for cb::Sender<PusTmInPool> {
impl EcssTmSender for cb::Sender<PusTmInPool> {
fn send_tm(
&self,
source_id: ComponentId,
@ -850,7 +853,7 @@ pub mod std_mod {
Ok(())
}
}
impl EcssTmSenderCore for cb::Sender<PusTmAsVec> {
impl EcssTmSender for cb::Sender<PusTmAsVec> {
fn send_tm(
&self,
source_id: ComponentId,
@ -1110,7 +1113,7 @@ pub mod std_mod {
pub struct PusServiceBase<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
VerificationReporter: VerificationReportingProvider,
> {
pub id: ComponentId,
@ -1130,7 +1133,7 @@ pub mod std_mod {
/// methods to make the generic parts of TC handling easier.
pub struct PusServiceHelper<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
> {
@ -1140,7 +1143,7 @@ pub mod std_mod {
impl<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
> PusServiceHelper<TcReceiver, TmSender, TcInMemConverter, VerificationReporter>
@ -1230,14 +1233,14 @@ pub mod std_mod {
pub type PusServiceHelperStaticWithMpsc<TcInMemConverter, VerificationReporter> =
PusServiceHelper<
MpscTcReceiver,
MpscTmInSharedPoolSender,
PacketSenderSharedPool,
TcInMemConverter,
VerificationReporter,
>;
pub type PusServiceHelperStaticWithBoundedMpsc<TcInMemConverter, VerificationReporter> =
PusServiceHelper<
MpscTcReceiver,
MpscTmInSharedPoolSenderBounded,
PacketSenderSharedPool,
TcInMemConverter,
VerificationReporter,
>;
@ -1307,7 +1310,7 @@ pub mod tests {
use crate::pool::{PoolProvider, SharedStaticMemoryPool, StaticMemoryPool, StaticPoolConfig};
use crate::pus::verification::{RequestId, VerificationReporter};
use crate::tmtc::tm_helper::SharedTmPool;
use crate::tmtc::SharedPacketPool;
use crate::ComponentId;
use super::test_util::{TEST_APID, TEST_COMPONENT_ID_0};
@ -1364,7 +1367,7 @@ pub mod tests {
pus_buf: RefCell<[u8; 2048]>,
tm_buf: [u8; 2048],
tc_pool: SharedStaticMemoryPool,
tm_pool: SharedTmPool,
tm_pool: SharedPacketPool,
tc_sender: mpsc::SyncSender<EcssTcAndToken>,
tm_receiver: mpsc::Receiver<PusTmInPool>,
}
@ -1386,7 +1389,7 @@ pub mod tests {
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 shared_tm_pool = SharedStaticMemoryPool::new(RwLock::new(tm_pool));
let (test_srv_tc_tx, test_srv_tc_rx) = mpsc::sync_channel(10);
let (tm_tx, tm_rx) = mpsc::sync_channel(10);

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

@ -2,11 +2,11 @@ use super::scheduler::PusSchedulerProvider;
use super::verification::{VerificationReporter, VerificationReportingProvider};
use super::{
EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcInVecConverter, EcssTcReceiverCore,
EcssTmSenderCore, MpscTcReceiver, MpscTmInSharedPoolSender, MpscTmInSharedPoolSenderBounded,
PusServiceHelper, PusTmAsVec,
EcssTmSender, MpscTcReceiver, PusServiceHelper, PusTmAsVec,
};
use crate::pool::PoolProvider;
use crate::pus::{PusPacketHandlerResult, PusPacketHandlingError};
use crate::tmtc::PacketSenderSharedPool;
use alloc::string::ToString;
use spacepackets::ecss::{scheduling, PusPacket};
use spacepackets::time::cds::CdsTime;
@ -22,7 +22,7 @@ use std::sync::mpsc;
/// telecommands when applicable.
pub struct PusSchedServiceHandler<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
PusScheduler: PusSchedulerProvider,
@ -34,7 +34,7 @@ pub struct PusSchedServiceHandler<
impl<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
Scheduler: PusSchedulerProvider,
@ -230,7 +230,7 @@ pub type PusService11SchedHandlerDynWithBoundedMpsc<PusScheduler> = PusSchedServ
/// mpsc queues.
pub type PusService11SchedHandlerStaticWithMpsc<PusScheduler> = PusSchedServiceHandler<
MpscTcReceiver,
MpscTmInSharedPoolSender,
PacketSenderSharedPool,
EcssTcInSharedStoreConverter,
VerificationReporter,
PusScheduler,
@ -239,7 +239,7 @@ pub type PusService11SchedHandlerStaticWithMpsc<PusScheduler> = PusSchedServiceH
/// mpsc queues.
pub type PusService11SchedHandlerStaticWithBoundedMpsc<PusScheduler> = PusSchedServiceHandler<
MpscTcReceiver,
MpscTmInSharedPoolSenderBounded,
PacketSenderSharedPool,
EcssTcInSharedStoreConverter,
VerificationReporter,
PusScheduler,
@ -249,7 +249,7 @@ pub type PusService11SchedHandlerStaticWithBoundedMpsc<PusScheduler> = PusSchedS
mod tests {
use crate::pool::{StaticMemoryPool, StaticPoolConfig};
use crate::pus::test_util::{PusTestHarness, TEST_APID};
use crate::pus::verification::{VerificationReporter, VerificationReportingProvider};
use crate::pus::verification::VerificationReporter;
use crate::pus::{
scheduler::{self, PusSchedulerProvider, TcInfo},
@ -257,10 +257,8 @@ mod tests {
verification::{RequestId, TcStateAccepted, VerificationToken},
EcssTcInSharedStoreConverter,
};
use crate::pus::{
MpscTcReceiver, MpscTmInSharedPoolSenderBounded, PusPacketHandlerResult,
PusPacketHandlingError,
};
use crate::pus::{MpscTcReceiver, PusPacketHandlerResult, PusPacketHandlingError};
use crate::tmtc::PacketSenderSharedPool;
use alloc::collections::VecDeque;
use delegate::delegate;
use spacepackets::ecss::scheduling::Subservice;
@ -279,7 +277,7 @@ mod tests {
common: PusServiceHandlerWithSharedStoreCommon,
handler: PusSchedServiceHandler<
MpscTcReceiver,
MpscTmInSharedPoolSenderBounded,
PacketSenderSharedPool,
EcssTcInSharedStoreConverter,
VerificationReporter,
TestScheduler,

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

@ -10,7 +10,7 @@ use std::sync::mpsc;
use super::verification::{VerificationReporter, VerificationReportingProvider};
use super::{
EcssTcInMemConverter, EcssTcInSharedStoreConverter, EcssTcInVecConverter, EcssTcReceiverCore,
EcssTmSenderCore, GenericConversionError, MpscTcReceiver, MpscTmInSharedPoolSender,
EcssTmSender, GenericConversionError, MpscTcReceiver, MpscTmInSharedPoolSender,
MpscTmInSharedPoolSenderBounded, PusServiceHelper,
};
@ -18,7 +18,7 @@ use super::{
/// This handler only processes ping requests and generates a ping reply for them accordingly.
pub struct PusService17TestHandler<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
> {
@ -28,7 +28,7 @@ pub struct PusService17TestHandler<
impl<
TcReceiver: EcssTcReceiverCore,
TmSender: EcssTmSenderCore,
TmSender: EcssTmSender,
TcInMemConverter: EcssTcInMemConverter,
VerificationReporter: VerificationReportingProvider,
> PusService17TestHandler<TcReceiver, TmSender, TcInMemConverter, VerificationReporter>

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

@ -81,7 +81,7 @@
//! The [integration test](https://egit.irs.uni-stuttgart.de/rust/fsrc-launchpad/src/branch/main/fsrc-core/tests/verification_test.rs)
//! for the verification module contains examples how this module could be used in a more complex
//! context involving multiple threads
use crate::pus::{source_buffer_large_enough, EcssTmSenderCore, EcssTmtcError};
use crate::pus::{source_buffer_large_enough, EcssTmSender, EcssTmtcError};
use core::fmt::{Debug, Display, Formatter};
use core::hash::{Hash, Hasher};
use core::marker::PhantomData;
@ -425,35 +425,35 @@ pub trait VerificationReportingProvider {
fn acceptance_success(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateNone>,
time_stamp: &[u8],
) -> Result<VerificationToken<TcStateAccepted>, EcssTmtcError>;
fn acceptance_failure(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateNone>,
params: FailParams,
) -> Result<(), EcssTmtcError>;
fn start_success(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateAccepted>,
time_stamp: &[u8],
) -> Result<VerificationToken<TcStateStarted>, EcssTmtcError>;
fn start_failure(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateAccepted>,
params: FailParams,
) -> Result<(), EcssTmtcError>;
fn step_success(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: &VerificationToken<TcStateStarted>,
time_stamp: &[u8],
step: impl EcssEnumeration,
@ -461,21 +461,21 @@ pub trait VerificationReportingProvider {
fn step_failure(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateStarted>,
params: FailParamsWithStep,
) -> Result<(), EcssTmtcError>;
fn completion_success<TcState: WasAtLeastAccepted + Copy>(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcState>,
time_stamp: &[u8],
) -> Result<(), EcssTmtcError>;
fn completion_failure<TcState: WasAtLeastAccepted + Copy>(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcState>,
params: FailParams,
) -> Result<(), EcssTmtcError>;
@ -1027,7 +1027,7 @@ pub mod alloc_mod {
/// Package and send a PUS TM\[1, 1\] packet, see 8.1.2.1 of the PUS standard
fn acceptance_success(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateNone>,
time_stamp: &[u8],
) -> Result<VerificationToken<TcStateAccepted>, EcssTmtcError> {
@ -1044,7 +1044,7 @@ pub mod alloc_mod {
/// Package and send a PUS TM\[1, 2\] packet, see 8.1.2.2 of the PUS standard
fn acceptance_failure(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateNone>,
params: FailParams,
) -> Result<(), EcssTmtcError> {
@ -1063,7 +1063,7 @@ pub mod alloc_mod {
/// Requires a token previously acquired by calling [Self::acceptance_success].
fn start_success(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateAccepted>,
time_stamp: &[u8],
) -> Result<VerificationToken<TcStateStarted>, EcssTmtcError> {
@ -1083,7 +1083,7 @@ pub mod alloc_mod {
/// the token because verification handling is done.
fn start_failure(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateAccepted>,
params: FailParams,
) -> Result<(), EcssTmtcError> {
@ -1102,7 +1102,7 @@ pub mod alloc_mod {
/// Requires a token previously acquired by calling [Self::start_success].
fn step_success(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: &VerificationToken<TcStateStarted>,
time_stamp: &[u8],
step: impl EcssEnumeration,
@ -1123,7 +1123,7 @@ pub mod alloc_mod {
/// token because verification handling is done.
fn step_failure(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateStarted>,
params: FailParamsWithStep,
) -> Result<(), EcssTmtcError> {
@ -1144,7 +1144,7 @@ pub mod alloc_mod {
fn completion_success<TcState: WasAtLeastAccepted + Copy>(
&self,
// sender_id: ComponentId,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcState>,
time_stamp: &[u8],
) -> Result<(), EcssTmtcError> {
@ -1164,7 +1164,7 @@ pub mod alloc_mod {
/// token because verification handling is done.
fn completion_failure<TcState: WasAtLeastAccepted + Copy>(
&self,
sender: &(impl EcssTmSenderCore + ?Sized),
sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcState>,
params: FailParams,
) -> Result<(), EcssTmtcError> {
@ -1269,7 +1269,7 @@ pub mod test_util {
fn acceptance_success(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateNone>,
time_stamp: &[u8],
) -> Result<VerificationToken<TcStateAccepted>, EcssTmtcError> {
@ -1288,7 +1288,7 @@ pub mod test_util {
fn acceptance_failure(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateNone>,
params: FailParams,
) -> Result<(), EcssTmtcError> {
@ -1306,7 +1306,7 @@ pub mod test_util {
fn start_success(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateAccepted>,
time_stamp: &[u8],
) -> Result<VerificationToken<TcStateStarted>, EcssTmtcError> {
@ -1325,7 +1325,7 @@ pub mod test_util {
fn start_failure(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<super::TcStateAccepted>,
params: FailParams,
) -> Result<(), EcssTmtcError> {
@ -1343,7 +1343,7 @@ pub mod test_util {
fn step_success(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: &VerificationToken<TcStateStarted>,
time_stamp: &[u8],
step: impl EcssEnumeration,
@ -1363,7 +1363,7 @@ pub mod test_util {
fn step_failure(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcStateStarted>,
params: FailParamsWithStep,
) -> Result<(), EcssTmtcError> {
@ -1381,7 +1381,7 @@ pub mod test_util {
fn completion_success<TcState: super::WasAtLeastAccepted + Copy>(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcState>,
time_stamp: &[u8],
) -> Result<(), EcssTmtcError> {
@ -1397,7 +1397,7 @@ pub mod test_util {
fn completion_failure<TcState: WasAtLeastAccepted + Copy>(
&self,
_sender: &(impl EcssTmSenderCore + ?Sized),
_sender: &(impl EcssTmSender + ?Sized),
token: VerificationToken<TcState>,
params: FailParams,
) -> Result<(), EcssTmtcError> {
@ -1640,7 +1640,7 @@ pub mod tests {
use crate::pus::test_util::{TEST_APID, TEST_COMPONENT_ID_0};
use crate::pus::tests::CommonTmInfo;
use crate::pus::verification::{
EcssTmSenderCore, EcssTmtcError, FailParams, FailParamsWithStep, RequestId, TcStateNone,
EcssTmSender, EcssTmtcError, FailParams, FailParamsWithStep, RequestId, TcStateNone,
VerificationReporter, VerificationReporterCfg, VerificationToken,
};
use crate::pus::{ChannelWithId, MpscTmInSharedPoolSender, PusTmVariant};
@ -1694,7 +1694,7 @@ pub mod tests {
}
}
impl EcssTmSenderCore for TestSender {
impl EcssTmSender for TestSender {
fn send_tm(&self, sender_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError> {
match tm {
PusTmVariant::InStore(_) => {

320
satrs/src/tmtc/mod.rs
View File

@ -7,16 +7,25 @@
//! all received telecommands are sent to a special handler object called TC source. Using
//! a design like this makes it simpler to add new TC packet sources or new telemetry generators:
//! They only need to send the received and generated data to these objects.
use crate::pool::{PoolProvider, SharedStaticMemoryPool, StoreAddr, StoreError};
#[cfg(feature = "std")]
use crate::queue::GenericSendError;
use core::cell::RefCell;
#[cfg(feature = "alloc")]
use downcast_rs::{impl_downcast, Downcast};
use spacepackets::SpHeader;
use spacepackets::{
ecss::{
tc::PusTcReader,
tm::{PusTmCreator, PusTmReader},
WritablePusPacket,
},
SpHeader,
};
#[cfg(feature = "std")]
use std::sync::mpsc;
#[cfg(feature = "std")]
pub mod tc_helper;
pub use std_mod::*;
pub mod tm_helper;
/// Generic trait for object which can receive any telecommands in form of a raw bytestream, with
@ -152,3 +161,308 @@ where
self
}
}
#[derive(Clone)]
pub struct SharedPacketPool(pub SharedStaticMemoryPool);
impl SharedPacketPool {
pub fn new(pool: &SharedStaticMemoryPool) -> Self {
Self(pool.clone())
}
}
pub trait CcsdsPacketPool {
fn add_ccsds_tc(&mut self, _: &SpHeader, tc_raw: &[u8]) -> Result<StoreAddr, StoreError> {
self.add_raw_tc(tc_raw)
}
fn add_raw_tc(&mut self, tc_raw: &[u8]) -> Result<StoreAddr, StoreError>;
}
pub trait PusTcPool {
fn add_pus_tc(&mut self, pus_tc: &PusTcReader) -> Result<StoreAddr, StoreError>;
}
pub trait PusTmPool {
fn add_pus_tm_from_reader(&mut self, pus_tm: &PusTmReader) -> Result<StoreAddr, StoreError>;
fn add_pus_tm_from_creator(&mut self, pus_tm: &PusTmCreator) -> Result<StoreAddr, StoreError>;
}
impl PusTcPool for SharedPacketPool {
fn add_pus_tc(&mut self, pus_tc: &PusTcReader) -> Result<StoreAddr, StoreError> {
let mut pg = self.0.write().map_err(|_| StoreError::LockError)?;
let addr = pg.free_element(pus_tc.len_packed(), |buf| {
buf[0..pus_tc.len_packed()].copy_from_slice(pus_tc.raw_data());
})?;
Ok(addr)
}
}
impl PusTmPool for SharedPacketPool {
fn add_pus_tm_from_reader(&mut self, pus_tm: &PusTmReader) -> Result<StoreAddr, StoreError> {
let mut pg = self.0.write().map_err(|_| StoreError::LockError)?;
let addr = pg.free_element(pus_tm.len_packed(), |buf| {
buf[0..pus_tm.len_packed()].copy_from_slice(pus_tm.raw_data());
})?;
Ok(addr)
}
fn add_pus_tm_from_creator(&mut self, pus_tm: &PusTmCreator) -> Result<StoreAddr, StoreError> {
let mut pg = self.0.write().map_err(|_| StoreError::LockError)?;
let mut result = Ok(0);
let addr = pg.free_element(pus_tm.len_written(), |buf| {
result = pus_tm.write_to_bytes(buf);
})?;
result?;
Ok(addr)
}
}
impl CcsdsPacketPool for SharedPacketPool {
fn add_raw_tc(&mut self, tc_raw: &[u8]) -> Result<StoreAddr, StoreError> {
let mut pg = self.0.write().map_err(|_| StoreError::LockError)?;
let addr = pg.free_element(tc_raw.len(), |buf| {
buf[0..tc_raw.len()].copy_from_slice(tc_raw);
})?;
Ok(addr)
}
}
#[cfg(feature = "std")]
pub mod std_mod {
use std::sync::mpsc;
use thiserror::Error;
use crate::pus::{EcssTmSender, ReceivesEcssPusTc};
use super::*;
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum StoreAndSendError {
#[error("Store error: {0}")]
Store(#[from] StoreError),
#[error("Genreric send error: {0}")]
Send(#[from] GenericSendError),
}
#[derive(Clone)]
pub struct PacketSenderSharedPool<PacketStore: CcsdsPacketPool = SharedPacketPool> {
pub tc_source: mpsc::SyncSender<StoreAddr>,
pub shared_pool: RefCell<PacketStore>,
}
impl<PacketStore: CcsdsPacketPool> PacketSenderSharedPool<PacketStore> {
pub fn new(tc_sender: mpsc::SyncSender<StoreAddr>, shared_pool: PacketStore) -> Self {
Self {
tc_source: tc_sender,
shared_pool: RefCell::new(shared_pool),
}
}
}
impl<PacketStore: CcsdsPacketPool + Clone> PacketSenderSharedPool<PacketStore> {
pub fn shared_packet_store(&self) -> PacketStore {
let pool = self.shared_pool.borrow();
pool.clone()
}
}
impl<PacketStore: CcsdsPacketPool + Send> PacketSenderRaw for PacketSenderSharedPool<PacketStore> {
type Error = StoreAndSendError;
fn send_raw_tc(&self, tc_raw: &[u8]) -> Result<(), Self::Error> {
let mut shared_pool = self.shared_pool.borrow_mut();
let addr = shared_pool.add_raw_tc(tc_raw)?;
drop(shared_pool);
self.tc_source.try_send(addr).map_err(|e| match e {
mpsc::TrySendError::Full(_) => GenericSendError::QueueFull(None),
mpsc::TrySendError::Disconnected(_) => GenericSendError::RxDisconnected,
})?;
Ok(())
}
}
impl<PacketStore: CcsdsPacketPool + PusTcPool + Send> ReceivesEcssPusTc
for PacketSenderSharedPool<PacketStore>
{
type Error = StoreAndSendError;
fn pass_pus_tc(&mut self, _: &SpHeader, pus_tc: &PusTcReader) -> Result<(), Self::Error> {
let mut shared_pool = self.shared_pool.borrow_mut();
let addr = shared_pool.add_raw_tc(pus_tc.raw_data())?;
drop(shared_pool);
self.tc_source.try_send(addr).map_err(|e| match e {
mpsc::TrySendError::Full(_) => GenericSendError::QueueFull(None),
mpsc::TrySendError::Disconnected(_) => GenericSendError::RxDisconnected,
})?;
Ok(())
}
}
impl<PacketStore: CcsdsPacketPool + Send> ReceivesCcsdsTc for PacketSenderSharedPool<PacketStore> {
type Error = StoreAndSendError;
fn pass_ccsds(&mut self, _sp_header: &SpHeader, tc_raw: &[u8]) -> Result<(), Self::Error> {
self.send_raw_tc(tc_raw)
}
}
impl<PacketStore: CcsdsPacketPool + PusTmPool + Send> EcssTmSender
for PacketSenderSharedPool<PacketStore>
{
fn send_tm(
&self,
source_id: crate::ComponentId,
tm: crate::pus::PusTmVariant,
) -> Result<(), crate::pus::EcssTmtcError> {
todo!()
}
}
}
#[cfg(test)]
pub(crate) mod tests {
use alloc::vec;
use std::sync::RwLock;
use crate::pool::{PoolProviderWithGuards, StaticMemoryPool, StaticPoolConfig};
use super::*;
use std::sync::mpsc;
pub(crate) fn send_with_sender<SendError>(
packet_sender: &(impl PacketSenderRaw<Error = SendError> + ?Sized),
packet: &[u8],
) -> Result<(), SendError> {
packet_sender.send_raw_tc(packet)
}
#[test]
fn test_basic_mpsc_channel_sender() {
let (tx, rx) = mpsc::channel();
let some_packet = vec![1, 2, 3, 4, 5];
send_with_sender(&tx, &some_packet).expect("failed to send packet");
let rx_packet = rx.try_recv().unwrap();
assert_eq!(some_packet, rx_packet);
}
#[test]
fn test_basic_mpsc_channel_receiver_dropped() {
let (tx, rx) = mpsc::channel();
let some_packet = vec![1, 2, 3, 4, 5];
drop(rx);
let result = send_with_sender(&tx, &some_packet);
assert!(result.is_err());
matches!(result.unwrap_err(), GenericSendError::RxDisconnected);
}
#[test]
fn test_basic_mpsc_sync_sender() {
let (tx, rx) = mpsc::sync_channel(3);
let some_packet = vec![1, 2, 3, 4, 5];
send_with_sender(&tx, &some_packet).expect("failed to send packet");
let rx_packet = rx.try_recv().unwrap();
assert_eq!(some_packet, rx_packet);
}
#[test]
fn test_basic_mpsc_sync_sender_receiver_dropped() {
let (tx, rx) = mpsc::sync_channel(3);
let some_packet = vec![1, 2, 3, 4, 5];
drop(rx);
let result = send_with_sender(&tx, &some_packet);
assert!(result.is_err());
matches!(result.unwrap_err(), GenericSendError::RxDisconnected);
}
#[test]
fn test_basic_mpsc_sync_sender_queue_full() {
let (tx, rx) = mpsc::sync_channel(1);
let some_packet = vec![1, 2, 3, 4, 5];
send_with_sender(&tx, &some_packet).expect("failed to send packet");
let result = send_with_sender(&tx, &some_packet);
assert!(result.is_err());
matches!(result.unwrap_err(), GenericSendError::QueueFull(None));
let rx_packet = rx.try_recv().unwrap();
assert_eq!(some_packet, rx_packet);
}
#[test]
fn test_basic_shared_store_sender() {
let (tc_tx, tc_rx) = mpsc::sync_channel(10);
let pool_cfg = StaticPoolConfig::new(vec![(2, 8)], true);
let shared_pool = SharedPacketPool::new(&SharedStaticMemoryPool::new(RwLock::new(
StaticMemoryPool::new(pool_cfg),
)));
let some_packet = vec![1, 2, 3, 4, 5];
let tc_sender = PacketSenderSharedPool::new(tc_tx, shared_pool.clone());
send_with_sender(&tc_sender, &some_packet).expect("failed to send packet");
let tc_tx_addr = tc_rx.try_recv().unwrap();
let mut pool = shared_pool.0.write().unwrap();
let read_guard = pool.read_with_guard(tc_tx_addr);
assert_eq!(read_guard.read_as_vec().unwrap(), some_packet);
}
#[test]
fn test_basic_shared_store_sender_rx_dropped() {
let (tc_tx, tc_rx) = mpsc::sync_channel(10);
let pool_cfg = StaticPoolConfig::new(vec![(2, 8)], true);
let shared_pool = SharedPacketPool::new(&SharedStaticMemoryPool::new(RwLock::new(
StaticMemoryPool::new(pool_cfg),
)));
let some_packet = vec![1, 2, 3, 4, 5];
drop(tc_rx);
let tc_sender = PacketSenderSharedPool::new(tc_tx, shared_pool.clone());
let result = send_with_sender(&tc_sender, &some_packet);
assert!(result.is_err());
matches!(
result.unwrap_err(),
StoreAndSendError::Send(GenericSendError::RxDisconnected)
);
}
#[test]
fn test_basic_shared_store_sender_queue_full() {
let (tc_tx, tc_rx) = mpsc::sync_channel(1);
let pool_cfg = StaticPoolConfig::new(vec![(2, 8)], true);
let shared_pool = SharedPacketPool::new(&SharedStaticMemoryPool::new(RwLock::new(
StaticMemoryPool::new(pool_cfg),
)));
let some_packet = vec![1, 2, 3, 4, 5];
let tc_sender = PacketSenderSharedPool::new(tc_tx, shared_pool.clone());
send_with_sender(&tc_sender, &some_packet).expect("failed to send packet");
let result = send_with_sender(&tc_sender, &some_packet);
assert!(result.is_err());
matches!(
result.unwrap_err(),
StoreAndSendError::Send(GenericSendError::RxDisconnected)
);
let tc_tx_addr = tc_rx.try_recv().unwrap();
let mut pool = shared_pool.0.write().unwrap();
let read_guard = pool.read_with_guard(tc_tx_addr);
assert_eq!(read_guard.read_as_vec().unwrap(), some_packet);
}
#[test]
fn test_basic_shared_store_store_error() {
let (tc_tx, tc_rx) = mpsc::sync_channel(1);
let pool_cfg = StaticPoolConfig::new(vec![(1, 8)], true);
let shared_pool = SharedPacketPool::new(&SharedStaticMemoryPool::new(RwLock::new(
StaticMemoryPool::new(pool_cfg),
)));
let some_packet = vec![1, 2, 3, 4, 5];
let tc_sender = PacketSenderSharedPool::new(tc_tx, shared_pool.clone());
send_with_sender(&tc_sender, &some_packet).expect("failed to send packet");
let result = send_with_sender(&tc_sender, &some_packet);
assert!(result.is_err());
matches!(
result.unwrap_err(),
StoreAndSendError::Store(StoreError::StoreFull(..))
);
let tc_tx_addr = tc_rx.try_recv().unwrap();
let mut pool = shared_pool.0.write().unwrap();
let read_guard = pool.read_with_guard(tc_tx_addr);
assert_eq!(read_guard.read_as_vec().unwrap(), some_packet);
}
}

110
satrs/src/tmtc/tc_helper.rs
View File

@ -1,110 +0,0 @@
use core::cell::RefCell;
use std::sync::mpsc;
use spacepackets::{ecss::tc::PusTcReader, SpHeader};
use thiserror::Error;
use crate::{
pool::{PoolProvider, SharedStaticMemoryPool, StoreAddr, StoreError},
pus::ReceivesEcssPusTc,
queue::GenericSendError,
};
use super::{PacketSenderRaw, ReceivesCcsdsTc};
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum StoreAndSendError {
#[error("Store error: {0}")]
Store(#[from] StoreError),
#[error("Genreric send error: {0}")]
Send(#[from] GenericSendError),
}
#[derive(Clone)]
pub struct SharedTcPool(pub SharedStaticMemoryPool);
impl SharedTcPool {
pub fn add_pus_tc(&mut self, pus_tc: &PusTcReader) -> Result<StoreAddr, StoreError> {
let mut pg = self.0.write().expect("error locking TC store");
let addr = pg.free_element(pus_tc.len_packed(), |buf| {
buf[0..pus_tc.len_packed()].copy_from_slice(pus_tc.raw_data());
})?;
Ok(addr)
}
pub fn add_ccsds_tc(&mut self, _: &SpHeader, tc_raw: &[u8]) -> Result<StoreAddr, StoreError> {
self.add_raw_tc(tc_raw)
}
pub fn add_raw_tc(&mut self, tc_raw: &[u8]) -> Result<StoreAddr, StoreError> {
let mut pg = self.0.write().expect("error locking TC store");
let addr = pg.free_element(tc_raw.len(), |buf| {
buf[0..tc_raw.len()].copy_from_slice(tc_raw);
})?;
Ok(addr)
}
}
#[derive(Clone)]
pub struct TcSenderSharedPool {
pub tc_source: mpsc::SyncSender<StoreAddr>,
pub shared_pool: RefCell<SharedTcPool>,
}
impl TcSenderSharedPool {
pub fn new(tc_source: mpsc::SyncSender<StoreAddr>, shared_pool: SharedTcPool) -> Self {
Self {
tc_source,
shared_pool: RefCell::new(shared_pool),
}
}
#[allow(dead_code)]
pub fn shared_pool(&self) -> SharedStaticMemoryPool {
let pool = self.shared_pool.borrow();
pool.0.clone()
}
}
impl PacketSenderRaw for TcSenderSharedPool {
type Error = StoreAndSendError;
fn send_raw_tc(&self, tc_raw: &[u8]) -> Result<(), Self::Error> {
let mut shared_pool = self.shared_pool.borrow_mut();
let addr = shared_pool.add_raw_tc(tc_raw)?;
drop(shared_pool);
self.tc_source.try_send(addr).map_err(|e| match e {
mpsc::TrySendError::Full(_) => GenericSendError::QueueFull(None),
mpsc::TrySendError::Disconnected(_) => GenericSendError::RxDisconnected,
})?;
Ok(())
}
}
impl ReceivesEcssPusTc for TcSenderSharedPool {
type Error = StoreAndSendError;
fn pass_pus_tc(&mut self, _: &SpHeader, pus_tc: &PusTcReader) -> Result<(), Self::Error> {
let mut shared_pool = self.shared_pool.borrow_mut();
let addr = shared_pool.add_raw_tc(pus_tc.raw_data())?;
drop(shared_pool);
self.tc_source.try_send(addr).map_err(|e| match e {
mpsc::TrySendError::Full(_) => GenericSendError::QueueFull(None),
mpsc::TrySendError::Disconnected(_) => GenericSendError::RxDisconnected,
})?;
Ok(())
}
}
impl ReceivesCcsdsTc for TcSenderSharedPool {
type Error = StoreAndSendError;
fn pass_ccsds(&mut self, _sp_header: &SpHeader, tc_raw: &[u8]) -> Result<(), Self::Error> {
self.send_raw_tc(tc_raw)
}
}
#[cfg(test)]
mod tests {
// TODO: Add tests for shared pool TC sender component.
}

44
satrs/src/tmtc/tm_helper.rs
View File

@ -3,50 +3,6 @@ use spacepackets::time::cds::CdsTime;
use spacepackets::time::TimeWriter;
use spacepackets::SpHeader;
#[cfg(feature = "std")]
pub use std_mod::*;
#[cfg(feature = "std")]
pub mod std_mod {
use crate::pool::{
PoolProvider, SharedStaticMemoryPool, StaticMemoryPool, StoreAddr, StoreError,
};
use crate::pus::EcssTmtcError;
use spacepackets::ecss::tm::PusTmCreator;
use spacepackets::ecss::WritablePusPacket;
use std::sync::{Arc, RwLock};
#[derive(Clone)]
pub struct SharedTmPool(pub SharedStaticMemoryPool);
impl SharedTmPool {
pub fn new(shared_pool: StaticMemoryPool) -> Self {
Self(Arc::new(RwLock::new(shared_pool)))
}
pub fn clone_backing_pool(&self) -> SharedStaticMemoryPool {
self.0.clone()
}
pub fn shared_pool(&self) -> &SharedStaticMemoryPool {
&self.0
}
pub fn shared_pool_mut(&mut self) -> &mut SharedStaticMemoryPool {
&mut self.0
}
pub fn add_pus_tm(&self, pus_tm: &PusTmCreator) -> Result<StoreAddr, EcssTmtcError> {
let mut pg = self.0.write().map_err(|_| StoreError::LockError)?;
let addr = pg.free_element(pus_tm.len_written(), |buf| {
pus_tm
.write_to_bytes(buf)
.expect("writing PUS TM to store failed");
})?;
Ok(addr)
}
}
}
pub struct PusTmWithCdsShortHelper {
apid: u16,
cds_short_buf: [u8; 7],