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

continue source handler
All checks were successful
Rust/sat-rs/pipeline/head This commit looks good
Details

Browse Source
This commit is contained in:
Robin Müller 2024-06-04 10:01:26 +02:00
parent 1f2d6c9474
commit 7de4ef9408
2 changed files with 221 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

274
satrs/src/cfdp/dest.rs
View File

@ -3,13 +3,13 @@ use core::str::{from_utf8, Utf8Error};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use super::{ use super::{
filestore::{FilestoreError, VirtualFilestore}, filestore::{FilestoreError, NativeFilestore, VirtualFilestore},
user::{CfdpUser, FileSegmentRecvdParams, MetadataReceivedParams}, user::{CfdpUser, FileSegmentRecvdParams, MetadataReceivedParams},
CheckTimerCreator, CountdownProvider, EntityType, LocalEntityConfig, PacketInfo, PacketTarget, CheckTimerProviderCreator, CountdownProvider, EntityType, LocalEntityConfig, PacketInfo,
RemoteEntityConfig, RemoteEntityConfigProvider, State, TimerContext, TransactionId, PacketTarget, RemoteEntityConfig, RemoteEntityConfigProvider, State, StdCheckTimer,
StdCheckTimerCreator, StdRemoteEntityConfigProvider, TimerContext, TransactionId,
TransactionStep, TransactionStep,
}; };
use alloc::boxed::Box;
use smallvec::SmallVec; use smallvec::SmallVec;
use spacepackets::{ use spacepackets::{
cfdp::{ cfdp::{
@ -43,7 +43,7 @@ enum CompletionDisposition {
} }
#[derive(Debug)] #[derive(Debug)]
struct TransferState { struct TransferState<CheckTimer: CountdownProvider> {
transaction_id: Option<TransactionId>, transaction_id: Option<TransactionId>,
metadata_params: MetadataGenericParams, metadata_params: MetadataGenericParams,
progress: u64, progress: u64,
@ -54,10 +54,10 @@ struct TransferState {
completion_disposition: CompletionDisposition, completion_disposition: CompletionDisposition,
checksum: u32, checksum: u32,
current_check_count: u32, current_check_count: u32,
current_check_timer: Option<Box<dyn CountdownProvider>>, current_check_timer: Option<CheckTimer>,
} }
impl Default for TransferState { impl<CheckTimer: CountdownProvider> Default for TransferState<CheckTimer> {
fn default() -> Self { fn default() -> Self {
Self { Self {
transaction_id: None, transaction_id: None,
@ -76,8 +76,8 @@ impl Default for TransferState {
} }
#[derive(Debug)] #[derive(Debug)]
struct TransactionParams { struct TransactionParams<CheckTimer: CountdownProvider> {
tstate: TransferState, tstate: TransferState<CheckTimer>,
pdu_conf: CommonPduConfig, pdu_conf: CommonPduConfig,
file_properties: FileProperties, file_properties: FileProperties,
cksum_buf: [u8; 1024], cksum_buf: [u8; 1024],
@ -86,7 +86,7 @@ struct TransactionParams {
remote_cfg: Option<RemoteEntityConfig>, remote_cfg: Option<RemoteEntityConfig>,
} }
impl TransactionParams { impl<CheckTimer: CountdownProvider> TransactionParams<CheckTimer> {
fn transmission_mode(&self) -> TransmissionMode { fn transmission_mode(&self) -> TransmissionMode {
self.pdu_conf.trans_mode self.pdu_conf.trans_mode
} }
@ -104,7 +104,7 @@ impl Default for FileProperties {
} }
} }
impl TransactionParams { impl<CheckTimer: CountdownProvider> TransactionParams<CheckTimer> {
fn file_size(&self) -> u64 { fn file_size(&self) -> u64 {
self.tstate.metadata_params.file_size self.tstate.metadata_params.file_size
} }
@ -114,7 +114,7 @@ impl TransactionParams {
} }
} }
impl Default for TransactionParams { impl<CheckTimer: CountdownProvider> Default for TransactionParams<CheckTimer> {
fn default() -> Self { fn default() -> Self {
Self { Self {
pdu_conf: Default::default(), pdu_conf: Default::default(),
@ -128,7 +128,7 @@ impl Default for TransactionParams {
} }
} }
impl TransactionParams { impl<CheckTimer: CountdownProvider> TransactionParams<CheckTimer> {
fn reset(&mut self) { fn reset(&mut self) {
self.tstate.condition_code = ConditionCode::NoError; self.tstate.condition_code = ConditionCode::NoError;
self.tstate.delivery_code = DeliveryCode::Incomplete; self.tstate.delivery_code = DeliveryCode::Incomplete;
@ -170,7 +170,7 @@ pub enum DestError {
pub trait CfdpPacketSender: Send { pub trait CfdpPacketSender: Send {
fn send_pdu( fn send_pdu(
&mut self, &self,
pdu_type: PduType, pdu_type: PduType,
file_directive_type: Option<FileDirectiveType>, file_directive_type: Option<FileDirectiveType>,
raw_pdu: &[u8], raw_pdu: &[u8],
@ -191,19 +191,41 @@ pub trait CfdpPacketSender: Send {
/// All generated packets are sent via the [CfdpPacketSender] trait, which is implemented by the /// All generated packets are sent via the [CfdpPacketSender] trait, which is implemented by the
/// user and passed as a constructor parameter. The number of generated packets is returned /// user and passed as a constructor parameter. The number of generated packets is returned
/// by the state machine call. /// by the state machine call.
pub struct DestinationHandler { pub struct DestinationHandler<
PduSender: CfdpPacketSender,
Vfs: VirtualFilestore,
RemoteCfgTable: RemoteEntityConfigProvider,
CheckTimerCreator: CheckTimerProviderCreator<CheckTimer = CheckTimerProvider>,
CheckTimerProvider: CountdownProvider,
> {
local_cfg: LocalEntityConfig, local_cfg: LocalEntityConfig,
step: TransactionStep, step: TransactionStep,
state: State, state: State,
tparams: TransactionParams, tparams: TransactionParams<CheckTimerProvider>,
packet_buf: alloc::vec::Vec<u8>, packet_buf: alloc::vec::Vec<u8>,
packet_sender: Box<dyn CfdpPacketSender>, pub pdu_sender: PduSender,
vfs: Box<dyn VirtualFilestore>, pub vfs: Vfs,
remote_cfg_table: Box<dyn RemoteEntityConfigProvider>, pub remote_cfg_table: RemoteCfgTable,
check_timer_creator: Box<dyn CheckTimerCreator>, pub check_timer_creator: CheckTimerCreator,
} }
impl DestinationHandler { #[cfg(feature = "std")]
pub type StdDestinationHandler<PduSender> = DestinationHandler<
PduSender,
NativeFilestore,
StdRemoteEntityConfigProvider,
StdCheckTimerCreator,
StdCheckTimer,
>;
impl<
PduSender: CfdpPacketSender,
Vfs: VirtualFilestore,
RemoteCfgTable: RemoteEntityConfigProvider,
CheckTimerCreator: CheckTimerProviderCreator<CheckTimer = CheckTimerProvider>,
CheckTimerProvider: CountdownProvider,
> DestinationHandler<PduSender, Vfs, RemoteCfgTable, CheckTimerCreator, CheckTimerProvider>
{
/// Constructs a new destination handler. /// Constructs a new destination handler.
/// ///
/// # Arguments /// # Arguments
@ -226,10 +248,10 @@ impl DestinationHandler {
pub fn new( pub fn new(
local_cfg: LocalEntityConfig, local_cfg: LocalEntityConfig,
max_packet_len: usize, max_packet_len: usize,
packet_sender: Box<dyn CfdpPacketSender>, packet_sender: PduSender,
vfs: Box<dyn VirtualFilestore>, vfs: Vfs,
remote_cfg_table: Box<dyn RemoteEntityConfigProvider>, remote_cfg_table: RemoteCfgTable,
check_timer_creator: Box<dyn CheckTimerCreator>, check_timer_creator: CheckTimerCreator,
) -> Self { ) -> Self {
Self { Self {
local_cfg, local_cfg,
@ -237,7 +259,7 @@ impl DestinationHandler {
state: State::Idle, state: State::Idle,
tparams: Default::default(), tparams: Default::default(),
packet_buf: alloc::vec![0; max_packet_len], packet_buf: alloc::vec![0; max_packet_len],
packet_sender, pdu_sender: packet_sender,
vfs, vfs,
remote_cfg_table, remote_cfg_table,
check_timer_creator, check_timer_creator,
@ -524,7 +546,7 @@ impl DestinationHandler {
self.step = TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling; self.step = TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling;
self.tparams.tstate.current_check_timer = Some( self.tparams.tstate.current_check_timer = Some(
self.check_timer_creator self.check_timer_creator
.get_check_timer_provider(TimerContext::CheckLimit { .create_check_timer_provider(TimerContext::CheckLimit {
local_id: self.local_cfg.id, local_id: self.local_cfg.id,
remote_id: self.tparams.remote_cfg.unwrap().entity_id, remote_id: self.tparams.remote_cfg.unwrap().entity_id,
entity_type: EntityType::Receiving, entity_type: EntityType::Receiving,
@ -763,7 +785,7 @@ impl DestinationHandler {
) )
}; };
finished_pdu.write_to_bytes(&mut self.packet_buf)?; finished_pdu.write_to_bytes(&mut self.packet_buf)?;
self.packet_sender.send_pdu( self.pdu_sender.send_pdu(
finished_pdu.pdu_type(), finished_pdu.pdu_type(),
finished_pdu.file_directive_type(), finished_pdu.file_directive_type(),
&self.packet_buf[0..finished_pdu.len_written()], &self.packet_buf[0..finished_pdu.len_written()],
@ -771,23 +793,26 @@ impl DestinationHandler {
Ok(1) Ok(1)
} }
fn tstate(&self) -> &TransferState { fn tstate(&self) -> &TransferState<CheckTimerProvider> {
&self.tparams.tstate &self.tparams.tstate
} }
fn tstate_mut(&mut self) -> &mut TransferState { fn tstate_mut(&mut self) -> &mut TransferState<CheckTimerProvider> {
&mut self.tparams.tstate &mut self.tparams.tstate
} }
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use core::{cell::Cell, sync::atomic::AtomicBool}; use core::{
cell::{Cell, RefCell},
sync::atomic::AtomicBool,
};
#[allow(unused_imports)] #[allow(unused_imports)]
use std::println; use std::println;
use std::{fs, sync::Mutex}; use std::{fs, sync::Mutex};
use alloc::{collections::VecDeque, string::String, sync::Arc, vec::Vec}; use alloc::{boxed::Box, collections::VecDeque, string::String, sync::Arc, vec::Vec};
use rand::Rng; use rand::Rng;
use spacepackets::{ use spacepackets::{
cfdp::{ cfdp::{
@ -799,7 +824,7 @@ mod tests {
}; };
use crate::cfdp::{ use crate::cfdp::{
filestore::NativeFilestore, user::OwnedMetadataRecvdParams, CheckTimerCreator, filestore::NativeFilestore, user::OwnedMetadataRecvdParams, CheckTimerProviderCreator,
CountdownProvider, DefaultFaultHandler, IndicationConfig, RemoteEntityConfig, CountdownProvider, DefaultFaultHandler, IndicationConfig, RemoteEntityConfig,
StdRemoteEntityConfigProvider, UserFaultHandler, CRC_32, StdRemoteEntityConfigProvider, UserFaultHandler, CRC_32,
}; };
@ -820,20 +845,19 @@ mod tests {
file_directive_type: Option<FileDirectiveType>, file_directive_type: Option<FileDirectiveType>,
raw_pdu: Vec<u8>, raw_pdu: Vec<u8>,
} }
type SharedPduPacketQueue = Arc<Mutex<VecDeque<SentPdu>>>; #[derive(Default)]
#[derive(Default, Clone)]
struct TestCfdpSender { struct TestCfdpSender {
packet_queue: SharedPduPacketQueue, packet_queue: RefCell<VecDeque<SentPdu>>,
} }
impl CfdpPacketSender for TestCfdpSender { impl CfdpPacketSender for TestCfdpSender {
fn send_pdu( fn send_pdu(
&mut self, &self,
pdu_type: PduType, pdu_type: PduType,
file_directive_type: Option<FileDirectiveType>, file_directive_type: Option<FileDirectiveType>,
raw_pdu: &[u8], raw_pdu: &[u8],
) -> Result<(), PduError> { ) -> Result<(), PduError> {
self.packet_queue.lock().unwrap().push_back(SentPdu { self.packet_queue.borrow_mut().push_back(SentPdu {
pdu_type, pdu_type,
file_directive_type, file_directive_type,
raw_pdu: raw_pdu.to_vec(), raw_pdu: raw_pdu.to_vec(),
@ -844,10 +868,10 @@ mod tests {
impl TestCfdpSender { impl TestCfdpSender {
pub fn retrieve_next_pdu(&self) -> Option<SentPdu> { pub fn retrieve_next_pdu(&self) -> Option<SentPdu> {
self.packet_queue.lock().unwrap().pop_front() self.packet_queue.borrow_mut().pop_front()
} }
pub fn queue_empty(&self) -> bool { pub fn queue_empty(&self) -> bool {
self.packet_queue.lock().unwrap().is_empty() self.packet_queue.borrow_mut().is_empty()
} }
} }
@ -1087,14 +1111,13 @@ mod tests {
} }
} }
impl CheckTimerCreator for TestCheckTimerCreator { impl CheckTimerProviderCreator for TestCheckTimerCreator {
fn get_check_timer_provider( type CheckTimer = TestCheckTimer;
&self,
timer_context: TimerContext, fn create_check_timer_provider(&self, timer_context: TimerContext) -> Self::CheckTimer {
) -> Box<dyn CountdownProvider> {
match timer_context { match timer_context {
TimerContext::CheckLimit { .. } => { TimerContext::CheckLimit { .. } => {
Box::new(TestCheckTimer::new(self.check_limit_expired_flag.clone())) TestCheckTimer::new(self.check_limit_expired_flag.clone())
} }
_ => { _ => {
panic!("invalid check timer creator, can only be used for check limit handling") panic!("invalid check timer creator, can only be used for check limit handling")
@ -1103,10 +1126,17 @@ mod tests {
} }
} }
type TestDestHandler = DestinationHandler<
TestCfdpSender,
NativeFilestore,
StdRemoteEntityConfigProvider,
TestCheckTimerCreator,
TestCheckTimer,
>;
struct DestHandlerTester { struct DestHandlerTester {
check_timer_expired: Arc<AtomicBool>, check_timer_expired: Arc<AtomicBool>,
pdu_sender: TestCfdpSender, handler: TestDestHandler,
handler: DestinationHandler,
src_path: PathBuf, src_path: PathBuf,
dest_path: PathBuf, dest_path: PathBuf,
check_dest_file: bool, check_dest_file: bool,
@ -1122,16 +1152,12 @@ mod tests {
fn new(fault_handler: TestFaultHandler, closure_requested: bool) -> Self { fn new(fault_handler: TestFaultHandler, closure_requested: bool) -> Self {
let check_timer_expired = Arc::new(AtomicBool::new(false)); let check_timer_expired = Arc::new(AtomicBool::new(false));
let test_sender = TestCfdpSender::default(); let test_sender = TestCfdpSender::default();
let dest_handler = default_dest_handler( let dest_handler =
fault_handler, default_dest_handler(fault_handler, test_sender, check_timer_expired.clone());
test_sender.clone(),
check_timer_expired.clone(),
);
let (src_path, dest_path) = init_full_filenames(); let (src_path, dest_path) = init_full_filenames();
assert!(!Path::exists(&dest_path)); assert!(!Path::exists(&dest_path));
let handler = Self { let handler = Self {
check_timer_expired, check_timer_expired,
pdu_sender: test_sender,
handler: dest_handler, handler: dest_handler,
src_path, src_path,
closure_requested, closure_requested,
@ -1288,7 +1314,13 @@ mod tests {
test_fault_handler: TestFaultHandler, test_fault_handler: TestFaultHandler,
test_packet_sender: TestCfdpSender, test_packet_sender: TestCfdpSender,
check_timer_expired: Arc<AtomicBool>, check_timer_expired: Arc<AtomicBool>,
) -> DestinationHandler { ) -> DestinationHandler<
TestCfdpSender,
NativeFilestore,
StdRemoteEntityConfigProvider,
TestCheckTimerCreator,
TestCheckTimer,
> {
let local_entity_cfg = LocalEntityConfig { let local_entity_cfg = LocalEntityConfig {
id: REMOTE_ID.into(), id: REMOTE_ID.into(),
indication_cfg: IndicationConfig::default(), indication_cfg: IndicationConfig::default(),
@ -1297,10 +1329,10 @@ mod tests {
DestinationHandler::new( DestinationHandler::new(
local_entity_cfg, local_entity_cfg,
2048, 2048,
Box::new(test_packet_sender), test_packet_sender,
Box::<NativeFilestore>::default(), NativeFilestore::default(),
Box::new(basic_remote_cfg_table()), basic_remote_cfg_table(),
Box::new(TestCheckTimerCreator::new(check_timer_expired)), TestCheckTimerCreator::new(check_timer_expired),
) )
} }
@ -1366,18 +1398,18 @@ mod tests {
#[test] #[test]
fn test_empty_file_transfer_not_acked_no_closure() { fn test_empty_file_transfer_not_acked_no_closure() {
let fault_handler = TestFaultHandler::default(); let fault_handler = TestFaultHandler::default();
let mut test_obj = DestHandlerTester::new(fault_handler.clone(), false); let mut testbench = DestHandlerTester::new(fault_handler.clone(), false);
let mut test_user = test_obj.test_user_from_cached_paths(0); let mut test_user = testbench.test_user_from_cached_paths(0);
test_obj testbench
.generic_transfer_init(&mut test_user, 0) .generic_transfer_init(&mut test_user, 0)
.expect("transfer init failed"); .expect("transfer init failed");
test_obj.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus); testbench.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus);
test_obj testbench
.generic_eof_no_error(&mut test_user, Vec::new()) .generic_eof_no_error(&mut test_user, Vec::new())
.expect("EOF no error insertion failed"); .expect("EOF no error insertion failed");
assert!(fault_handler.all_queues_empty()); assert!(fault_handler.all_queues_empty());
assert!(test_obj.pdu_sender.queue_empty()); assert!(testbench.handler.pdu_sender.queue_empty());
test_obj.state_check(State::Idle, TransactionStep::Idle); testbench.state_check(State::Idle, TransactionStep::Idle);
} }
#[test] #[test]
@ -1387,21 +1419,21 @@ mod tests {
let file_size = file_data.len() as u64; let file_size = file_data.len() as u64;
let fault_handler = TestFaultHandler::default(); let fault_handler = TestFaultHandler::default();
let mut test_obj = DestHandlerTester::new(fault_handler.clone(), false); let mut testbench = DestHandlerTester::new(fault_handler.clone(), false);
let mut test_user = test_obj.test_user_from_cached_paths(file_size); let mut test_user = testbench.test_user_from_cached_paths(file_size);
test_obj testbench
.generic_transfer_init(&mut test_user, file_size) .generic_transfer_init(&mut test_user, file_size)
.expect("transfer init failed"); .expect("transfer init failed");
test_obj.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus); testbench.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus);
test_obj testbench
.generic_file_data_insert(&mut test_user, 0, file_data) .generic_file_data_insert(&mut test_user, 0, file_data)
.expect("file data insertion failed"); .expect("file data insertion failed");
test_obj testbench
.generic_eof_no_error(&mut test_user, file_data.to_vec()) .generic_eof_no_error(&mut test_user, file_data.to_vec())
.expect("EOF no error insertion failed"); .expect("EOF no error insertion failed");
assert!(fault_handler.all_queues_empty()); assert!(fault_handler.all_queues_empty());
assert!(test_obj.pdu_sender.queue_empty()); assert!(testbench.handler.pdu_sender.queue_empty());
test_obj.state_check(State::Idle, TransactionStep::Idle); testbench.state_check(State::Idle, TransactionStep::Idle);
} }
#[test] #[test]
@ -1413,28 +1445,28 @@ mod tests {
let segment_len = 256; let segment_len = 256;
let fault_handler = TestFaultHandler::default(); let fault_handler = TestFaultHandler::default();
let mut test_obj = DestHandlerTester::new(fault_handler.clone(), false); let mut testbench = DestHandlerTester::new(fault_handler.clone(), false);
let mut test_user = test_obj.test_user_from_cached_paths(file_size); let mut test_user = testbench.test_user_from_cached_paths(file_size);
test_obj testbench
.generic_transfer_init(&mut test_user, file_size) .generic_transfer_init(&mut test_user, file_size)
.expect("transfer init failed"); .expect("transfer init failed");
test_obj.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus); testbench.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus);
test_obj testbench
.generic_file_data_insert(&mut test_user, 0, &random_data[0..segment_len]) .generic_file_data_insert(&mut test_user, 0, &random_data[0..segment_len])
.expect("file data insertion failed"); .expect("file data insertion failed");
test_obj testbench
.generic_file_data_insert( .generic_file_data_insert(
&mut test_user, &mut test_user,
segment_len as u64, segment_len as u64,
&random_data[segment_len..], &random_data[segment_len..],
) )
.expect("file data insertion failed"); .expect("file data insertion failed");
test_obj testbench
.generic_eof_no_error(&mut test_user, random_data.to_vec()) .generic_eof_no_error(&mut test_user, random_data.to_vec())
.expect("EOF no error insertion failed"); .expect("EOF no error insertion failed");
assert!(fault_handler.all_queues_empty()); assert!(fault_handler.all_queues_empty());
assert!(test_obj.pdu_sender.queue_empty()); assert!(testbench.handler.pdu_sender.queue_empty());
test_obj.state_check(State::Idle, TransactionStep::Idle); testbench.state_check(State::Idle, TransactionStep::Idle);
} }
#[test] #[test]
@ -1446,32 +1478,32 @@ mod tests {
let segment_len = 256; let segment_len = 256;
let fault_handler = TestFaultHandler::default(); let fault_handler = TestFaultHandler::default();
let mut test_obj = DestHandlerTester::new(fault_handler.clone(), false); let mut testbench = DestHandlerTester::new(fault_handler.clone(), false);
let mut test_user = test_obj.test_user_from_cached_paths(file_size); let mut test_user = testbench.test_user_from_cached_paths(file_size);
let transaction_id = test_obj let transaction_id = testbench
.generic_transfer_init(&mut test_user, file_size) .generic_transfer_init(&mut test_user, file_size)
.expect("transfer init failed"); .expect("transfer init failed");
test_obj.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus); testbench.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus);
test_obj testbench
.generic_file_data_insert(&mut test_user, 0, &random_data[0..segment_len]) .generic_file_data_insert(&mut test_user, 0, &random_data[0..segment_len])
.expect("file data insertion 0 failed"); .expect("file data insertion 0 failed");
test_obj testbench
.generic_eof_no_error(&mut test_user, random_data.to_vec()) .generic_eof_no_error(&mut test_user, random_data.to_vec())
.expect("EOF no error insertion failed"); .expect("EOF no error insertion failed");
test_obj.state_check( testbench.state_check(
State::Busy, State::Busy,
TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling, TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling,
); );
test_obj testbench
.generic_file_data_insert( .generic_file_data_insert(
&mut test_user, &mut test_user,
segment_len as u64, segment_len as u64,
&random_data[segment_len..], &random_data[segment_len..],
) )
.expect("file data insertion 1 failed"); .expect("file data insertion 1 failed");
test_obj.set_check_timer_expired(); testbench.set_check_timer_expired();
test_obj testbench
.handler .handler
.state_machine(&mut test_user, None) .state_machine(&mut test_user, None)
.expect("fsm failure"); .expect("fsm failure");
@ -1482,8 +1514,8 @@ mod tests {
assert_eq!(cancelled.0, transaction_id); assert_eq!(cancelled.0, transaction_id);
assert_eq!(cancelled.1, ConditionCode::FileChecksumFailure); assert_eq!(cancelled.1, ConditionCode::FileChecksumFailure);
assert_eq!(cancelled.2, segment_len as u64); assert_eq!(cancelled.2, segment_len as u64);
assert!(test_obj.pdu_sender.queue_empty()); assert!(testbench.handler.pdu_sender.queue_empty());
test_obj.state_check(State::Idle, TransactionStep::Idle); testbench.state_check(State::Idle, TransactionStep::Idle);
} }
#[test] #[test]
@ -1495,38 +1527,38 @@ mod tests {
let segment_len = 256; let segment_len = 256;
let fault_handler = TestFaultHandler::default(); let fault_handler = TestFaultHandler::default();
let mut test_obj = DestHandlerTester::new(fault_handler.clone(), false); let mut testbench = DestHandlerTester::new(fault_handler.clone(), false);
let mut test_user = test_obj.test_user_from_cached_paths(file_size); let mut test_user = testbench.test_user_from_cached_paths(file_size);
let transaction_id = test_obj let transaction_id = testbench
.generic_transfer_init(&mut test_user, file_size) .generic_transfer_init(&mut test_user, file_size)
.expect("transfer init failed"); .expect("transfer init failed");
test_obj.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus); testbench.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus);
test_obj testbench
.generic_file_data_insert(&mut test_user, 0, &random_data[0..segment_len]) .generic_file_data_insert(&mut test_user, 0, &random_data[0..segment_len])
.expect("file data insertion 0 failed"); .expect("file data insertion 0 failed");
test_obj testbench
.generic_eof_no_error(&mut test_user, random_data.to_vec()) .generic_eof_no_error(&mut test_user, random_data.to_vec())
.expect("EOF no error insertion failed"); .expect("EOF no error insertion failed");
test_obj.state_check( testbench.state_check(
State::Busy, State::Busy,
TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling, TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling,
); );
test_obj.set_check_timer_expired(); testbench.set_check_timer_expired();
test_obj testbench
.handler .handler
.state_machine(&mut test_user, None) .state_machine(&mut test_user, None)
.expect("fsm error"); .expect("fsm error");
test_obj.state_check( testbench.state_check(
State::Busy, State::Busy,
TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling, TransactionStep::ReceivingFileDataPdusWithCheckLimitHandling,
); );
test_obj.set_check_timer_expired(); testbench.set_check_timer_expired();
test_obj testbench
.handler .handler
.state_machine(&mut test_user, None) .state_machine(&mut test_user, None)
.expect("fsm error"); .expect("fsm error");
test_obj.state_check(State::Idle, TransactionStep::Idle); testbench.state_check(State::Idle, TransactionStep::Idle);
assert!(fault_handler assert!(fault_handler
.notice_of_suspension_queue .notice_of_suspension_queue
@ -1550,15 +1582,15 @@ mod tests {
drop(cancelled_queue); drop(cancelled_queue);
assert!(test_obj.pdu_sender.queue_empty()); assert!(testbench.handler.pdu_sender.queue_empty());
// Check that the broken file exists. // Check that the broken file exists.
test_obj.check_dest_file = false; testbench.check_dest_file = false;
assert!(Path::exists(test_obj.dest_path())); assert!(Path::exists(testbench.dest_path()));
let read_content = fs::read(test_obj.dest_path()).expect("reading back string failed"); let read_content = fs::read(testbench.dest_path()).expect("reading back string failed");
assert_eq!(read_content.len(), segment_len); assert_eq!(read_content.len(), segment_len);
assert_eq!(read_content, &random_data[0..segment_len]); assert_eq!(read_content, &random_data[0..segment_len]);
assert!(fs::remove_file(test_obj.dest_path().as_path()).is_ok()); assert!(fs::remove_file(testbench.dest_path().as_path()).is_ok());
} }
fn check_finished_pdu_success(sent_pdu: &SentPdu) { fn check_finished_pdu_success(sent_pdu: &SentPdu) {
@ -1578,21 +1610,21 @@ mod tests {
#[test] #[test]
fn test_file_transfer_with_closure() { fn test_file_transfer_with_closure() {
let fault_handler = TestFaultHandler::default(); let fault_handler = TestFaultHandler::default();
let mut test_obj = DestHandlerTester::new(fault_handler.clone(), true); let mut testbench = DestHandlerTester::new(fault_handler.clone(), true);
let mut test_user = test_obj.test_user_from_cached_paths(0); let mut test_user = testbench.test_user_from_cached_paths(0);
test_obj testbench
.generic_transfer_init(&mut test_user, 0) .generic_transfer_init(&mut test_user, 0)
.expect("transfer init failed"); .expect("transfer init failed");
test_obj.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus); testbench.state_check(State::Busy, TransactionStep::ReceivingFileDataPdus);
let sent_packets = test_obj let sent_packets = testbench
.generic_eof_no_error(&mut test_user, Vec::new()) .generic_eof_no_error(&mut test_user, Vec::new())
.expect("EOF no error insertion failed"); .expect("EOF no error insertion failed");
assert_eq!(sent_packets, 1); assert_eq!(sent_packets, 1);
assert!(fault_handler.all_queues_empty()); assert!(fault_handler.all_queues_empty());
// The Finished PDU was sent, so the state machine is done. // The Finished PDU was sent, so the state machine is done.
test_obj.state_check(State::Idle, TransactionStep::Idle); testbench.state_check(State::Idle, TransactionStep::Idle);
assert!(!test_obj.pdu_sender.queue_empty()); assert!(!testbench.handler.pdu_sender.queue_empty());
let sent_pdu = test_obj.pdu_sender.retrieve_next_pdu().unwrap(); let sent_pdu = testbench.handler.pdu_sender.retrieve_next_pdu().unwrap();
check_finished_pdu_success(&sent_pdu); check_finished_pdu_success(&sent_pdu);
} }

98
satrs/src/cfdp/mod.rs
View File

@ -1,6 +1,6 @@
//! This module contains the implementation of the CFDP high level classes as specified in the //! This module contains the implementation of the CFDP high level classes as specified in the
//! CCSDS 727.0-B-5. //! CCSDS 727.0-B-5.
use core::{cell::RefCell, fmt::Debug, hash::Hash}; use core::{cell::RefCell, fmt::Debug, hash::Hash, time::Duration};
use crc::{Crc, CRC_32_CKSUM}; use crc::{Crc, CRC_32_CKSUM};
use hashbrown::HashMap; use hashbrown::HashMap;
@ -27,6 +27,9 @@ pub mod filestore;
pub mod source; pub mod source;
pub mod user; pub mod user;
#[cfg(feature = "std")]
pub use std_mod::*;
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EntityType { pub enum EntityType {
Sending, Sending,
@ -84,42 +87,77 @@ pub enum TimerContext {
/// The timer will be used to perform the Positive Acknowledgement Procedures as specified in /// The timer will be used to perform the Positive Acknowledgement Procedures as specified in
/// 4.7. 1of the CFDP standard. The expiration period will be provided by the Positive ACK timer /// 4.7. 1of the CFDP standard. The expiration period will be provided by the Positive ACK timer
/// interval of the remote entity configuration. /// interval of the remote entity configuration.
#[cfg(feature = "alloc")] pub trait CheckTimerProviderCreator {
pub trait CheckTimerCreator { type CheckTimer: CountdownProvider;
fn get_check_timer_provider(&self, timer_context: TimerContext) -> Box<dyn CountdownProvider>;
}
/// Simple implementation of the [CheckTimerCreator] trait assuming a standard runtime. fn create_check_timer_provider(&self, timer_context: TimerContext) -> Self::CheckTimer;
/// It also assumes that a second accuracy of the check timer period is sufficient.
#[cfg(feature = "std")]
#[derive(Debug)]
pub struct StdCheckTimer {
expiry_time_seconds: u64,
start_time: std::time::Instant,
} }
#[cfg(feature = "std")] #[cfg(feature = "std")]
impl StdCheckTimer { pub mod std_mod {
pub fn new(expiry_time_seconds: u64) -> Self { use super::*;
Self {
expiry_time_seconds,
start_time: std::time::Instant::now(),
}
}
}
#[cfg(feature = "std")] /// Simple implementation of the [CheckTimerCreator] trait assuming a standard runtime.
impl CountdownProvider for StdCheckTimer { /// It also assumes that a second accuracy of the check timer period is sufficient.
fn has_expired(&self) -> bool { #[derive(Debug)]
let elapsed_time = self.start_time.elapsed(); pub struct StdCheckTimer {
if elapsed_time.as_secs() > self.expiry_time_seconds { expiry_time_seconds: u64,
return true; start_time: std::time::Instant,
}
false
} }
fn reset(&mut self) { impl StdCheckTimer {
self.start_time = std::time::Instant::now(); pub fn new(expiry_time_seconds: u64) -> Self {
Self {
expiry_time_seconds,
start_time: std::time::Instant::now(),
}
}
}
impl CountdownProvider for StdCheckTimer {
fn has_expired(&self) -> bool {
let elapsed_time = self.start_time.elapsed();
if elapsed_time.as_secs() > self.expiry_time_seconds {
return true;
}
false
}
fn reset(&mut self) {
self.start_time = std::time::Instant::now();
}
}
pub struct StdCheckTimerCreator {
pub check_limit_timeout_secs: u64,
}
impl Default for StdCheckTimerCreator {
fn default() -> Self {
Self {
check_limit_timeout_secs: 5,
}
}
}
impl CheckTimerProviderCreator for StdCheckTimerCreator {
type CheckTimer = StdCheckTimer;
fn create_check_timer_provider(&self, timer_context: TimerContext) -> Self::CheckTimer {
match timer_context {
TimerContext::CheckLimit {
local_id: _,
remote_id: _,
entity_type: _,
} => StdCheckTimer::new(self.check_limit_timeout_secs),
TimerContext::NakActivity {
expiry_time_seconds,
} => StdCheckTimer::new(Duration::from_secs_f32(expiry_time_seconds).as_secs()),
TimerContext::PositiveAck {
expiry_time_seconds,
} => StdCheckTimer::new(Duration::from_secs_f32(expiry_time_seconds).as_secs()),
}
}
} }
} }