rust/cfdp
7
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases 1 Wiki Activity

continue example
All checks were successful
Rust/cfdp/pipeline/head This commit looks good
Details

Browse Source
This commit is contained in:
Robin Müller
2024-08-28 14:02:39 +02:00
parent e7d6663022
commit 6771787e49
7 changed files with 224 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.toml
View File

@ -55,6 +55,8 @@ serde = ["dep:serde", "spacepackets/serde"]
tempfile = "3" tempfile = "3"
rand = "0.8" rand = "0.8"
log = "0.4" log = "0.4"
fern = "0.6"
chrono = "0.4"
[package.metadata.docs.rs] [package.metadata.docs.rs]
all-features = true all-features = true

9
examples/python-interop/main.py
View File

@ -92,8 +92,9 @@ REMOTE_CFG_OF_PY_ENTITY = RemoteEntityCfg(
REMOTE_CFG_OF_REMOTE_ENTITY = copy.copy(REMOTE_CFG_OF_PY_ENTITY) REMOTE_CFG_OF_REMOTE_ENTITY = copy.copy(REMOTE_CFG_OF_PY_ENTITY)
REMOTE_CFG_OF_REMOTE_ENTITY.entity_id = RUST_ENTITY_ID REMOTE_CFG_OF_REMOTE_ENTITY.entity_id = RUST_ENTITY_ID
LOCAL_PORT = 5111 RUST_PORT = 5111
REMOTE_PORT = 5222 PY_PORT = 5222
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -622,8 +623,8 @@ def main():
_LOGGER.info(f"Put request will be sent to remote destination {remote_addr}") _LOGGER.info(f"Put request will be sent to remote destination {remote_addr}")
udp_server = UdpServer( udp_server = UdpServer(
sleep_time=0.1, sleep_time=0.1,
addr=(str(local_addr), LOCAL_PORT), addr=(str(local_addr), PY_PORT),
explicit_remote_addr=(str(remote_addr), REMOTE_PORT), explicit_remote_addr=(str(remote_addr), RUST_PORT),
tx_queue=TM_QUEUE, tx_queue=TM_QUEUE,
source_entity_rx_queue=SOURCE_ENTITY_QUEUE, source_entity_rx_queue=SOURCE_ENTITY_QUEUE,
dest_entity_rx_queue=DEST_ENTITY_QUEUE, dest_entity_rx_queue=DEST_ENTITY_QUEUE,

283
examples/python-interop/main.rs
View File

@ -2,7 +2,7 @@ use std::{
fmt::Debug, fmt::Debug,
fs::OpenOptions, fs::OpenOptions,
io::{self, ErrorKind, Write}, io::{self, ErrorKind, Write},
net::{SocketAddr, ToSocketAddrs, UdpSocket}, net::{IpAddr, Ipv4Addr, SocketAddr, ToSocketAddrs, UdpSocket},
sync::{atomic::AtomicBool, mpsc, Arc}, sync::{atomic::AtomicBool, mpsc, Arc},
thread, thread,
time::Duration, time::Duration,
@ -10,24 +10,28 @@ use std::{
use cfdp::{ use cfdp::{
dest::DestinationHandler, dest::DestinationHandler,
determine_packet_target,
filestore::NativeFilestore, filestore::NativeFilestore,
request::{PutRequestOwned, StaticPutRequestCacher}, request::{PutRequestOwned, StaticPutRequestCacher},
source::SourceHandler, source::SourceHandler,
user::{CfdpUser, FileSegmentRecvdParams, MetadataReceivedParams, TransactionFinishedParams}, user::{CfdpUser, FileSegmentRecvdParams, MetadataReceivedParams, TransactionFinishedParams},
EntityType, IndicationConfig, LocalEntityConfig, PduWithInfo, RemoteEntityConfig, EntityType, IndicationConfig, LocalEntityConfig, PduOwnedWithInfo, PduProvider,
StdCheckTimerCreator, TransactionId, UserFaultHookProvider, RemoteEntityConfig, StdCheckTimerCreator, TransactionId, UserFaultHookProvider,
}; };
use log::{info, warn}; use log::{debug, info, warn};
use spacepackets::{ use spacepackets::{
cfdp::{pdu::PduError, ChecksumType, ConditionCode, TransmissionMode}, cfdp::{pdu::PduError, ChecksumType, ConditionCode, TransmissionMode},
seq_count::SeqCountProviderSyncU16, seq_count::SeqCountProviderSyncU16,
util::UnsignedByteFieldU16, util::{UnsignedByteFieldU16, UnsignedEnum},
}; };
const PYTHON_ID: UnsignedByteFieldU16 = UnsignedByteFieldU16::new(1); const PYTHON_ID: UnsignedByteFieldU16 = UnsignedByteFieldU16::new(1);
const RUST_ID: UnsignedByteFieldU16 = UnsignedByteFieldU16::new(2); const RUST_ID: UnsignedByteFieldU16 = UnsignedByteFieldU16::new(2);
const RUST_PORT: u16 = 5111;
const PY_PORT: u16 = 5222;
const LOG_LEVEL: log::LevelFilter = log::LevelFilter::Info;
const FILE_DATA: &str = "Hello World!"; const FILE_DATA: &str = "Hello World!";
#[derive(Default)] #[derive(Default)]
@ -163,15 +167,14 @@ pub struct UdpServer {
pub socket: UdpSocket, pub socket: UdpSocket,
recv_buf: Vec<u8>, recv_buf: Vec<u8>,
sender_addr: Option<SocketAddr>, sender_addr: Option<SocketAddr>,
src_tx: mpsc::SyncSender<Vec<u8>>, source_tc_tx: mpsc::Sender<PduOwnedWithInfo>,
dest_tx: mpsc::SyncSender<Vec<u8>>, dest_tc_tx: mpsc::Sender<PduOwnedWithInfo>,
tm_rx: mpsc::Receiver<Vec<u8>>, source_tm_rx: mpsc::Receiver<PduOwnedWithInfo>,
dest_tm_rx: mpsc::Receiver<PduOwnedWithInfo>,
} }
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum UdpServerError { pub enum UdpServerError {
#[error("nothing was received")]
NothingReceived,
#[error(transparent)] #[error(transparent)]
Io(#[from] io::Error), Io(#[from] io::Error),
#[error("pdu error: {0}")] #[error("pdu error: {0}")]
@ -184,58 +187,69 @@ impl UdpServer {
pub fn new<A: ToSocketAddrs>( pub fn new<A: ToSocketAddrs>(
addr: A, addr: A,
max_recv_size: usize, max_recv_size: usize,
src_tx: mpsc::SyncSender<Vec<u8>>, source_tc_tx: mpsc::Sender<PduOwnedWithInfo>,
dest_tx: mpsc::SyncSender<Vec<u8>>, dest_tc_tx: mpsc::Sender<PduOwnedWithInfo>,
tm_rx: mpsc::Receiver<Vec<u8>>, source_tm_rx: mpsc::Receiver<PduOwnedWithInfo>,
dest_tm_rx: mpsc::Receiver<PduOwnedWithInfo>,
) -> Result<Self, io::Error> { ) -> Result<Self, io::Error> {
let server = Self { let server = Self {
socket: UdpSocket::bind(addr)?, socket: UdpSocket::bind(addr)?,
recv_buf: vec![0; max_recv_size], recv_buf: vec![0; max_recv_size],
tm_rx, source_tc_tx,
src_tx, dest_tc_tx,
dest_tx,
sender_addr: None, sender_addr: None,
source_tm_rx,
dest_tm_rx,
}; };
server.socket.set_nonblocking(true)?; server.socket.set_nonblocking(true)?;
Ok(server) Ok(server)
} }
pub fn try_recv_tc(&mut self) -> Result<(usize, SocketAddr), UdpServerError> { pub fn try_recv_tc(
&mut self,
) -> Result<Option<(PduOwnedWithInfo, SocketAddr)>, UdpServerError> {
let res = match self.socket.recv_from(&mut self.recv_buf) { let res = match self.socket.recv_from(&mut self.recv_buf) {
Ok(res) => res, Ok(res) => res,
Err(e) => { Err(e) => {
return if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut { return if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut {
Err(UdpServerError::NothingReceived) Ok(None)
} else { } else {
Err(e.into()) Err(e.into())
} }
} }
}; };
let (num_bytes, from) = res; let (_, from) = res;
self.sender_addr = Some(from); self.sender_addr = Some(from);
let packet_target = determine_packet_target(&self.recv_buf)?; let pdu_owned = PduOwnedWithInfo::new_from_raw_packet(&self.recv_buf)?;
match packet_target { match pdu_owned.packet_target()? {
cfdp::PacketTarget::SourceEntity => { cfdp::PacketTarget::SourceEntity => {
self.src_tx self.source_tc_tx
.send(self.recv_buf[0..num_bytes].to_vec()) .send(pdu_owned.clone())
.map_err(|_| UdpServerError::Send)?; .map_err(|_| UdpServerError::Send)?;
} }
cfdp::PacketTarget::DestEntity => { cfdp::PacketTarget::DestEntity => {
self.dest_tx self.dest_tc_tx
.send(self.recv_buf[0..num_bytes].to_vec()) .send(pdu_owned.clone())
.map_err(|_| UdpServerError::Send)?; .map_err(|_| UdpServerError::Send)?;
} }
} }
Ok(res) Ok(Some((pdu_owned, from)))
} }
pub fn send_tm(&mut self, socket: &UdpSocket, recv_addr: &SocketAddr) { pub fn recv_and_send_telemetry(&mut self) {
while let Ok(tm) = self.tm_rx.try_recv() { if self.last_sender().is_none() {
let result = socket.send_to(&tm, recv_addr); return;
if let Err(e) = result {
warn!("Sending TM with UDP socket failed: {e}")
}
} }
let tm_handler = |receiver: &mpsc::Receiver<PduOwnedWithInfo>| {
while let Ok(tm) = receiver.try_recv() {
let result = self.socket.send_to(tm.pdu(), self.last_sender().unwrap());
if let Err(e) = result {
warn!("Sending TM with UDP socket failed: {e}")
}
}
};
tm_handler(&self.source_tm_rx);
tm_handler(&self.dest_tm_rx);
} }
pub fn last_sender(&self) -> Option<SocketAddr> { pub fn last_sender(&self) -> Option<SocketAddr> {
@ -244,6 +258,20 @@ impl UdpServer {
} }
fn main() { fn main() {
fern::Dispatch::new()
.format(|out, message, record| {
out.finish(format_args!(
"{}[{}][{}] {}",
chrono::Local::now().format("[%Y-%m-%d][%H:%M:%S]"),
std::thread::current().name().expect("thread is not named"),
record.level(),
message
))
})
.level(LOG_LEVEL)
.chain(std::io::stdout())
.apply()
.unwrap();
// Simplified event handling using atomic signals. // Simplified event handling using atomic signals.
let stop_signal_source = Arc::new(AtomicBool::new(false)); let stop_signal_source = Arc::new(AtomicBool::new(false));
let stop_signal_dest = stop_signal_source.clone(); let stop_signal_dest = stop_signal_source.clone();
@ -270,8 +298,8 @@ fn main() {
IndicationConfig::default(), IndicationConfig::default(),
ExampleFaultHandler::default(), ExampleFaultHandler::default(),
); );
let (source_tx, source_rx) = mpsc::channel::<PduWithInfo>(); let (source_tm_tx, source_tm_rx) = mpsc::channel::<PduOwnedWithInfo>();
let (dest_tx, dest_rx) = mpsc::channel::<PduWithInfo>(); let (dest_tm_tx, dest_tm_rx) = mpsc::channel::<PduOwnedWithInfo>();
let put_request_cacher = StaticPutRequestCacher::new(2048); let put_request_cacher = StaticPutRequestCacher::new(2048);
let remote_cfg_python = RemoteEntityConfig::new_with_default_values( let remote_cfg_python = RemoteEntityConfig::new_with_default_values(
PYTHON_ID.into(), PYTHON_ID.into(),
@ -284,7 +312,7 @@ fn main() {
let seq_count_provider = SeqCountProviderSyncU16::default(); let seq_count_provider = SeqCountProviderSyncU16::default();
let mut source_handler = SourceHandler::new( let mut source_handler = SourceHandler::new(
local_cfg_source, local_cfg_source,
source_tx, source_tm_tx,
NativeFilestore::default(), NativeFilestore::default(),
put_request_cacher, put_request_cacher,
2048, 2048,
@ -301,7 +329,7 @@ fn main() {
let mut dest_handler = DestinationHandler::new( let mut dest_handler = DestinationHandler::new(
local_cfg_dest, local_cfg_dest,
1024, 1024,
dest_tx, dest_tm_tx,
NativeFilestore::default(), NativeFilestore::default(),
remote_cfg_python, remote_cfg_python,
StdCheckTimerCreator::default(), StdCheckTimerCreator::default(),
@ -317,88 +345,134 @@ fn main() {
) )
.expect("put request creation failed"); .expect("put request creation failed");
let (source_tc_tx, source_tc_rx) = mpsc::channel();
let (dest_tc_tx, dest_tc_rx) = mpsc::channel();
let sock_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), RUST_PORT);
let mut udp_server = UdpServer::new(
sock_addr,
2048,
source_tc_tx,
dest_tc_tx,
source_tm_rx,
dest_tm_rx,
)
.expect("creating UDP server failed");
let start = std::time::Instant::now(); let start = std::time::Instant::now();
let jh_source = thread::Builder::new()
let jh_source = thread::spawn(move || { .name("cfdp src entity".to_string())
source_handler .spawn(move || {
.put_request(&put_request) info!("Starting RUST SRC");
.expect("put request failed"); source_handler
loop { .put_request(&put_request)
let mut next_delay = None; .expect("put request failed");
let mut undelayed_call_count = 0; loop {
let packet_info = match dest_rx.try_recv() { let mut next_delay = None;
Ok(pdu_with_info) => Some(pdu_with_info), let mut undelayed_call_count = 0;
Err(e) => match e { let packet_info = match source_tc_rx.try_recv() {
mpsc::TryRecvError::Empty => None, Ok(pdu_with_info) => Some(pdu_with_info),
mpsc::TryRecvError::Disconnected => { Err(e) => match e {
panic!("unexpected disconnect from destination channel sender"); mpsc::TryRecvError::Empty => None,
mpsc::TryRecvError::Disconnected => {
panic!("unexpected disconnect from destination channel sender");
}
},
};
match source_handler.state_machine(&mut cfdp_user_source, packet_info.as_ref()) {
Ok(sent_packets) => {
if sent_packets == 0 {
next_delay = Some(Duration::from_millis(50));
}
} }
}, Err(e) => {
}; println!("Source handler error: {}", e);
match source_handler.state_machine(&mut cfdp_user_source, packet_info.as_ref()) {
Ok(sent_packets) => {
if sent_packets == 0 {
next_delay = Some(Duration::from_millis(50)); next_delay = Some(Duration::from_millis(50));
} }
} }
Err(e) => { if let Some(delay) = next_delay {
println!("Source handler error: {}", e); thread::sleep(delay);
next_delay = Some(Duration::from_millis(50)); } else {
undelayed_call_count += 1;
}
if stop_signal_source.load(std::sync::atomic::Ordering::Relaxed) {
break;
}
// Safety feature against configuration errors.
if undelayed_call_count >= 200 {
panic!("Source handler state machine possible in permanent loop");
} }
} }
if let Some(delay) = next_delay { })
thread::sleep(delay); .unwrap();
} else {
undelayed_call_count += 1;
}
if stop_signal_source.load(std::sync::atomic::Ordering::Relaxed) {
break;
}
// Safety feature against configuration errors.
if undelayed_call_count >= 200 {
panic!("Source handler state machine possible in permanent loop");
}
}
});
let jh_dest = thread::spawn(move || { let jh_dest = thread::Builder::new()
loop { .name("cfdp dest entity".to_string())
let mut next_delay = None; .spawn(move || {
let mut undelayed_call_count = 0; info!("Starting RUST DEST. Local ID {}", RUST_ID.value());
let packet_info = match source_rx.try_recv() { loop {
Ok(pdu_with_info) => Some(pdu_with_info), let mut next_delay = None;
Err(e) => match e { let mut undelayed_call_count = 0;
mpsc::TryRecvError::Empty => None, let packet_info = match dest_tc_rx.try_recv() {
mpsc::TryRecvError::Disconnected => { Ok(pdu_with_info) => Some(pdu_with_info),
panic!("unexpected disconnect from destination channel sender"); Err(e) => match e {
mpsc::TryRecvError::Empty => None,
mpsc::TryRecvError::Disconnected => {
panic!("unexpected disconnect from destination channel sender");
}
},
};
match dest_handler.state_machine(&mut cfdp_user_dest, packet_info.as_ref()) {
Ok(sent_packets) => {
if sent_packets == 0 {
next_delay = Some(Duration::from_millis(50));
}
} }
}, Err(e) => {
}; println!("Source handler error: {}", e);
match dest_handler.state_machine(&mut cfdp_user_dest, packet_info.as_ref()) {
Ok(sent_packets) => {
if sent_packets == 0 {
next_delay = Some(Duration::from_millis(50)); next_delay = Some(Duration::from_millis(50));
} }
} }
Err(e) => { if let Some(delay) = next_delay {
println!("Source handler error: {}", e); thread::sleep(delay);
next_delay = Some(Duration::from_millis(50)); } else {
undelayed_call_count += 1;
}
if stop_signal_dest.load(std::sync::atomic::Ordering::Relaxed) {
break;
}
// Safety feature against configuration errors.
if undelayed_call_count >= 200 {
panic!("Destination handler state machine possible in permanent loop");
} }
} }
if let Some(delay) = next_delay { })
thread::sleep(delay); .unwrap();
} else {
undelayed_call_count += 1; let jh_udp_server = thread::Builder::new()
.name("cfdp udp server".to_string())
.spawn(move || {
info!("Starting UDP server on {}", sock_addr);
loop {
loop {
match udp_server.try_recv_tc() {
Ok(result) => match result {
Some((pdu, _addr)) => {
debug!("Received PDU on UDP server: {:?}", pdu);
}
None => break,
},
Err(e) => {
warn!("UDP server error: {}", e);
break;
}
}
}
udp_server.recv_and_send_telemetry();
thread::sleep(Duration::from_millis(50));
} }
if stop_signal_dest.load(std::sync::atomic::Ordering::Relaxed) { })
break; .unwrap();
}
// Safety feature against configuration errors.
if undelayed_call_count >= 200 {
panic!("Destination handler state machine possible in permanent loop");
}
}
});
loop { loop {
if completion_signal_source_main.load(std::sync::atomic::Ordering::Relaxed) if completion_signal_source_main.load(std::sync::atomic::Ordering::Relaxed)
@ -418,4 +492,5 @@ fn main() {
jh_source.join().unwrap(); jh_source.join().unwrap();
jh_dest.join().unwrap(); jh_dest.join().unwrap();
jh_udp_server.join().unwrap();
} }

12
src/dest.rs
View File

@ -876,8 +876,8 @@ mod tests {
basic_remote_cfg_table, SentPdu, TestCfdpSender, TestCfdpUser, TestFaultHandler, basic_remote_cfg_table, SentPdu, TestCfdpSender, TestCfdpUser, TestFaultHandler,
LOCAL_ID, LOCAL_ID,
}, },
CheckTimerProviderCreator, CountdownProvider, FaultHandler, IndicationConfig, PacketInfo, CheckTimerProviderCreator, CountdownProvider, FaultHandler, IndicationConfig,
StdRemoteEntityConfigProvider, CRC_32, PduRawWithInfo, StdRemoteEntityConfigProvider, CRC_32,
}; };
use super::*; use super::*;
@ -1069,7 +1069,7 @@ mod tests {
filedata_pdu filedata_pdu
.write_to_bytes(&mut self.buf) .write_to_bytes(&mut self.buf)
.expect("writing file data PDU failed"); .expect("writing file data PDU failed");
let packet_info = PacketInfo::new(&self.buf).expect("creating packet info failed"); let packet_info = PduRawWithInfo::new(&self.buf).expect("creating packet info failed");
let result = self.handler.state_machine(user, Some(&packet_info)); let result = self.handler.state_machine(user, Some(&packet_info));
if self.indication_cfg().file_segment_recv { if self.indication_cfg().file_segment_recv {
assert!(!user.file_seg_recvd_queue.is_empty()); assert!(!user.file_seg_recvd_queue.is_empty());
@ -1198,11 +1198,11 @@ mod tests {
fn create_packet_info<'a>( fn create_packet_info<'a>(
pdu: &'a impl WritablePduPacket, pdu: &'a impl WritablePduPacket,
buf: &'a mut [u8], buf: &'a mut [u8],
) -> PacketInfo<'a> { ) -> PduRawWithInfo<'a> {
let written_len = pdu let written_len = pdu
.write_to_bytes(buf) .write_to_bytes(buf)
.expect("writing metadata PDU failed"); .expect("writing metadata PDU failed");
PacketInfo::new(&buf[..written_len]).expect("generating packet info failed") PduRawWithInfo::new(&buf[..written_len]).expect("generating packet info failed")
} }
fn create_no_error_eof(file_data: &[u8], pdu_header: &PduHeader) -> EofPdu { fn create_no_error_eof(file_data: &[u8], pdu_header: &PduHeader) -> EofPdu {
@ -1460,7 +1460,7 @@ mod tests {
FileStatus::Retained, FileStatus::Retained,
); );
let finished_pdu_raw = finished_pdu.to_vec().unwrap(); let finished_pdu_raw = finished_pdu.to_vec().unwrap();
let packet_info = PacketInfo::new(&finished_pdu_raw).unwrap(); let packet_info = PduRawWithInfo::new(&finished_pdu_raw).unwrap();
let error = tb.handler.state_machine(&mut user, Some(&packet_info)); let error = tb.handler.state_machine(&mut user, Some(&packet_info));
assert!(error.is_err()); assert!(error.is_err());
let error = error.unwrap_err(); let error = error.unwrap_err();

39
src/lib.rs
View File

@ -546,14 +546,14 @@ pub mod std_mod {
use super::*; use super::*;
impl PduSendProvider for mpsc::Sender<PduWithInfo> { impl PduSendProvider for mpsc::Sender<PduOwnedWithInfo> {
fn send_pdu( fn send_pdu(
&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<(), GenericSendError> { ) -> Result<(), GenericSendError> {
self.send(PduWithInfo::new( self.send(PduOwnedWithInfo::new(
pdu_type, pdu_type,
file_directive_type, file_directive_type,
raw_pdu.to_vec(), raw_pdu.to_vec(),
@ -721,7 +721,7 @@ impl PduProvider for DummyPduProvider {
/// This is a helper struct which contains base information about a particular PDU packet. /// This is a helper struct which contains base information about a particular PDU packet.
/// This is also necessary information for CFDP packet routing. For example, some packet types /// This is also necessary information for CFDP packet routing. For example, some packet types
/// like file data PDUs can only be used by CFDP source entities. /// like file data PDUs can only be used by CFDP source entities.
pub struct PacketInfo<'raw_packet> { pub struct PduRawWithInfo<'raw_packet> {
pdu_type: PduType, pdu_type: PduType,
file_directive_type: Option<FileDirectiveType>, file_directive_type: Option<FileDirectiveType>,
packet_len: usize, packet_len: usize,
@ -776,7 +776,7 @@ pub fn determine_packet_target(raw_pdu: &[u8]) -> Result<PacketTarget, PduError>
Ok(packet_target) Ok(packet_target)
} }
impl<'raw> PacketInfo<'raw> { impl<'raw> PduRawWithInfo<'raw> {
pub fn new(raw_packet: &'raw [u8]) -> Result<Self, PduError> { pub fn new(raw_packet: &'raw [u8]) -> Result<Self, PduError> {
let (pdu_header, header_len) = PduHeader::from_bytes(raw_packet)?; let (pdu_header, header_len) = PduHeader::from_bytes(raw_packet)?;
if pdu_header.pdu_type() == PduType::FileData { if pdu_header.pdu_type() == PduType::FileData {
@ -813,7 +813,7 @@ impl<'raw> PacketInfo<'raw> {
} }
} }
impl PduProvider for PacketInfo<'_> { impl PduProvider for PduRawWithInfo<'_> {
fn pdu_type(&self) -> PduType { fn pdu_type(&self) -> PduType {
self.pdu_type self.pdu_type
} }
@ -838,15 +838,20 @@ pub mod alloc_mod {
PduType, PduType,
}; };
use crate::{determine_packet_target, PacketTarget, PduProvider}; use crate::{determine_packet_target, PacketTarget, PduProvider, PduRawWithInfo};
pub struct PduWithInfo { #[derive(Debug, PartialEq, Eq, Clone)]
pub struct PduOwnedWithInfo {
pub pdu_type: PduType, pub pdu_type: PduType,
pub file_directive_type: Option<FileDirectiveType>, pub file_directive_type: Option<FileDirectiveType>,
pub pdu: alloc::vec::Vec<u8>, pub pdu: alloc::vec::Vec<u8>,
} }
impl PduWithInfo { impl PduOwnedWithInfo {
pub fn new_from_raw_packet(raw_packet: &[u8]) -> Result<Self, PduError> {
Ok(PduRawWithInfo::new(raw_packet)?.into())
}
pub fn new( pub fn new(
pdu_type: PduType, pdu_type: PduType,
file_directive_type: Option<FileDirectiveType>, file_directive_type: Option<FileDirectiveType>,
@ -860,7 +865,17 @@ pub mod alloc_mod {
} }
} }
impl PduProvider for PduWithInfo { impl From<PduRawWithInfo<'_>> for PduOwnedWithInfo {
fn from(value: PduRawWithInfo) -> Self {
Self::new(
value.pdu_type(),
value.file_directive_type(),
value.raw_packet().to_vec(),
)
}
}
impl PduProvider for PduOwnedWithInfo {
fn pdu_type(&self) -> PduType { fn pdu_type(&self) -> PduType {
self.pdu_type self.pdu_type
} }
@ -1191,7 +1206,7 @@ pub(crate) mod tests {
.write_to_bytes(&mut buf) .write_to_bytes(&mut buf)
.expect("writing metadata PDU failed"); .expect("writing metadata PDU failed");
let packet_info = PacketInfo::new(&buf).expect("creating packet info failed"); let packet_info = PduRawWithInfo::new(&buf).expect("creating packet info failed");
assert_eq!(packet_info.pdu_type(), PduType::FileDirective); assert_eq!(packet_info.pdu_type(), PduType::FileDirective);
assert!(packet_info.file_directive_type().is_some()); assert!(packet_info.file_directive_type().is_some());
assert_eq!( assert_eq!(
@ -1216,7 +1231,7 @@ pub(crate) mod tests {
file_data_pdu file_data_pdu
.write_to_bytes(&mut buf) .write_to_bytes(&mut buf)
.expect("writing file data PDU failed"); .expect("writing file data PDU failed");
let packet_info = PacketInfo::new(&buf).expect("creating packet info failed"); let packet_info = PduRawWithInfo::new(&buf).expect("creating packet info failed");
assert_eq!( assert_eq!(
packet_info.raw_packet(), packet_info.raw_packet(),
&buf[0..file_data_pdu.len_written()] &buf[0..file_data_pdu.len_written()]
@ -1237,7 +1252,7 @@ pub(crate) mod tests {
eof_pdu eof_pdu
.write_to_bytes(&mut buf) .write_to_bytes(&mut buf)
.expect("writing file data PDU failed"); .expect("writing file data PDU failed");
let packet_info = PacketInfo::new(&buf).expect("creating packet info failed"); let packet_info = PduRawWithInfo::new(&buf).expect("creating packet info failed");
assert_eq!(packet_info.pdu_type(), PduType::FileDirective); assert_eq!(packet_info.pdu_type(), PduType::FileDirective);
assert!(packet_info.file_directive_type().is_some()); assert!(packet_info.file_directive_type().is_some());
assert_eq!(packet_info.raw_packet(), &buf[0..eof_pdu.len_written()]); assert_eq!(packet_info.raw_packet(), &buf[0..eof_pdu.len_written()]);

4
src/source.rs
View File

@ -816,7 +816,7 @@ mod tests {
filestore::NativeFilestore, filestore::NativeFilestore,
request::PutRequestOwned, request::PutRequestOwned,
tests::{basic_remote_cfg_table, SentPdu, TestCfdpSender, TestCfdpUser, TestFaultHandler}, tests::{basic_remote_cfg_table, SentPdu, TestCfdpSender, TestCfdpUser, TestFaultHandler},
FaultHandler, IndicationConfig, PacketInfo, StdRemoteEntityConfigProvider, CRC_32, FaultHandler, IndicationConfig, PduRawWithInfo, StdRemoteEntityConfigProvider, CRC_32,
}; };
use spacepackets::seq_count::SeqCountProviderSimple; use spacepackets::seq_count::SeqCountProviderSimple;
@ -1139,7 +1139,7 @@ mod tests {
FileStatus::Retained, FileStatus::Retained,
); );
let finished_pdu_vec = finished_pdu.to_vec().unwrap(); let finished_pdu_vec = finished_pdu.to_vec().unwrap();
let packet_info = PacketInfo::new(&finished_pdu_vec).unwrap(); let packet_info = PduRawWithInfo::new(&finished_pdu_vec).unwrap();
self.handler self.handler
.state_machine(user, Some(&packet_info)) .state_machine(user, Some(&packet_info))
.unwrap(); .unwrap();

6
tests/end-to-end.rs
View File

@ -13,7 +13,7 @@ use cfdp::{
request::{PutRequestOwned, StaticPutRequestCacher}, request::{PutRequestOwned, StaticPutRequestCacher},
source::SourceHandler, source::SourceHandler,
user::{CfdpUser, FileSegmentRecvdParams, MetadataReceivedParams, TransactionFinishedParams}, user::{CfdpUser, FileSegmentRecvdParams, MetadataReceivedParams, TransactionFinishedParams},
EntityType, IndicationConfig, LocalEntityConfig, PduWithInfo, RemoteEntityConfig, EntityType, IndicationConfig, LocalEntityConfig, PduOwnedWithInfo, RemoteEntityConfig,
StdCheckTimerCreator, TransactionId, UserFaultHookProvider, StdCheckTimerCreator, TransactionId, UserFaultHookProvider,
}; };
use spacepackets::{ use spacepackets::{
@ -183,8 +183,8 @@ fn end_to_end_test(with_closure: bool) {
IndicationConfig::default(), IndicationConfig::default(),
ExampleFaultHandler::default(), ExampleFaultHandler::default(),
); );
let (source_tx, source_rx) = mpsc::channel::<PduWithInfo>(); let (source_tx, source_rx) = mpsc::channel::<PduOwnedWithInfo>();
let (dest_tx, dest_rx) = mpsc::channel::<PduWithInfo>(); let (dest_tx, dest_rx) = mpsc::channel::<PduOwnedWithInfo>();
let put_request_cacher = StaticPutRequestCacher::new(2048); let put_request_cacher = StaticPutRequestCacher::new(2048);
let remote_cfg_of_dest = RemoteEntityConfig::new_with_default_values( let remote_cfg_of_dest = RemoteEntityConfig::new_with_default_values(
REMOTE_ID.into(), REMOTE_ID.into(),