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

rework TCP spacepackets validation process
Some checks failed
Rust/sat-rs/pipeline/pr-main There was a failure building this commit
Details

Browse Source
This commit is contained in:
Robin Müller 2024-04-15 18:43:11 +02:00
parent 31b208737e
commit ca6e4816ab
Signed by: muellerr
GPG Key ID: A649FB78196E3849
8 changed files with 206 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
satrs-example/src/interface/tcp.rs
View File

@ -7,14 +7,35 @@ use std::{
use log::{info, warn}; use log::{info, warn};
use satrs::{ use satrs::{
encoding::ccsds::{SpacePacketValidation, SpacePacketValidator},
hal::std::tcp_server::{HandledConnectionHandler, ServerConfig, TcpSpacepacketsServer}, hal::std::tcp_server::{HandledConnectionHandler, ServerConfig, TcpSpacepacketsServer},
spacepackets::PacketId, spacepackets::{CcsdsPacket, PacketId},
tmtc::{PacketSenderRaw, PacketSource}, tmtc::{PacketSenderRaw, PacketSource},
}; };
#[derive(Default)] #[derive(Default)]
pub struct ConnectionFinishedHandler {} pub struct ConnectionFinishedHandler {}
pub struct SimplePacketValidator {
pub valid_ids: HashSet<PacketId>,
}
impl SpacePacketValidator for SimplePacketValidator {
fn validate(
&self,
sp_header: &satrs::spacepackets::SpHeader,
_raw_buf: &[u8],
) -> satrs::encoding::ccsds::SpacePacketValidation {
if self.valid_ids.contains(&sp_header.packet_id()) {
return SpacePacketValidation::Valid;
}
log::warn!("ignoring space packet with header {:?}", sp_header);
// We could perform a CRC check.. but lets keep this simple and assume that TCP ensures
// data integrity.
SpacePacketValidation::Ignore
}
}
impl HandledConnectionHandler for ConnectionFinishedHandler { impl HandledConnectionHandler for ConnectionFinishedHandler {
fn handled_connection(&mut self, info: satrs::hal::std::tcp_server::HandledConnectionInfo) { fn handled_connection(&mut self, info: satrs::hal::std::tcp_server::HandledConnectionInfo) {
info!( info!(
@ -83,7 +104,7 @@ impl PacketSource for SyncTcpTmSource {
pub type TcpServer<ReceivesTc, SendError> = TcpSpacepacketsServer< pub type TcpServer<ReceivesTc, SendError> = TcpSpacepacketsServer<
SyncTcpTmSource, SyncTcpTmSource,
ReceivesTc, ReceivesTc,
HashSet<PacketId>, SimplePacketValidator,
ConnectionFinishedHandler, ConnectionFinishedHandler,
(), (),
SendError, SendError,
@ -101,14 +122,14 @@ impl<TcSender: PacketSenderRaw<Error = SendError>, SendError: Debug + 'static>
cfg: ServerConfig, cfg: ServerConfig,
tm_source: SyncTcpTmSource, tm_source: SyncTcpTmSource,
tc_sender: TcSender, tc_sender: TcSender,
packet_id_lookup: HashSet<PacketId>, valid_ids: HashSet<PacketId>,
) -> Result<Self, std::io::Error> { ) -> Result<Self, std::io::Error> {
Ok(Self( Ok(Self(
TcpSpacepacketsServer::new( TcpSpacepacketsServer::new(
cfg, cfg,
tm_source, tm_source,
tc_sender, tc_sender,
packet_id_lookup, SimplePacketValidator { valid_ids },
ConnectionFinishedHandler::default(), ConnectionFinishedHandler::default(),
None, None,
)?, )?,

131
satrs/src/encoding/ccsds.rs
View File

@ -1,20 +1,45 @@
use crate::{tmtc::PacketSenderRaw, ComponentId, ValidatorU16Id}; use spacepackets::{CcsdsPacket, SpHeader};
use crate::{tmtc::PacketSenderRaw, ComponentId};
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum SpacePacketValidation {
Valid,
/// The space packet header can be assumed to have a valid format, but the packet should
/// be ignored.
Ignore,
/// The space packet or space packet header has an invalid format, for example a CRC check
/// failed. In that case, the parser loses the packet synchronization and needs to check for
/// the start of a new space packet header start again. The space packet header
/// [spacepackets::PacketId] can be used as a synchronization marker to detect the validity
/// of a packet.
Invalid,
}
/// Simple trait to allow user code to check the validity of a space packet.
pub trait SpacePacketValidator {
fn validate(&self, sp_header: &SpHeader, raw_buf: &[u8]) -> SpacePacketValidation;
}
/// This function parses a given buffer for tightly packed CCSDS space packets. It uses the /// This function parses a given buffer for tightly packed CCSDS space packets. It uses the
/// [spacepackets::PacketId] field of the CCSDS packets to detect the start of a CCSDS space packet /// [spacepackets::SpHeader] of the CCSDS packets and a user provided [SpacePacketValidator]
/// and then uses the length field of the packet to extract CCSDS packets. /// to check whether a received space packet is relevant for processing.
/// ///
/// This function is also able to deal with broken tail packets at the end as long a the parser /// This function is also able to deal with broken tail packets at the end as long a the parser
/// can read the full 7 bytes which constitue a space packet header plus one byte minimal size. /// can read the full 7 bytes which constitue a space packet header plus one byte minimal size.
/// If broken tail packets are detected, they are moved to the front of the buffer, and the write /// If broken tail packets are detected, they are moved to the front of the buffer, and the write
/// index for future write operations will be written to the `next_write_idx` argument. /// index for future write operations will be written to the `next_write_idx` argument.
/// ///
/// The parser will forward all packets which were decoded successfully to the given /// The parser will forward all packets for which the user provided [SpacePacketValidator] returned
/// `packet_sender` and return the number of packets found. If the [PacketSenderRaw::send_packet] /// [SpacePacketValidation::Valid] to the given `packet_sender` and return the number of packets
/// calls fails, the error will be returned. /// found. If the [PacketSenderRaw::send_packet] calls fails, the error will be returned.
///
/// If the user provided [SpacePacketValidator] returns [SpacePacketValidation::Invalid], the
/// parser assumes that the synchronization is lost and tries to find the start of a new space
/// packet header by scanning all the following bytes.
pub fn parse_buffer_for_ccsds_space_packets<SendError>( pub fn parse_buffer_for_ccsds_space_packets<SendError>(
buf: &mut [u8], buf: &mut [u8],
packet_id_validator: &(impl ValidatorU16Id + ?Sized), packet_validator: &(impl SpacePacketValidator + ?Sized),
sender_id: ComponentId, sender_id: ComponentId,
packet_sender: &(impl PacketSenderRaw<Error = SendError> + ?Sized), packet_sender: &(impl PacketSenderRaw<Error = SendError> + ?Sized),
next_write_idx: &mut usize, next_write_idx: &mut usize,
@ -27,28 +52,33 @@ pub fn parse_buffer_for_ccsds_space_packets<SendError>(
if current_idx + 7 >= buf.len() { if current_idx + 7 >= buf.len() {
break; break;
} }
let packet_id = u16::from_be_bytes(buf[current_idx..current_idx + 2].try_into().unwrap()); let sp_header = SpHeader::from_be_bytes(&buf[current_idx..]).unwrap().0;
if packet_id_validator.validate(packet_id) { // let packet_id = u16::from_be_bytes(buf[current_idx..current_idx + 2].try_into().unwrap());
let length_field = match packet_validator.validate(&sp_header, &buf[current_idx..]) {
u16::from_be_bytes(buf[current_idx + 4..current_idx + 6].try_into().unwrap()); SpacePacketValidation::Valid => {
let packet_size = length_field + 7; let packet_size = sp_header.total_len();
if (current_idx + packet_size as usize) <= buf_len { if (current_idx + packet_size) <= buf_len {
packet_sender.send_packet( packet_sender
sender_id, .send_packet(sender_id, &buf[current_idx..current_idx + packet_size])?;
&buf[current_idx..current_idx + packet_size as usize], packets_found += 1;
)?; } else {
packets_found += 1; // Move packet to start of buffer if applicable.
} else { if current_idx > 0 {
// Move packet to start of buffer if applicable. buf.copy_within(current_idx.., 0);
if current_idx > 0 { *next_write_idx = buf.len() - current_idx;
buf.copy_within(current_idx.., 0); }
*next_write_idx = buf.len() - current_idx;
} }
current_idx += packet_size;
continue;
}
SpacePacketValidation::Ignore => {
current_idx += sp_header.total_len();
}
// We might have lost sync. Try to find the start of a new space packet header.
SpacePacketValidation::Invalid => {
current_idx += 1;
} }
current_idx += packet_size as usize;
continue;
} }
current_idx += 1;
} }
Ok(packets_found) Ok(packets_found)
} }
@ -57,12 +87,14 @@ pub fn parse_buffer_for_ccsds_space_packets<SendError>(
mod tests { mod tests {
use spacepackets::{ use spacepackets::{
ecss::{tc::PusTcCreator, WritablePusPacket}, ecss::{tc::PusTcCreator, WritablePusPacket},
PacketId, SpHeader, CcsdsPacket, PacketId, SpHeader,
}; };
use crate::{encoding::tests::TcCacher, ComponentId}; use crate::{encoding::tests::TcCacher, ComponentId};
use super::parse_buffer_for_ccsds_space_packets; use super::{
parse_buffer_for_ccsds_space_packets, SpacePacketValidation, SpacePacketValidator,
};
const PARSER_ID: ComponentId = 0x05; const PARSER_ID: ComponentId = 0x05;
const TEST_APID_0: u16 = 0x02; const TEST_APID_0: u16 = 0x02;
@ -70,6 +102,30 @@ mod tests {
const TEST_PACKET_ID_0: PacketId = PacketId::new_for_tc(true, TEST_APID_0); const TEST_PACKET_ID_0: PacketId = PacketId::new_for_tc(true, TEST_APID_0);
const TEST_PACKET_ID_1: PacketId = PacketId::new_for_tc(true, TEST_APID_1); const TEST_PACKET_ID_1: PacketId = PacketId::new_for_tc(true, TEST_APID_1);
#[derive(Default)]
struct SimpleVerificator {
pub enable_second_id: bool,
}
impl SimpleVerificator {
pub fn new_with_second_id() -> Self {
Self {
enable_second_id: true,
}
}
}
impl SpacePacketValidator for SimpleVerificator {
fn validate(&self, sp_header: &SpHeader, _raw_buf: &[u8]) -> super::SpacePacketValidation {
if sp_header.packet_id() == TEST_PACKET_ID_0
|| (self.enable_second_id && sp_header.packet_id() == TEST_PACKET_ID_1)
{
return SpacePacketValidation::Valid;
}
SpacePacketValidation::Ignore
}
}
#[test] #[test]
fn test_basic() { fn test_basic() {
let sph = SpHeader::new_from_apid(TEST_APID_0); let sph = SpHeader::new_from_apid(TEST_APID_0);
@ -78,12 +134,11 @@ mod tests {
let packet_len = ping_tc let packet_len = ping_tc
.write_to_bytes(&mut buffer) .write_to_bytes(&mut buffer)
.expect("writing packet failed"); .expect("writing packet failed");
let valid_packet_ids = [TEST_PACKET_ID_0];
let tc_cacher = TcCacher::default(); let tc_cacher = TcCacher::default();
let mut next_write_idx = 0; let mut next_write_idx = 0;
let parse_result = parse_buffer_for_ccsds_space_packets( let parse_result = parse_buffer_for_ccsds_space_packets(
&mut buffer, &mut buffer,
valid_packet_ids.as_slice(), &SimpleVerificator::default(),
PARSER_ID, PARSER_ID,
&tc_cacher, &tc_cacher,
&mut next_write_idx, &mut next_write_idx,
@ -110,12 +165,11 @@ mod tests {
let packet_len_action = action_tc let packet_len_action = action_tc
.write_to_bytes(&mut buffer[packet_len_ping..]) .write_to_bytes(&mut buffer[packet_len_ping..])
.expect("writing packet failed"); .expect("writing packet failed");
let valid_packet_ids = [TEST_PACKET_ID_0];
let tc_cacher = TcCacher::default(); let tc_cacher = TcCacher::default();
let mut next_write_idx = 0; let mut next_write_idx = 0;
let parse_result = parse_buffer_for_ccsds_space_packets( let parse_result = parse_buffer_for_ccsds_space_packets(
&mut buffer, &mut buffer,
valid_packet_ids.as_slice(), &SimpleVerificator::default(),
PARSER_ID, PARSER_ID,
&tc_cacher, &tc_cacher,
&mut next_write_idx, &mut next_write_idx,
@ -149,12 +203,12 @@ mod tests {
let packet_len_action = action_tc let packet_len_action = action_tc
.write_to_bytes(&mut buffer[packet_len_ping..]) .write_to_bytes(&mut buffer[packet_len_ping..])
.expect("writing packet failed"); .expect("writing packet failed");
let valid_packet_ids = [TEST_PACKET_ID_0, TEST_PACKET_ID_1];
let tc_cacher = TcCacher::default(); let tc_cacher = TcCacher::default();
let mut next_write_idx = 0; let mut next_write_idx = 0;
let verificator = SimpleVerificator::new_with_second_id();
let parse_result = parse_buffer_for_ccsds_space_packets( let parse_result = parse_buffer_for_ccsds_space_packets(
&mut buffer, &mut buffer,
valid_packet_ids.as_slice(), &verificator,
PARSER_ID, PARSER_ID,
&tc_cacher, &tc_cacher,
&mut next_write_idx, &mut next_write_idx,
@ -186,12 +240,12 @@ mod tests {
let packet_len_action = action_tc let packet_len_action = action_tc
.write_to_bytes(&mut buffer[packet_len_ping..]) .write_to_bytes(&mut buffer[packet_len_ping..])
.expect("writing packet failed"); .expect("writing packet failed");
let valid_packet_ids = [TEST_PACKET_ID_0, TEST_PACKET_ID_1];
let tc_cacher = TcCacher::default(); let tc_cacher = TcCacher::default();
let mut next_write_idx = 0; let mut next_write_idx = 0;
let verificator = SimpleVerificator::new_with_second_id();
let parse_result = parse_buffer_for_ccsds_space_packets( let parse_result = parse_buffer_for_ccsds_space_packets(
&mut buffer[..packet_len_ping + packet_len_action - 4], &mut buffer[..packet_len_ping + packet_len_action - 4],
valid_packet_ids.as_slice(), &verificator,
PARSER_ID, PARSER_ID,
&tc_cacher, &tc_cacher,
&mut next_write_idx, &mut next_write_idx,
@ -215,12 +269,13 @@ mod tests {
let packet_len_ping = ping_tc let packet_len_ping = ping_tc
.write_to_bytes(&mut buffer) .write_to_bytes(&mut buffer)
.expect("writing packet failed"); .expect("writing packet failed");
let valid_packet_ids = [TEST_PACKET_ID_0, TEST_PACKET_ID_1];
let tc_cacher = TcCacher::default(); let tc_cacher = TcCacher::default();
let verificator = SimpleVerificator::new_with_second_id();
let mut next_write_idx = 0; let mut next_write_idx = 0;
let parse_result = parse_buffer_for_ccsds_space_packets( let parse_result = parse_buffer_for_ccsds_space_packets(
&mut buffer[..packet_len_ping - 4], &mut buffer[..packet_len_ping - 4],
valid_packet_ids.as_slice(), &verificator,
PARSER_ID, PARSER_ID,
&tc_cacher, &tc_cacher,
&mut next_write_idx, &mut next_write_idx,

10
satrs/src/hal/std/tcp_cobs_server.rs
View File

@ -181,7 +181,7 @@ impl<
/// useful if using the port number 0 for OS auto-assignment. /// useful if using the port number 0 for OS auto-assignment.
pub fn local_addr(&self) -> std::io::Result<SocketAddr>; pub fn local_addr(&self) -> std::io::Result<SocketAddr>;
/// Delegation to the [TcpTmtcGenericServer::handle_next_connection] call. /// Delegation to the [TcpTmtcGenericServer::handle_all_connections] call.
pub fn handle_all_connections( pub fn handle_all_connections(
&mut self, &mut self,
poll_duration: Option<Duration>, poll_duration: Option<Duration>,
@ -274,7 +274,7 @@ mod tests {
let set_if_done = conn_handled.clone(); let set_if_done = conn_handled.clone();
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
thread::spawn(move || { thread::spawn(move || {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(100))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(100)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }
@ -330,7 +330,7 @@ mod tests {
let set_if_done = conn_handled.clone(); let set_if_done = conn_handled.clone();
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
thread::spawn(move || { thread::spawn(move || {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(100))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(100)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }
@ -436,7 +436,7 @@ mod tests {
let start = Instant::now(); let start = Instant::now();
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
let thread_jh = thread::spawn(move || loop { let thread_jh = thread::spawn(move || loop {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(20))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(20)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }
@ -470,7 +470,7 @@ mod tests {
let start = Instant::now(); let start = Instant::now();
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
let thread_jh = thread::spawn(move || loop { let thread_jh = thread::spawn(move || loop {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(20))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(20)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }

4
satrs/src/hal/std/tcp_server.rs
View File

@ -19,9 +19,7 @@ use thiserror::Error;
// Re-export the TMTC in COBS server. // Re-export the TMTC in COBS server.
pub use crate::hal::std::tcp_cobs_server::{CobsTcParser, CobsTmSender, TcpTmtcInCobsServer}; pub use crate::hal::std::tcp_cobs_server::{CobsTcParser, CobsTmSender, TcpTmtcInCobsServer};
pub use crate::hal::std::tcp_spacepackets_server::{ pub use crate::hal::std::tcp_spacepackets_server::{SpacepacketsTmSender, TcpSpacepacketsServer};
SpacepacketsTcParser, SpacepacketsTmSender, TcpSpacepacketsServer,
};
/// Configuration struct for the generic TCP TMTC server /// Configuration struct for the generic TCP TMTC server
/// ///

89
satrs/src/hal/std/tcp_spacepackets_server.rs
View File

@ -5,9 +5,9 @@ use mio::net::{TcpListener, TcpStream};
use std::{io::Write, net::SocketAddr}; use std::{io::Write, net::SocketAddr};
use crate::{ use crate::{
encoding::parse_buffer_for_ccsds_space_packets, encoding::{ccsds::SpacePacketValidator, parse_buffer_for_ccsds_space_packets},
tmtc::{PacketSenderRaw, PacketSource}, tmtc::{PacketSenderRaw, PacketSource},
ComponentId, ValidatorU16Id, ComponentId,
}; };
use super::tcp_server::{ use super::tcp_server::{
@ -15,20 +15,7 @@ use super::tcp_server::{
TcpTmSender, TcpTmtcError, TcpTmtcGenericServer, TcpTmSender, TcpTmtcError, TcpTmtcGenericServer,
}; };
/// Concrete [TcpTcParser] implementation for the [TcpSpacepacketsServer]. impl<T: SpacePacketValidator, TmError, TcError: 'static> TcpTcParser<TmError, TcError> for T {
pub struct SpacepacketsTcParser<PacketIdChecker: ValidatorU16Id> {
packet_id_lookup: PacketIdChecker,
}
impl<PacketIdChecker: ValidatorU16Id> SpacepacketsTcParser<PacketIdChecker> {
pub fn new(packet_id_lookup: PacketIdChecker) -> Self {
Self { packet_id_lookup }
}
}
impl<PacketIdChecker: ValidatorU16Id, TmError, TcError: 'static> TcpTcParser<TmError, TcError>
for SpacepacketsTcParser<PacketIdChecker>
{
fn handle_tc_parsing( fn handle_tc_parsing(
&mut self, &mut self,
tc_buffer: &mut [u8], tc_buffer: &mut [u8],
@ -41,7 +28,7 @@ impl<PacketIdChecker: ValidatorU16Id, TmError, TcError: 'static> TcpTcParser<TmE
// Reader vec full, need to parse for packets. // Reader vec full, need to parse for packets.
conn_result.num_received_tcs += parse_buffer_for_ccsds_space_packets( conn_result.num_received_tcs += parse_buffer_for_ccsds_space_packets(
&mut tc_buffer[..current_write_idx], &mut tc_buffer[..current_write_idx],
&self.packet_id_lookup, self,
sender_id, sender_id,
tc_sender, tc_sender,
next_write_idx, next_write_idx,
@ -87,9 +74,9 @@ impl<TmError, TcError> TcpTmSender<TmError, TcError> for SpacepacketsTmSender {
/// ///
/// This serves only works if /// This serves only works if
/// [CCSDS 133.0-B-2 space packets](https://public.ccsds.org/Pubs/133x0b2e1.pdf) are the only /// [CCSDS 133.0-B-2 space packets](https://public.ccsds.org/Pubs/133x0b2e1.pdf) are the only
/// packet type being exchanged. It uses the CCSDS [spacepackets::PacketId] as the packet delimiter /// packet type being exchanged. It uses the CCSDS space packet header [spacepackets::SpHeader] and
/// and start marker when parsing for packets. The user specifies a set of expected /// a user specified [SpacePacketValidator] to determine the space packets relevant for further
/// [spacepackets::PacketId]s as part of the server configuration for that purpose. /// processing.
/// ///
/// ## Example /// ## Example
/// The [TCP server integration tests](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs/tests/tcp_servers.rs) /// The [TCP server integration tests](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs/tests/tcp_servers.rs)
@ -97,7 +84,7 @@ impl<TmError, TcError> TcpTmSender<TmError, TcError> for SpacepacketsTmSender {
pub struct TcpSpacepacketsServer< pub struct TcpSpacepacketsServer<
TmSource: PacketSource<Error = TmError>, TmSource: PacketSource<Error = TmError>,
TcSender: PacketSenderRaw<Error = SendError>, TcSender: PacketSenderRaw<Error = SendError>,
PacketIdChecker: ValidatorU16Id, Validator: SpacePacketValidator,
HandledConnection: HandledConnectionHandler, HandledConnection: HandledConnectionHandler,
TmError, TmError,
SendError: 'static, SendError: 'static,
@ -106,7 +93,7 @@ pub struct TcpSpacepacketsServer<
TmSource, TmSource,
TcSender, TcSender,
SpacepacketsTmSender, SpacepacketsTmSender,
SpacepacketsTcParser<PacketIdChecker>, Validator,
HandledConnection, HandledConnection,
TmError, TmError,
SendError, SendError,
@ -116,19 +103,11 @@ pub struct TcpSpacepacketsServer<
impl< impl<
TmSource: PacketSource<Error = TmError>, TmSource: PacketSource<Error = TmError>,
TcReceiver: PacketSenderRaw<Error = TcError>, TcReceiver: PacketSenderRaw<Error = TcError>,
PacketIdChecker: ValidatorU16Id, Validator: SpacePacketValidator,
HandledConnection: HandledConnectionHandler, HandledConnection: HandledConnectionHandler,
TmError: 'static, TmError: 'static,
TcError: 'static, TcError: 'static,
> > TcpSpacepacketsServer<TmSource, TcReceiver, Validator, HandledConnection, TmError, TcError>
TcpSpacepacketsServer<
TmSource,
TcReceiver,
PacketIdChecker,
HandledConnection,
TmError,
TcError,
>
{ {
/// ///
/// ## Parameter /// ## Parameter
@ -144,14 +123,14 @@ impl<
cfg: ServerConfig, cfg: ServerConfig,
tm_source: TmSource, tm_source: TmSource,
tc_receiver: TcReceiver, tc_receiver: TcReceiver,
packet_id_checker: PacketIdChecker, validator: Validator,
handled_connection: HandledConnection, handled_connection: HandledConnection,
stop_signal: Option<Arc<AtomicBool>>, stop_signal: Option<Arc<AtomicBool>>,
) -> Result<Self, std::io::Error> { ) -> Result<Self, std::io::Error> {
Ok(Self { Ok(Self {
generic_server: TcpTmtcGenericServer::new( generic_server: TcpTmtcGenericServer::new(
cfg, cfg,
SpacepacketsTcParser::new(packet_id_checker), validator,
SpacepacketsTmSender::default(), SpacepacketsTmSender::default(),
tm_source, tm_source,
tc_receiver, tc_receiver,
@ -169,7 +148,7 @@ impl<
/// useful if using the port number 0 for OS auto-assignment. /// useful if using the port number 0 for OS auto-assignment.
pub fn local_addr(&self) -> std::io::Result<SocketAddr>; pub fn local_addr(&self) -> std::io::Result<SocketAddr>;
/// Delegation to the [TcpTmtcGenericServer::handle_next_connection] call. /// Delegation to the [TcpTmtcGenericServer::handle_all_connections] call.
pub fn handle_all_connections( pub fn handle_all_connections(
&mut self, &mut self,
poll_timeout: Option<Duration> poll_timeout: Option<Duration>
@ -197,10 +176,11 @@ mod tests {
use hashbrown::HashSet; use hashbrown::HashSet;
use spacepackets::{ use spacepackets::{
ecss::{tc::PusTcCreator, WritablePusPacket}, ecss::{tc::PusTcCreator, WritablePusPacket},
PacketId, SpHeader, CcsdsPacket, PacketId, SpHeader,
}; };
use crate::{ use crate::{
encoding::ccsds::{SpacePacketValidation, SpacePacketValidator},
hal::std::tcp_server::{ hal::std::tcp_server::{
tests::{ConnectionFinishedHandler, SyncTmSource}, tests::{ConnectionFinishedHandler, SyncTmSource},
ConnectionResult, ServerConfig, ConnectionResult, ServerConfig,
@ -218,16 +198,29 @@ mod tests {
const TEST_APID_1: u16 = 0x10; const TEST_APID_1: u16 = 0x10;
const TEST_PACKET_ID_1: PacketId = PacketId::new_for_tc(true, TEST_APID_1); const TEST_PACKET_ID_1: PacketId = PacketId::new_for_tc(true, TEST_APID_1);
#[derive(Default)]
pub struct SimpleValidator(pub HashSet<PacketId>);
impl SpacePacketValidator for SimpleValidator {
fn validate(&self, sp_header: &SpHeader, _raw_buf: &[u8]) -> SpacePacketValidation {
if self.0.contains(&sp_header.packet_id()) {
return SpacePacketValidation::Valid;
}
// Simple case: Assume that the interface always contains valid space packets.
SpacePacketValidation::Ignore
}
}
fn generic_tmtc_server( fn generic_tmtc_server(
addr: &SocketAddr, addr: &SocketAddr,
tc_sender: mpsc::Sender<PacketAsVec>, tc_sender: mpsc::Sender<PacketAsVec>,
tm_source: SyncTmSource, tm_source: SyncTmSource,
packet_id_lookup: HashSet<PacketId>, validator: SimpleValidator,
stop_signal: Option<Arc<AtomicBool>>, stop_signal: Option<Arc<AtomicBool>>,
) -> TcpSpacepacketsServer< ) -> TcpSpacepacketsServer<
SyncTmSource, SyncTmSource,
mpsc::Sender<PacketAsVec>, mpsc::Sender<PacketAsVec>,
HashSet<PacketId>, SimpleValidator,
ConnectionFinishedHandler, ConnectionFinishedHandler,
(), (),
GenericSendError, GenericSendError,
@ -236,7 +229,7 @@ mod tests {
ServerConfig::new(TCP_SERVER_ID, *addr, Duration::from_millis(2), 1024, 1024), ServerConfig::new(TCP_SERVER_ID, *addr, Duration::from_millis(2), 1024, 1024),
tm_source, tm_source,
tc_sender, tc_sender,
packet_id_lookup, validator,
ConnectionFinishedHandler::default(), ConnectionFinishedHandler::default(),
stop_signal, stop_signal,
) )
@ -248,13 +241,13 @@ mod tests {
let auto_port_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0); let auto_port_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0);
let (tc_sender, tc_receiver) = mpsc::channel(); let (tc_sender, tc_receiver) = mpsc::channel();
let tm_source = SyncTmSource::default(); let tm_source = SyncTmSource::default();
let mut packet_id_lookup = HashSet::new(); let mut validator = SimpleValidator::default();
packet_id_lookup.insert(TEST_PACKET_ID_0); validator.0.insert(TEST_PACKET_ID_0);
let mut tcp_server = generic_tmtc_server( let mut tcp_server = generic_tmtc_server(
&auto_port_addr, &auto_port_addr,
tc_sender.clone(), tc_sender.clone(),
tm_source, tm_source,
packet_id_lookup, validator,
None, None,
); );
let dest_addr = tcp_server let dest_addr = tcp_server
@ -264,7 +257,7 @@ mod tests {
let set_if_done = conn_handled.clone(); let set_if_done = conn_handled.clone();
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
thread::spawn(move || { thread::spawn(move || {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(100))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(100)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }
@ -323,14 +316,14 @@ mod tests {
tm_source.add_tm(&tm_1); tm_source.add_tm(&tm_1);
// Set up server // Set up server
let mut packet_id_lookup = HashSet::new(); let mut validator = SimpleValidator::default();
packet_id_lookup.insert(TEST_PACKET_ID_0); validator.0.insert(TEST_PACKET_ID_0);
packet_id_lookup.insert(TEST_PACKET_ID_1); validator.0.insert(TEST_PACKET_ID_1);
let mut tcp_server = generic_tmtc_server( let mut tcp_server = generic_tmtc_server(
&auto_port_addr, &auto_port_addr,
tc_sender.clone(), tc_sender.clone(),
tm_source, tm_source,
packet_id_lookup, validator,
None, None,
); );
let dest_addr = tcp_server let dest_addr = tcp_server
@ -341,7 +334,7 @@ mod tests {
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
thread::spawn(move || { thread::spawn(move || {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(100))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(100)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }

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

@ -6,7 +6,9 @@ use crate::pool::{PoolAddr, PoolError};
use crate::pus::verification::{TcStateAccepted, TcStateToken, VerificationToken}; use crate::pus::verification::{TcStateAccepted, TcStateToken, VerificationToken};
use crate::queue::{GenericReceiveError, GenericSendError}; use crate::queue::{GenericReceiveError, GenericSendError};
use crate::request::{GenericMessage, MessageMetadata, RequestId}; use crate::request::{GenericMessage, MessageMetadata, RequestId};
use crate::tmtc::{PacketAsVec, PacketInPool}; #[cfg(feature = "alloc")]
use crate::tmtc::PacketAsVec;
use crate::tmtc::PacketInPool;
use crate::ComponentId; use crate::ComponentId;
use core::fmt::{Display, Formatter}; use core::fmt::{Display, Formatter};
use core::time::Duration; use core::time::Duration;

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

@ -172,17 +172,6 @@ where
} }
} }
/// Newtype wrapper around the [SharedStaticMemoryPool] to enable extension helper traits on
/// top of the regular shared memory pool API.
#[derive(Clone)]
pub struct SharedPacketPool(pub SharedStaticMemoryPool);
impl SharedPacketPool {
pub fn new(pool: &SharedStaticMemoryPool) -> Self {
Self(pool.clone())
}
}
/// Helper trait for any generic (static) store which allows storing raw or CCSDS packets. /// Helper trait for any generic (static) store which allows storing raw or CCSDS packets.
pub trait CcsdsPacketPool { pub trait CcsdsPacketPool {
fn add_ccsds_tc(&mut self, _: &SpHeader, tc_raw: &[u8]) -> Result<PoolAddr, PoolError> { fn add_ccsds_tc(&mut self, _: &SpHeader, tc_raw: &[u8]) -> Result<PoolAddr, PoolError> {
@ -283,6 +272,17 @@ pub mod std_mod {
use super::*; use super::*;
/// Newtype wrapper around the [SharedStaticMemoryPool] to enable extension helper traits on
/// top of the regular shared memory pool API.
#[derive(Clone)]
pub struct SharedPacketPool(pub SharedStaticMemoryPool);
impl SharedPacketPool {
pub fn new(pool: &SharedStaticMemoryPool) -> Self {
Self(pool.clone())
}
}
#[cfg(feature = "std")] #[cfg(feature = "std")]
impl PacketSenderRaw for mpsc::Sender<PacketAsVec> { impl PacketSenderRaw for mpsc::Sender<PacketAsVec> {
type Error = GenericSendError; type Error = GenericSendError;

33
satrs/tests/tcp_servers.rs
View File

@ -23,7 +23,10 @@ use std::{
use hashbrown::HashSet; use hashbrown::HashSet;
use satrs::{ use satrs::{
encoding::cobs::encode_packet_with_cobs, encoding::{
ccsds::{SpacePacketValidation, SpacePacketValidator},
cobs::encode_packet_with_cobs,
},
hal::std::tcp_server::{ hal::std::tcp_server::{
ConnectionResult, HandledConnectionHandler, HandledConnectionInfo, ServerConfig, ConnectionResult, HandledConnectionHandler, HandledConnectionInfo, ServerConfig,
TcpSpacepacketsServer, TcpTmtcInCobsServer, TcpSpacepacketsServer, TcpTmtcInCobsServer,
@ -33,7 +36,7 @@ use satrs::{
}; };
use spacepackets::{ use spacepackets::{
ecss::{tc::PusTcCreator, WritablePusPacket}, ecss::{tc::PusTcCreator, WritablePusPacket},
PacketId, SpHeader, CcsdsPacket, PacketId, SpHeader,
}; };
use std::{collections::VecDeque, sync::Arc, vec::Vec}; use std::{collections::VecDeque, sync::Arc, vec::Vec};
@ -130,7 +133,7 @@ fn test_cobs_server() {
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
thread::spawn(move || { thread::spawn(move || {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(400))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(400)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }
@ -192,6 +195,24 @@ fn test_cobs_server() {
const TEST_APID_0: u16 = 0x02; const TEST_APID_0: u16 = 0x02;
const TEST_PACKET_ID_0: PacketId = PacketId::new_for_tc(true, TEST_APID_0); const TEST_PACKET_ID_0: PacketId = PacketId::new_for_tc(true, TEST_APID_0);
#[derive(Default)]
pub struct SimpleVerificator {
pub valid_ids: HashSet<PacketId>,
}
impl SpacePacketValidator for SimpleVerificator {
fn validate(
&self,
sp_header: &SpHeader,
_raw_buf: &[u8],
) -> satrs::encoding::ccsds::SpacePacketValidation {
if self.valid_ids.contains(&sp_header.packet_id()) {
return SpacePacketValidation::Valid;
}
SpacePacketValidation::Ignore
}
}
#[test] #[test]
fn test_ccsds_server() { fn test_ccsds_server() {
let (tc_sender, tc_receiver) = mpsc::channel(); let (tc_sender, tc_receiver) = mpsc::channel();
@ -200,8 +221,8 @@ fn test_ccsds_server() {
let verif_tm = PusTcCreator::new_simple(sph, 1, 1, &[], true); let verif_tm = PusTcCreator::new_simple(sph, 1, 1, &[], true);
let tm_0 = verif_tm.to_vec().expect("tm generation failed"); let tm_0 = verif_tm.to_vec().expect("tm generation failed");
tm_source.add_tm(&tm_0); tm_source.add_tm(&tm_0);
let mut packet_id_lookup = HashSet::new(); let mut packet_id_lookup = SimpleVerificator::default();
packet_id_lookup.insert(TEST_PACKET_ID_0); packet_id_lookup.valid_ids.insert(TEST_PACKET_ID_0);
let mut tcp_server = TcpSpacepacketsServer::new( let mut tcp_server = TcpSpacepacketsServer::new(
ServerConfig::new( ServerConfig::new(
TCP_SERVER_ID, TCP_SERVER_ID,
@ -224,7 +245,7 @@ fn test_ccsds_server() {
let set_if_done = conn_handled.clone(); let set_if_done = conn_handled.clone();
// Call the connection handler in separate thread, does block. // Call the connection handler in separate thread, does block.
thread::spawn(move || { thread::spawn(move || {
let result = tcp_server.handle_next_connection(Some(Duration::from_millis(500))); let result = tcp_server.handle_all_connections(Some(Duration::from_millis(500)));
if result.is_err() { if result.is_err() {
panic!("handling connection failed: {:?}", result.unwrap_err()); panic!("handling connection failed: {:?}", result.unwrap_err());
} }