diff --git a/.idea/runConfigurations/Check.xml b/.idea/runConfigurations/Check.xml deleted file mode 100644 index 44c4c7f..0000000 --- a/.idea/runConfigurations/Check.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Clippy.xml b/.idea/runConfigurations/Clippy.xml deleted file mode 100644 index 8634d06..0000000 --- a/.idea/runConfigurations/Clippy.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Clippy_Fix.xml b/.idea/runConfigurations/Clippy_Fix.xml deleted file mode 100644 index 9722c38..0000000 --- a/.idea/runConfigurations/Clippy_Fix.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Docs.xml b/.idea/runConfigurations/Docs.xml deleted file mode 100644 index 43ef045..0000000 --- a/.idea/runConfigurations/Docs.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Doctest.xml b/.idea/runConfigurations/Doctest.xml deleted file mode 100644 index 73a8b79..0000000 --- a/.idea/runConfigurations/Doctest.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Examples.xml b/.idea/runConfigurations/Examples.xml deleted file mode 100644 index e63f661..0000000 --- a/.idea/runConfigurations/Examples.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Format.xml b/.idea/runConfigurations/Format.xml deleted file mode 100644 index e6f696d..0000000 --- a/.idea/runConfigurations/Format.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Run.xml b/.idea/runConfigurations/Run.xml deleted file mode 100644 index 9f99c6e..0000000 --- a/.idea/runConfigurations/Run.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Run_obsw_example.xml b/.idea/runConfigurations/Run_obsw_example.xml deleted file mode 100644 index df610ba..0000000 --- a/.idea/runConfigurations/Run_obsw_example.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Run_obsw_simple_client.xml b/.idea/runConfigurations/Run_obsw_simple_client.xml deleted file mode 100644 index 4679680..0000000 --- a/.idea/runConfigurations/Run_obsw_simple_client.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Test.xml b/.idea/runConfigurations/Test.xml deleted file mode 100644 index 81a2166..0000000 --- a/.idea/runConfigurations/Test.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Test_All.xml b/.idea/runConfigurations/Test_All.xml deleted file mode 100644 index 6f884df..0000000 --- a/.idea/runConfigurations/Test_All.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - \ No newline at end of file diff --git a/.idea/runConfigurations/Test_satrs_core.xml b/.idea/runConfigurations/Test_satrs_core.xml deleted file mode 100644 index fe5e4da..0000000 --- a/.idea/runConfigurations/Test_satrs_core.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index f8a4267..56dbd33 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "satrs-example/client", "satrs-example/minisim", "satrs-shared", + "tmtc-utils", "embedded-examples/embedded-client", "embedded-examples/types", ] diff --git a/embedded-examples/stm32h7-nucleo-rtic/Cargo.toml b/embedded-examples/stm32h7-nucleo-rtic/Cargo.toml index 126406e..5271a72 100644 --- a/embedded-examples/stm32h7-nucleo-rtic/Cargo.toml +++ b/embedded-examples/stm32h7-nucleo-rtic/Cargo.toml @@ -1,5 +1,4 @@ [package] -authors = ["Robin Mueller "] name = "satrs-stm32h7-nucleo-rtic" edition = "2021" version = "0.1.0" diff --git a/satrs-example/Cargo.toml b/satrs-example/Cargo.toml index e7a8477..a953281 100644 --- a/satrs-example/Cargo.toml +++ b/satrs-example/Cargo.toml @@ -2,7 +2,6 @@ name = "satrs-example" version = "0.1.1" edition = "2024" -authors = ["Robin Mueller "] default-run = "satrs-example" homepage = "https://egit.irs.uni-stuttgart.de/rust/sat-rs" repository = "https://egit.irs.uni-stuttgart.de/rust/sat-rs" diff --git a/satrs-mib/Cargo.toml b/satrs-mib/Cargo.toml index 6c0c351..79d4d74 100644 --- a/satrs-mib/Cargo.toml +++ b/satrs-mib/Cargo.toml @@ -3,7 +3,6 @@ name = "satrs-mib" version = "0.1.3" edition = "2021" rust-version = "1.61" -authors = ["Robin Mueller "] description = """ Helper crate of the sat-rs framework to build a mission information base (MIB) from the On-Board Software (OBSW) code directly.""" diff --git a/satrs-shared/Cargo.toml b/satrs-shared/Cargo.toml index 4b172cd..64c5827 100644 --- a/satrs-shared/Cargo.toml +++ b/satrs-shared/Cargo.toml @@ -3,7 +3,6 @@ name = "satrs-shared" description = "Components shared by multiple sat-rs crates" version = "0.2.4" edition = "2021" -authors = ["Robin Mueller "] homepage = "https://absatsw.irs.uni-stuttgart.de/projects/sat-rs/" repository = "https://egit.irs.uni-stuttgart.de/rust/sat-rs" license = "Apache-2.0" diff --git a/satrs/Cargo.toml b/satrs/Cargo.toml index a104eea..8140fee 100644 --- a/satrs/Cargo.toml +++ b/satrs/Cargo.toml @@ -3,7 +3,6 @@ name = "satrs" version = "0.3.0-alpha.3" edition = "2024" rust-version = "1.85.0" -authors = ["Robin Mueller "] description = "A library collection to build software for remote systems" homepage = "https://github.com/us-irs/sat-rs" repository = "https://egit.irs.uni-stuttgart.de/rust/sat-rs" diff --git a/tmtc-utils/.gitignore b/tmtc-utils/.gitignore new file mode 100644 index 0000000..4fffb2f --- /dev/null +++ b/tmtc-utils/.gitignore @@ -0,0 +1,2 @@ +/target +/Cargo.lock diff --git a/tmtc-utils/Cargo.toml b/tmtc-utils/Cargo.toml new file mode 100644 index 0000000..57f7c91 --- /dev/null +++ b/tmtc-utils/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "tmtc-utils" +description = "Commonly required utilities for writing TMTC clients with Rust" +version = "0.1.0" +edition = "2024" +homepage = "https://absatsw.irs.uni-stuttgart.de/projects/sat-rs/" +repository = "https://egit.irs.uni-stuttgart.de/rust/sat-rs" +license = "Apache-2.0" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +thiserror = "2" +serialport = { version = "4" } +cobs = "0.5" +log = "0.4" diff --git a/tmtc-utils/README.md b/tmtc-utils/README.md new file mode 100644 index 0000000..00322a7 --- /dev/null +++ b/tmtc-utils/README.md @@ -0,0 +1,6 @@ +TMTC Utilities +========= + +This crate contains commonly required utilities when writing TMTC clients with Rust. +This includes a transport module which introduces a packet based communication abstraction and +some concrete implementations for common communication interfaces. diff --git a/tmtc-utils/src/lib.rs b/tmtc-utils/src/lib.rs new file mode 100644 index 0000000..b662403 --- /dev/null +++ b/tmtc-utils/src/lib.rs @@ -0,0 +1,5 @@ +//! # TMTC Utilities +//! +//! This crate contains commonly required utilities when writing TMTC clients with Rust. +#![deny(missing_docs)] +pub mod transport; diff --git a/tmtc-utils/src/transport/mod.rs b/tmtc-utils/src/transport/mod.rs new file mode 100644 index 0000000..c4db470 --- /dev/null +++ b/tmtc-utils/src/transport/mod.rs @@ -0,0 +1,51 @@ +//! # Packet Transport Module. +//! +//! Introduces a communication abstraction for packet based communication and some concrete +//! implementations for common communication interfaces. This can be useful for exchanging +//! something like CCSDS space packets over different transport mechanisms. +pub mod serial; +pub mod tcp; +pub mod udp; + +/// Generic send error. +#[derive(Debug, thiserror::Error)] +pub enum SendError { + /// Queue is full. + #[error("queue is full")] + QueueFull, + /// IO error. + #[error("io error: {0}")] + Io(#[from] std::io::Error), + /// Other error. + #[error("other error")] + Other, +} + +/// Generic reception error. +#[derive(Debug, thiserror::Error)] +pub enum ReceiveError { + /// IO error. + #[error("io error: {0}")] + Io(#[from] std::io::Error), + /// Other error. + #[error("other error")] + Other, +} + +/// Generic packet transport trait. +/// +/// This abstraction allows different transport mechanism for packetized data like CCSDS space +/// packets. +pub trait PacketTransport { + /// Send a packet. + fn send(&mut self, packet: &[u8]) -> Result<(), SendError>; + + /// Receivd packets. + /// + /// For each received packet, the closure will be called with the packet as an argument. + /// The function will return the number of received packets. + fn receive(&mut self, f: F) -> Result; + + /// Close the connection, used for graceful shutdowns. + fn close(&mut self); +} diff --git a/tmtc-utils/src/transport/serial.rs b/tmtc-utils/src/transport/serial.rs new file mode 100644 index 0000000..49c93e3 --- /dev/null +++ b/tmtc-utils/src/transport/serial.rs @@ -0,0 +1,320 @@ +//! # Serial packet transport with COBS encoding. +use std::time::Duration; + +use cobs::CobsDecoderOwned; + +use crate::transport::PacketTransport; + +/// Packet transport via a serial interface with COBS encoding. +pub struct PacketTransportSerialCobs { + /// Underlying serial port. + pub serial: Box, + /// Enables/disables logging of decoding errors. + pub log_decoding_errors: bool, + reception_buffer: [u8; 1024], + decoder: cobs::CobsDecoderOwned, +} + +impl PacketTransportSerialCobs { + /// Constructor which constructs the [serialport::SerialPort] and [cobs::CobsDecoderOwned] from + /// the passed parameters. + /// + /// The `max_rx_packet_size` parameter defines the expected maximum size of a received packet. + /// On non-linux platforms, the serial timeout parameter has to be specified as well. + pub fn new_from_params( + port_name: &str, + baud_rate: u32, + max_rx_packet_size: usize, + ) -> Result { + let serial = serialport::new(port_name, baud_rate).open_native()?; + // Not merged yet in upstream.. + //#[cfg(target_os = "linux")] + //serial.set_read_mode(ReadMode::Immediate)?; + Ok(Self::new( + Box::new(serial), + CobsDecoderOwned::new(max_rx_packet_size), + )) + } + + /// Generic constructor. + pub fn new(serial: Box, decoder: cobs::CobsDecoderOwned) -> Self { + Self { + serial, + decoder, + reception_buffer: [0u8; 1024], + log_decoding_errors: true, + } + } + + /// Set the serial port timeout. + pub fn set_serial_timeout(&mut self, timeout: Duration) -> Result<(), serialport::Error> { + self.serial.set_timeout(timeout) + } + + /// Send a packet. + /// + /// It encodes the packet using COBS encoding before sending it over the serial port. + pub fn send(&mut self, packet: &[u8]) -> Result<(), super::SendError> { + let encoded = cobs::encode_vec_including_sentinels(packet); + log::debug!("sending COBS encoded packet: {:?}", encoded); + self.serial.write_all(&encoded)?; + // This is required to avoid timeout errors on bursty data. + self.serial.flush()?; + Ok(()) + } + + /// Received packets. + /// + /// This function pulls bytes from the serial port and feeds them into the COBS decoder. + /// For each received packet, the closure will be called with the decoded packet as an argument. + /// The function will return the number of received packets. + pub fn receive(&mut self, mut f: F) -> Result { + let mut decoded_packets = 0; + loop { + match self.serial.read(&mut self.reception_buffer) { + Ok(read_bytes) => { + if read_bytes == 0 { + break; + } + for byte in self.reception_buffer[..read_bytes].iter() { + match self.decoder.feed(*byte) { + Ok(Some(packet_len)) => { + f(&self.decoder.dest()[0..packet_len]); + decoded_packets += 1; + } + Ok(None) => (), + Err(e) => self.error_handler(e), + } + } + } + Err(e) => { + if e.kind() == std::io::ErrorKind::TimedOut + || e.kind() == std::io::ErrorKind::WouldBlock + { + break; + } + return Err(super::ReceiveError::Io(e)); + } + } + } + Ok(decoded_packets) + } + + fn error_handler(&self, error: cobs::DecodeError) { + if self.log_decoding_errors { + log::warn!("COBS decoding error: {:?}", error); + } + } +} + +impl PacketTransport for PacketTransportSerialCobs { + fn send(&mut self, packet: &[u8]) -> Result<(), super::SendError> { + self.send(packet) + } + + fn receive(&mut self, f: F) -> Result { + self.receive(f) + } + + fn close(&mut self) {} +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + use serialport::SerialPort; + + #[derive(Debug)] + pub struct SerialPortMock { + baud: u32, + pub tx_data: std::sync::mpsc::Sender>, + pub rx_data: std::sync::mpsc::Receiver>, + } + + impl std::io::Write for SerialPortMock { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.tx_data.send(buf.to_vec()).unwrap(); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl std::io::Read for SerialPortMock { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + match self.rx_data.try_recv() { + Ok(packet) => { + buf[0..packet.len()].copy_from_slice(&packet); + Ok(packet.len()) + } + Err(e) => match e { + std::sync::mpsc::TryRecvError::Empty => Ok(0), + std::sync::mpsc::TryRecvError::Disconnected => panic!("sender disconnected"), + }, + } + } + } + + impl SerialPort for SerialPortMock { + fn name(&self) -> Option { + Some("mock".into()) + } + + fn baud_rate(&self) -> serialport::Result { + Ok(115200) + } + + fn data_bits(&self) -> serialport::Result { + Ok(serialport::DataBits::Eight) + } + + fn flow_control(&self) -> serialport::Result { + Ok(serialport::FlowControl::None) + } + + fn parity(&self) -> serialport::Result { + Ok(serialport::Parity::None) + } + + fn stop_bits(&self) -> serialport::Result { + Ok(serialport::StopBits::One) + } + + fn timeout(&self) -> std::time::Duration { + Duration::from_millis(0) + } + + fn set_baud_rate(&mut self, baud_rate: u32) -> serialport::Result<()> { + self.baud = baud_rate; + Ok(()) + } + + fn set_data_bits(&mut self, _data_bits: serialport::DataBits) -> serialport::Result<()> { + Ok(()) + } + + fn set_flow_control( + &mut self, + _flow_control: serialport::FlowControl, + ) -> serialport::Result<()> { + Ok(()) + } + + fn set_parity(&mut self, _parity: serialport::Parity) -> serialport::Result<()> { + Ok(()) + } + + fn set_stop_bits(&mut self, _stop_bits: serialport::StopBits) -> serialport::Result<()> { + Ok(()) + } + + fn set_timeout(&mut self, _timeout: std::time::Duration) -> serialport::Result<()> { + Ok(()) + } + + fn write_request_to_send(&mut self, _level: bool) -> serialport::Result<()> { + Ok(()) + } + + fn write_data_terminal_ready(&mut self, _level: bool) -> serialport::Result<()> { + Ok(()) + } + + fn read_clear_to_send(&mut self) -> serialport::Result { + Ok(true) + } + + fn read_data_set_ready(&mut self) -> serialport::Result { + Ok(true) + } + + fn read_ring_indicator(&mut self) -> serialport::Result { + Ok(true) + } + + fn read_carrier_detect(&mut self) -> serialport::Result { + Ok(true) + } + + fn bytes_to_read(&self) -> serialport::Result { + Err(serialport::Error::new( + serialport::ErrorKind::NoDevice, + "Mock does not support bytes_to_read", + )) + } + + fn bytes_to_write(&self) -> serialport::Result { + Ok(0) + } + + fn clear(&self, _buffer_to_clear: serialport::ClearBuffer) -> serialport::Result<()> { + Ok(()) + } + + fn try_clone(&self) -> serialport::Result> { + serialport::Result::Err(serialport::Error::new( + serialport::ErrorKind::NoDevice, + "Mock clone not supported", + )) + } + + fn set_break(&self) -> serialport::Result<()> { + Ok(()) + } + + fn clear_break(&self) -> serialport::Result<()> { + Ok(()) + } + } + + #[test] + fn basic_tx_test() { + let (tx_tx, tx_rx) = std::sync::mpsc::channel(); + let (_rx_tx, rx_rx) = std::sync::mpsc::channel(); + let mut transport = PacketTransportSerialCobs::new( + Box::new(SerialPortMock { + baud: 115200, + tx_data: tx_tx, + rx_data: rx_rx, + }), + CobsDecoderOwned::new(128), + ); + let sent_data = [1, 2, 3, 4]; + transport.send(&sent_data).unwrap(); + let encoded_data = tx_rx.recv().unwrap(); + assert_eq!( + encoded_data, + cobs::encode_vec_including_sentinels(&sent_data) + ); + } + + #[test] + fn basic_rx_test() { + let (tx_tx, _tx_rx) = std::sync::mpsc::channel(); + let (rx_tx, rx_rx) = std::sync::mpsc::channel(); + let mut transport = PacketTransportSerialCobs::new( + Box::new(SerialPortMock { + baud: 115200, + tx_data: tx_tx, + rx_data: rx_rx, + }), + CobsDecoderOwned::new(128), + ); + let rx_data = [1, 2, 3, 4]; + rx_tx + .send(cobs::encode_vec_including_sentinels(&rx_data)) + .unwrap(); + assert_eq!( + transport + .receive(|packet| { + assert_eq!(packet, &rx_data); + }) + .unwrap(), + 1 + ); + } +} diff --git a/tmtc-utils/src/transport/tcp.rs b/tmtc-utils/src/transport/tcp.rs new file mode 100644 index 0000000..7af22e2 --- /dev/null +++ b/tmtc-utils/src/transport/tcp.rs @@ -0,0 +1,158 @@ +//! # Packet transport via TCP with COBS encoding. +use std::{ + io::{Read as _, Write as _}, + time::Duration, +}; + +use crate::transport::PacketTransport; + +/// Packet transport via TCP with COBS encoding. +/// +/// Currently only allows a maxium packet size of 4096. +pub struct PacketTransportTcpWithCobs { + /// Underlying TCP stream. + pub tcp_stream: std::net::TcpStream, + /// Can be used to disable logging of decoding errors. + pub log_decoding_errors: bool, + /// Decoder object. + decoder: cobs::CobsDecoderOwned, + reception_buffer: [u8; 4096], +} + +impl PacketTransportTcpWithCobs { + /// Generic constructor. + /// + /// The `tcp_stream` parameter is the underlying TCP stream which should already be connected. + pub fn new( + tcp_stream: std::net::TcpStream, + decoder: cobs::CobsDecoderOwned, + ) -> std::io::Result { + tcp_stream.set_nonblocking(true)?; + tcp_stream.set_read_timeout(Some(Duration::from_millis(100)))?; + Ok(Self { + tcp_stream, + decoder, + reception_buffer: [0u8; 4096], + log_decoding_errors: true, + }) + } + + /// Send a packet. + /// + /// It encodes the packet using COBS encoding before sending it over the TCP stream. + pub fn send(&mut self, packet: &[u8]) -> Result<(), super::SendError> { + let cobs_encoded_packet = cobs::encode_vec_including_sentinels(packet); + self.tcp_stream.write_all(&cobs_encoded_packet)?; + Ok(()) + } + + /// Received packets. + /// + /// This function pulls bytes from the TCP stream and feeds them into the COBS decoder. + /// For each received packet, the closure will be called with the decoded packet as an argument. + /// The function will return the number of received packets. + /// + /// Please note that this function may block on the TCP stream read call, but it will not + /// block indifinitely due to the read timeout set on the TCP stream. + pub fn receive(&mut self, mut f: impl FnMut(&[u8])) -> Result { + let mut decoded_packets = 0; + loop { + let read_size = self + .tcp_stream + .read(&mut self.reception_buffer) + .unwrap_or(0); + if read_size == 0 { + break; + } + for byte in &self.reception_buffer[0..read_size] { + match self.decoder.feed(*byte) { + Ok(Some(packet_len)) => { + f(&self.decoder.dest()[0..packet_len]); + decoded_packets += 1; + } + Ok(None) => (), + Err(e) => self.error_handler(e), + } + } + } + Ok(decoded_packets) + } + + fn error_handler(&self, error: cobs::DecodeError) { + if self.log_decoding_errors { + log::warn!("COBS decoding error: {:?}", error); + } + } + + /// Close the connection by shutting down the TCP stream. + pub fn close(&mut self) -> std::io::Result<()> { + self.tcp_stream.shutdown(std::net::Shutdown::Both) + } +} + +impl PacketTransport for PacketTransportTcpWithCobs { + fn send(&mut self, packet: &[u8]) -> Result<(), super::SendError> { + self.send(packet) + } + + fn receive(&mut self, f: F) -> Result { + self.receive(f) + } + + fn close(&mut self) { + let _ = self.close(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn basic_send_test() { + let tcp_server = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + tcp_server + .set_nonblocking(true) + .expect("failed to set blocking mode"); + let addr = tcp_server.local_addr().unwrap(); + let tcp_client = std::net::TcpStream::connect(addr).unwrap(); + let mut transport = + PacketTransportTcpWithCobs::new(tcp_client, cobs::CobsDecoderOwned::new(1024)).unwrap(); + let packet = [1, 2, 3, 4]; + transport.send(&packet).unwrap(); + tcp_server + .accept() + .map(|(mut stream, _)| { + let mut buffer = [0u8; 1024]; + let read_size = stream.read(&mut buffer).unwrap(); + let decoded_packet = cobs::decode_vec(&buffer[0..read_size]).unwrap(); + assert_eq!(decoded_packet, packet); + }) + .unwrap(); + } + + #[test] + fn basic_receive_test() { + let tcp_server = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + tcp_server + .set_nonblocking(true) + .expect("failed to set blocking mode"); + let addr = tcp_server.local_addr().unwrap(); + let tcp_client = std::net::TcpStream::connect(addr).unwrap(); + let mut transport = + PacketTransportTcpWithCobs::new(tcp_client, cobs::CobsDecoderOwned::new(1024)).unwrap(); + let rx_data = [1, 2, 3, 4]; + let encoded_data = cobs::encode_vec_including_sentinels(&rx_data); + tcp_server + .accept() + .map(|(mut stream, _)| { + stream.write_all(&encoded_data).unwrap(); + }) + .unwrap(); + transport + .receive(|packet| { + assert_eq!(packet, &rx_data); + }) + .unwrap(); + } +} diff --git a/tmtc-utils/src/transport/udp.rs b/tmtc-utils/src/transport/udp.rs new file mode 100644 index 0000000..21cdcd3 --- /dev/null +++ b/tmtc-utils/src/transport/udp.rs @@ -0,0 +1,236 @@ +//! # Packet transport via UDP + +use std::io::ErrorKind; +use std::net::ToSocketAddrs; + +use crate::transport::PacketTransport; + +/// Generic packet transport via UDP. +/// +/// Currently only allows a maxium packet size of 4096. +pub struct PacketTransportUdp { + /// Underlying UDP socket. + pub socket: std::net::UdpSocket, + target: std::net::SocketAddr, + reception_buffer: [u8; 4096], +} + +impl PacketTransportUdp { + /// Generic constructor. + /// + /// The `socket` parameter is the underlying UDP stream which should already be connected. + /// It will be set non-blocking by the construtor. + pub fn new( + socket: std::net::UdpSocket, + target: std::net::SocketAddr, + ) -> Result { + socket.set_nonblocking(true)?; + Ok(Self { + socket, + target, + reception_buffer: [0u8; 4096], + }) + } + + /// Update default target. + pub fn set_default_target(&mut self, target: std::net::SocketAddr) { + self.target = target; + } + + /// Send a packet to the target address specified in the constructor. + pub fn send(&mut self, packet: &[u8]) -> Result<(), super::SendError> { + self.socket + .send_to(packet, self.target) + .map_err(super::SendError::Io)?; + Ok(()) + } + + /// Send packet to a specific address. + pub fn send_to( + &mut self, + packet: &[u8], + addr: A, + ) -> Result<(), super::SendError> { + self.socket + .send_to(packet, addr) + .map_err(super::SendError::Io)?; + Ok(()) + } + + /// Receive packets and call the provided callback for each received packet. + pub fn receive(&mut self, mut f: F) -> Result { + let mut packets_received = 0; + loop { + match self.socket.recv_from(&mut self.reception_buffer) { + Ok((bytes, _)) => { + packets_received += 1; + f(&self.reception_buffer[..bytes]); + } + Err(e) => { + if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut { + break; + } + log::error!("UDP reception error: {e}"); + } + } + } + Ok(packets_received) + } +} + +impl PacketTransport for PacketTransportUdp { + fn send(&mut self, packet: &[u8]) -> Result<(), super::SendError> { + self.send(packet) + } + + fn receive(&mut self, f: F) -> Result { + self.receive(f) + } + + fn close(&mut self) {} +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn basic_send_test() { + let receiver = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let sender = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + + let receiver_addr = receiver.local_addr().unwrap(); + sender.connect(receiver_addr).unwrap(); + + let mut transport = PacketTransportUdp::new(sender, receiver_addr).unwrap(); + let payload = [1u8, 2, 3, 4]; + + transport.send(&payload).unwrap(); + + let mut buf = [0u8; 16]; + let (len, from) = receiver.recv_from(&mut buf).unwrap(); + + assert_eq!(&buf[..len], &payload); + assert_eq!(from, transport.socket.local_addr().unwrap()); + } + + #[test] + fn receive_is_non_blocking_when_no_data_is_available() { + let receiver = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let sender = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + + let receiver_addr = receiver.local_addr().unwrap(); + let mut transport = PacketTransportUdp::new(sender, receiver_addr).unwrap(); + + let start = std::time::Instant::now(); + let mut callback_called = false; + + let packets_received = transport + .receive(|_| { + callback_called = true; + }) + .unwrap(); + + let elapsed = start.elapsed(); + + assert_eq!(packets_received, 0); + assert!(!callback_called); + assert!( + elapsed < std::time::Duration::from_millis(50), + "receive() took too long for a non-blocking socket: {:?}", + elapsed + ); + } + + #[test] + fn basic_receive_test_single() { + let plain_sender = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let transport_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + + let transport_addr = transport_socket.local_addr().unwrap(); + + let mut transport = PacketTransportUdp::new(transport_socket, transport_addr).unwrap(); + + plain_sender + .send_to(&[1u8, 2, 3, 4], transport_addr) + .unwrap(); + + let mut received_packets: Vec> = Vec::new(); + let packets_received = transport + .receive(|packet| received_packets.push(packet.to_vec())) + .unwrap(); + + assert_eq!(packets_received, 1); + assert_eq!(received_packets.len(), 1); + assert_eq!(received_packets[0], vec![1u8, 2, 3, 4]); + } + + #[test] + fn multi_packet_receive_test() { + let plain_sender = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let transport_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + + let receiver_addr = transport_socket.local_addr().unwrap(); + let mut transport = PacketTransportUdp::new(transport_socket, receiver_addr).unwrap(); + + plain_sender + .send_to(&[1u8, 2, 3, 4], receiver_addr) + .unwrap(); + plain_sender + .send_to(&[5u8, 6, 7, 8], receiver_addr) + .unwrap(); + plain_sender + .send_to(&[9u8, 10, 11, 12], receiver_addr) + .unwrap(); + + let mut received_packets: Vec> = Vec::new(); + let packets_received = transport + .receive(|packet| received_packets.push(packet.to_vec())) + .unwrap(); + + assert_eq!(packets_received, 3); + assert_eq!(received_packets.len(), 3); + assert_eq!(received_packets[0], vec![1u8, 2, 3, 4]); + assert_eq!(received_packets[1], vec![5u8, 6, 7, 8]); + assert_eq!(received_packets[2], vec![9u8, 10, 11, 12]); + } + + #[test] + fn send_and_receive_test() { + // Bind both sockets + let plain_receiver = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let sender_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + + let receiver_addr = plain_receiver.local_addr().unwrap(); + + // Build the transport around the sender socket + let mut send_transport = PacketTransportUdp::new(sender_socket, receiver_addr).unwrap(); + + // Send a packet + let payload = [1u8, 2, 3, 4]; + send_transport.send(&payload).unwrap(); + + // Plain receiver reads what was sent + let mut buf = [0u8; 4096]; + let (n, src) = plain_receiver.recv_from(&mut buf).unwrap(); + assert_eq!(n, payload.len()); + assert_eq!(&buf[..n], &payload); + + // Plain receiver sends back some test data + let reply = [5u8, 6, 7, 8]; + plain_receiver.send_to(&reply, src).unwrap(); + + // Read the reply via the transport (non-blocking socket, reply should already be in flight) + let mut received_packets: Vec> = Vec::new(); + // Small yield to ensure the reply has arrived on the loopback interface + std::thread::sleep(std::time::Duration::from_millis(10)); + + let packets_received = send_transport + .receive(|packet| received_packets.push(packet.to_vec())) + .unwrap(); + + assert_eq!(packets_received, 1); + assert_eq!(received_packets.len(), 1); + assert_eq!(received_packets[0], reply.to_vec()); + } +}