From 012509f325fea9bc4dba8f8d459b16c6d4982e5f Mon Sep 17 00:00:00 2001 From: Belousov Oleg Date: Sun, 12 Jul 2026 17:53:29 +0300 Subject: [PATCH] TCP transport for KISS interface --- crates/rns-interface/src/kiss_iface.rs | 461 ++++++++++++------ crates/rns-interface/src/lib.rs | 2 + crates/rns-interface/src/rnode.rs | 272 +---------- crates/rns-interface/src/serial_tcp_stream.rs | 425 ++++++++++++++++ 4 files changed, 752 insertions(+), 408 deletions(-) create mode 100644 crates/rns-interface/src/serial_tcp_stream.rs diff --git a/crates/rns-interface/src/kiss_iface.rs b/crates/rns-interface/src/kiss_iface.rs index 1de8a8bf..ba815fb8 100644 --- a/crates/rns-interface/src/kiss_iface.rs +++ b/crates/rns-interface/src/kiss_iface.rs @@ -1,4 +1,10 @@ -//! Serial port + KISS framing; CMD_DATA + CMD_READY flow control. +//! KISS framing over a serial port or TCP socket; CMD_DATA + CMD_READY flow +//! control. Transport selection mirrors [`crate::rnode`]: +//! - `/dev/ttyUSB0`, `COM3`, etc. -> serial +//! - `tcp://192.168.1.1` -> TCP, default port 7633 +//! - `tcp://192.168.1.1:9000` -> TCP, explicit port +//! +//! A TCP KISS interface reconnects automatically after the link drops. use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; @@ -8,14 +14,14 @@ use bytes::Bytes; use tokio::sync::mpsc; use crate::kiss; +use crate::serial_tcp_stream::{PortConfig, SerialTcpStream, read_stream, reconnect_delay}; use crate::traits::{InterfaceDirection, InterfaceHandle, InterfaceId, InterfaceMode}; use rns_transport::messages::{InboundPacket, TransportMessage}; -const KISS_READ_TIMEOUT_MS: u64 = 100; - #[derive(Debug, Clone)] pub struct KissInterfaceConfig { pub name: String, + /// Serial device path (`/dev/ttyUSB0`) **or** TCP URL (`tcp://host[:port]`). pub port: String, pub baud_rate: u32, pub data_bits: serialport::DataBits, @@ -65,43 +71,53 @@ pub fn beacon_frame_payload(id_callsign: &[u8]) -> Vec { frame } -pub async fn spawn_kiss_interface( - config: KissInterfaceConfig, - id: InterfaceId, - transport_tx: mpsc::Sender, -) -> Result { - let port = serialport::new(&config.port, config.baud_rate) - .data_bits(config.data_bits) - .parity(config.parity) - .stop_bits(config.stop_bits) - .timeout(Duration::from_millis(KISS_READ_TIMEOUT_MS)) - .open() - .map_err(|e| { - crate::traits::InterfaceError::SendFailed(format!("kiss serial open: {}", e)) - })?; +/// Open the transport and push TNC tuning (TXDELAY/P/SLOTTIME/TXTAIL) before +/// the main loops start, so it takes effect before the first data frame. +/// Called again on every reconnect, since a fresh TNC session has forgotten +/// any previously pushed tuning. +async fn open_configured_kiss_stream( + config: &KissInterfaceConfig, + port_cfg: &PortConfig, +) -> Result { + let port = match port_cfg { + #[cfg(feature = "serial")] + PortConfig::Serial { path, baud } => { + tracing::info!( + name = %config.name, + port = %path, + baud = baud, + "KISS serial interface opening" + ); + SerialTcpStream::open_serial(path, *baud).map_err(|e| { + crate::traits::InterfaceError::SendFailed(format!("kiss serial open: {}", e)) + })? + } + PortConfig::Tcp { addr } => { + tracing::info!( + name = %config.name, + addr = %addr, + "KISS TCP interface connecting" + ); + let addr = addr.clone(); + tokio::task::spawn_blocking(move || SerialTcpStream::connect_tcp(&addr)) + .await + .map_err(|e| { + crate::traits::InterfaceError::SendFailed(format!("kiss tcp spawn: {}", e)) + })? + .map_err(|e| { + crate::traits::InterfaceError::SendFailed(format!("kiss tcp connect: {}", e)) + })? + } + }; tracing::info!( name = %config.name, - port = %config.port, - baud = config.baud_rate, + endpoint = %port.description(), "KISS interface opened" ); - let online = Arc::new(AtomicBool::new(true)); - let shared_rxb = Arc::new(AtomicU64::new(0)); - let shared_txb = Arc::new(AtomicU64::new(0)); - let (tx, mut rx) = mpsc::channel::(256); - let name = config.name.clone(); - let mode = config.mode; - let flow_control = config.flow_control; - - let port_write = port - .try_clone() - .map_err(|e| crate::traits::InterfaceError::SendFailed(format!("kiss clone: {}", e)))?; - - // Push TNC tuning before main loops so they take effect before first frame. { - let mut init_port = port_write.try_clone().map_err(|e| { + let mut init_port = port.try_clone().map_err(|e| { crate::traits::InterfaceError::SendFailed(format!("kiss init clone: {}", e)) })?; let mut init_frames = Vec::with_capacity(16); @@ -119,144 +135,241 @@ pub async fn spawn_kiss_interface( } if !init_frames.is_empty() { use std::io::Write; - let _ = init_port.write_all(&init_frames); - let _ = init_port.flush(); + init_port.write_all(&init_frames).map_err(|e| { + crate::traits::InterfaceError::SendFailed(format!("kiss init write: {}", e)) + })?; + init_port.flush().map_err(|e| { + crate::traits::InterfaceError::SendFailed(format!("kiss init flush: {}", e)) + })?; } } - let ready = Arc::new(AtomicBool::new(true)); + Ok(port) +} - let online_w = online.clone(); - let ready_w = ready.clone(); - let txb_w = shared_txb.clone(); - let beacon = config +pub async fn spawn_kiss_interface( + config: KissInterfaceConfig, + id: InterfaceId, + transport_tx: mpsc::Sender, +) -> Result { + let port_cfg = PortConfig::parse(&config.port, config.baud_rate) + .map_err(|e| crate::traits::InterfaceError::SendFailed(format!("kiss port parse: {}", e)))?; + + let port = open_configured_kiss_stream(&config, &port_cfg).await?; + + let online = Arc::new(AtomicBool::new(true)); + let shared_rxb = Arc::new(AtomicU64::new(0)); + let shared_txb = Arc::new(AtomicU64::new(0)); + // Outer channel: survives reconnects. Each connection attempt gets its + // own inner `conn_tx`/write-task; a forwarding task bridges the two so + // callers holding `tx` never notice a reconnect happened underneath. + let (tx, rx) = mpsc::channel::(256); + let rx = Arc::new(tokio::sync::Mutex::new(rx)); + let name = config.name.clone(); + let mode = config.mode; + let flow_control = config.flow_control; + let beacon: Option<(Duration, Bytes)> = config .id_interval .zip(config.id_callsign.clone()) .map(|(interval, callsign)| (Duration::from_secs(interval), Bytes::from(callsign))); - tokio::spawn(async move { - let mut port_w = port_write; - // Python first_tx semantics: armed by the first data TX after a - // beacon; cleared when the beacon goes out. - let mut first_tx: Option = None; + + let online_r = online.clone(); + let rxb_r = shared_rxb.clone(); + let txb_r = shared_txb.clone(); + let task_config = config.clone(); + let task_port_cfg = port_cfg.clone(); + let task_name = config.name.clone(); + let read_task = tokio::spawn(async move { + let mut next_port = Some(port); + loop { - let data = if let Some((interval, ref callsign)) = beacon { - match tokio::time::timeout(Duration::from_secs(1), rx.recv()).await { - Ok(Some(data)) => data, - Ok(None) => break, - Err(_) => { - let due = first_tx.is_some_and(|t| t.elapsed() >= interval); - if !due { - continue; - } - tracing::debug!("KISS transmitting station-ID beacon"); - Bytes::from(beacon_frame_payload(callsign)) + let mut port_r = match next_port.take() { + Some(port) => port, + None => match open_configured_kiss_stream(&task_config, &task_port_cfg).await { + Ok(port) => port, + Err(e) => { + online_r.store(false, Ordering::SeqCst); + tracing::warn!( + name = %task_name, + error = %e, + "KISS reconnect failed" + ); + tokio::time::sleep(reconnect_delay()).await; + continue; } - } - } else { - match rx.recv().await { - Some(data) => data, - None => break, + }, + }; + + online_r.store(true, Ordering::SeqCst); + let port_write = match port_r.try_clone() { + Ok(port) => port, + Err(e) => { + tracing::warn!(error = %e, "KISS clone failed before reconnect"); + online_r.store(false, Ordering::SeqCst); + tokio::time::sleep(reconnect_delay()).await; + continue; } }; - // Python KISSInterface.py:267-271 compares the unpadded callsign, - // so a padded (<15 byte) beacon re-arms the timer and beacons - // repeat every id_interval once anything has been sent. Kept - // bug-for-bug for parity. - let is_beacon = beacon - .as_ref() - .is_some_and(|(_, callsign)| data == *callsign); - if is_beacon { - first_tx = None; - } else if first_tx.is_none() { - first_tx = Some(tokio::time::Instant::now()); - } + let ready = Arc::new(AtomicBool::new(true)); + let (conn_tx, mut conn_rx) = mpsc::channel::(256); - txb_w.fetch_add(data.len() as u64, std::sync::atomic::Ordering::Relaxed); - // Flow control: bounded wait so a stuck TNC can't hang transmit. - if flow_control { - let deadline = tokio::time::Instant::now() + Duration::from_secs(5); - while !ready_w.load(Ordering::SeqCst) { - if tokio::time::Instant::now() >= deadline { - tracing::warn!("KISS flow control timeout, proceeding anyway"); - break; + let online_w = online_r.clone(); + let ready_w = ready.clone(); + let txb_w = txb_r.clone(); + let beacon_w = beacon.clone(); + let write_handle = tokio::spawn(async move { + let mut port_w = port_write; + // Python first_tx semantics: armed by the first data TX after a + // beacon; cleared when the beacon goes out. + let mut first_tx: Option = None; + loop { + let data = if let Some((interval, ref callsign)) = beacon_w { + match tokio::time::timeout(Duration::from_secs(1), conn_rx.recv()).await { + Ok(Some(data)) => data, + Ok(None) => break, + Err(_) => { + let due = first_tx.is_some_and(|t| t.elapsed() >= interval); + if !due { + continue; + } + tracing::debug!("KISS transmitting station-ID beacon"); + Bytes::from(beacon_frame_payload(callsign)) + } + } + } else { + match conn_rx.recv().await { + Some(data) => data, + None => break, + } + }; + + // Python KISSInterface.py:267-271 compares the unpadded + // callsign, so a padded (<15 byte) beacon re-arms the + // timer and beacons repeat every id_interval once + // anything has been sent. Kept bug-for-bug for parity. + let is_beacon = beacon_w + .as_ref() + .is_some_and(|(_, callsign)| data == *callsign); + if is_beacon { + first_tx = None; + } else if first_tx.is_none() { + first_tx = Some(tokio::time::Instant::now()); } - tokio::time::sleep(Duration::from_millis(10)).await; - if !online_w.load(Ordering::SeqCst) { - return; + + txb_w.fetch_add(data.len() as u64, Ordering::Relaxed); + // Flow control: bounded wait so a stuck TNC can't hang transmit. + if flow_control { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while !ready_w.load(Ordering::SeqCst) { + if tokio::time::Instant::now() >= deadline { + tracing::warn!("KISS flow control timeout, proceeding anyway"); + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + if !online_w.load(Ordering::SeqCst) { + return; + } + } + } + let framed = kiss::frame(&data); + match crate::serial_io::blocking_write_all(port_w, framed).await { + Ok(p) => { + port_w = p; + } + Err(e) => { + tracing::warn!(error = %e, "KISS write error"); + online_w.store(false, Ordering::SeqCst); + break; + } } } - } - let framed = kiss::frame(&data); - match crate::serial_io::blocking_write_all(port_w, framed).await { - Ok(p) => { - port_w = p; + }); + + let rx_ref = rx.clone(); + let fwd_handle = tokio::spawn(async move { + let mut guard = rx_ref.lock().await; + while let Some(data) = guard.recv().await { + if conn_tx.send(data).await.is_err() { + break; + } } - Err(e) => { - tracing::warn!(error = %e, "KISS write error"); - online_w.store(false, Ordering::SeqCst); + }); + + let mut deframer = kiss::KissDeframer::new(); + let mut buf = [0u8; 1024]; + let mut transport_closed = false; + + loop { + if !online_r.load(Ordering::SeqCst) { break; } - } - } - }); + let result = tokio::task::spawn_blocking(move || read_stream(port_r, buf)).await; - let online_r = online.clone(); - let ready_r = ready; - let rxb_r = shared_rxb.clone(); - let read_task = tokio::spawn(async move { - let mut port_r = port; - let mut deframer = kiss::KissDeframer::new(); - let mut buf = [0u8; 1024]; - - loop { - if !online_r.load(Ordering::SeqCst) { - break; - } - match crate::serial_io::poll_read(port_r, buf).await { - Ok((p, b, n)) => { - port_r = p; - buf = b; - if n > 0 { - rxb_r.fetch_add(n as u64, std::sync::atomic::Ordering::Relaxed); - for (cmd, frame) in deframer.feed(&buf[..n]) { - match cmd { - kiss::CMD_DATA => { - if frame.is_empty() { - continue; + match result { + Ok(Ok((p, b, n))) => { + port_r = p; + buf = b; + if n > 0 { + rxb_r.fetch_add(n as u64, Ordering::Relaxed); + for (cmd, frame) in deframer.feed(&buf[..n]) { + match cmd { + kiss::CMD_DATA => { + if frame.is_empty() { + continue; + } + let msg = TransportMessage::Inbound(InboundPacket { + raw: Bytes::from(frame), + interface_id: id, + rssi: None, + snr: None, + q: None, + }); + if transport_tx.send(msg).await.is_err() { + tracing::warn!(id, "transport channel closed"); + transport_closed = true; + break; + } } - let msg = TransportMessage::Inbound(InboundPacket { - raw: Bytes::from(frame), - interface_id: id, - rssi: None, - snr: None, - q: None, - }); - if transport_tx.send(msg).await.is_err() { - tracing::warn!(id, "transport channel closed"); - online_r.store(false, Ordering::SeqCst); - return; + kiss::CMD_READY => { + // Nonzero = TNC ready to accept data. + let is_ready = frame.first().copied().unwrap_or(0) != 0; + ready.store(is_ready, Ordering::SeqCst); + tracing::debug!(id, ready = is_ready, "KISS flow control"); + } + _ => { + tracing::debug!(id, cmd, "ignoring KISS command"); } - } - kiss::CMD_READY => { - // Nonzero = TNC ready to accept data. - let is_ready = frame.first().copied().unwrap_or(0) != 0; - ready_r.store(is_ready, Ordering::SeqCst); - tracing::debug!(id, ready = is_ready, "KISS flow control"); - } - _ => { - tracing::debug!(id, cmd, "ignoring KISS command"); } } + if transport_closed { + break; + } } } - } - Err(e) => { - tracing::warn!(error = %e, "KISS read error"); - online_r.store(false, Ordering::SeqCst); - return; + Ok(Err((_p, e))) => { + tracing::warn!(error = %e, "KISS read error"); + break; + } + Err(e) => { + tracing::warn!(error = %e, "KISS read task panicked"); + break; + } } } + + online_r.store(false, Ordering::SeqCst); + fwd_handle.abort(); + let _ = fwd_handle.await; + write_handle.abort(); + let _ = write_handle.await; + + if transport_closed { + return; + } + + tracing::info!(name = %task_name, "KISS reconnecting"); + tokio::time::sleep(reconnect_delay()).await; } }); @@ -333,4 +446,66 @@ mod tests { let cfg = KissInterfaceConfig::new("kiss0", "/dev/ttyS0", 9600); assert_eq!(cfg.mode, InterfaceMode::Full); } + + #[test] + fn test_kiss_port_config_tcp() { + let cfg = PortConfig::parse("tcp://192.168.1.50", 9600).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "192.168.1.50:7633"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_kiss_port_config_serial() { + let cfg = PortConfig::parse("/dev/ttyUSB0", 9600).unwrap(); + assert!(matches!(cfg, PortConfig::Serial { path, baud } if path == "/dev/ttyUSB0" && baud == 9600)); + } + + /// A TCP KISS interface reconnects instead of dying when the peer + /// closes the connection. + #[tokio::test] + async fn test_kiss_tcp_reconnects_after_eof() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let config = KissInterfaceConfig::new("kiss-tcp", &format!("tcp://{addr}"), 9600); + let (accepted_tx, mut accepted_rx) = tokio::sync::mpsc::unbounded_channel(); + + let server = std::thread::spawn(move || { + for attempt in 1..=2 { + let (stream, _) = listener.accept().unwrap(); + if attempt == 1 { + let _ = stream.shutdown(std::net::Shutdown::Both); + } + accepted_tx.send(attempt).unwrap(); + if attempt == 2 { + std::thread::sleep(Duration::from_millis(500)); + } + } + }); + + let (transport_tx, _transport_rx) = mpsc::channel::(8); + let handle = spawn_kiss_interface(config, 99, transport_tx).await.unwrap(); + + assert_eq!( + tokio::time::timeout(Duration::from_secs(2), accepted_rx.recv()) + .await + .unwrap() + .unwrap(), + 1 + ); + assert_eq!( + tokio::time::timeout(Duration::from_secs(7), accepted_rx.recv()) + .await + .unwrap() + .unwrap(), + 2 + ); + assert!(handle.online.load(Ordering::SeqCst)); + + handle.read_task.abort(); + drop(handle.tx); + server.join().unwrap(); + } } diff --git a/crates/rns-interface/src/lib.rs b/crates/rns-interface/src/lib.rs index cc686065..b1810a54 100644 --- a/crates/rns-interface/src/lib.rs +++ b/crates/rns-interface/src/lib.rs @@ -35,6 +35,8 @@ pub mod rnode_multi; pub mod serial; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] mod serial_io; +#[cfg(any(feature = "serial", feature = "rnode-tcp"))] +pub mod serial_tcp_stream; pub mod socket_tuning; pub mod tcp; pub mod traits; diff --git a/crates/rns-interface/src/rnode.rs b/crates/rns-interface/src/rnode.rs index 41e232d8..f251db54 100644 --- a/crates/rns-interface/src/rnode.rs +++ b/crates/rns-interface/src/rnode.rs @@ -102,28 +102,13 @@ pub const REQUIRED_FW_VER_MIN: u8 = 52; pub const RSSI_OFFSET: i32 = 157; -pub const RECONNECT_WAIT: u64 = 5; +pub const RECONNECT_WAIT: u64 = crate::serial_tcp_stream::RECONNECT_WAIT_SECS; pub const RADIO_STATE_ON: u8 = 0x01; pub const RADIO_STATE_OFF: u8 = 0x00; /// Default TCP port for RNode-over-IP. -pub const DEFAULT_TCP_PORT: u16 = 7633; - -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -const RNODE_READ_TIMEOUT_MS: u64 = 100; -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -const RNODE_TCP_CONNECT_TIMEOUT_SECS: u64 = 5; -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -const RNODE_TCP_KEEPIDLE_SECS: u64 = 5; -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -const RNODE_TCP_KEEPINTVL_SECS: u64 = 2; -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -const RNODE_TCP_KEEPCNT: u32 = 12; -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -const RNODE_TCP_USER_TIMEOUT_SECS: u64 = 24; -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -const RNODE_TCP_BUFFER_BYTES: usize = 131_072; +pub use crate::serial_tcp_stream::DEFAULT_TCP_PORT; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] type RNodeStopRegistry = Mutex>>; @@ -207,247 +192,13 @@ async fn send_detach_request(conn_tx: &mpsc::Sender, id: Inte } } -// Transport abstraction - -/// Parsed representation of the `port` config field. +// Transport abstraction — shared with kiss_iface via `serial_tcp_stream`. #[cfg(any(feature = "serial", feature = "rnode-tcp"))] -#[derive(Debug, Clone)] -pub enum PortConfig { - /// A local serial device path, e.g. `/dev/ttyUSB0` or `COM3`. - #[cfg(feature = "serial")] - Serial { path: String, baud: u32 }, - /// A TCP endpoint, e.g. `tcp://192.168.1.1` or `tcp://192.168.1.1:9000`. - Tcp { addr: String }, -} - +use crate::serial_tcp_stream::PortConfig; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] -impl PortConfig { - pub fn parse(port: &str, baud: u32) -> Result { - #[cfg(not(feature = "serial"))] - let _ = baud; - - if let Some(rest) = strip_tcp_scheme(port) { - let addr = parse_tcp_endpoint(rest)?; - Ok(Self::Tcp { addr }) - } else { - #[cfg(feature = "serial")] - { - Ok(Self::Serial { - path: port.to_string(), - baud, - }) - } - #[cfg(not(feature = "serial"))] - Err("RNode serial ports require the 'serial' feature; use tcp://host[:port] for TCP RNodes".to_string()) - } - } -} - +use crate::serial_tcp_stream::SerialTcpStream as RNodeStream; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] -fn strip_tcp_scheme(port: &str) -> Option<&str> { - const TCP_SCHEME: &str = "tcp://"; - port.get(..TCP_SCHEME.len()) - .filter(|prefix| prefix.eq_ignore_ascii_case(TCP_SCHEME)) - .and_then(|_| port.get(TCP_SCHEME.len()..)) -} - -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -fn parse_tcp_endpoint(endpoint: &str) -> Result { - if endpoint.is_empty() { - return Err("missing TCP host".to_string()); - } - - if let Some(rest) = endpoint.strip_prefix('[') { - let Some(closing) = rest.find(']') else { - return Err("missing closing ']' in IPv6 TCP host".to_string()); - }; - let host = &rest[..closing]; - if host.is_empty() { - return Err("missing TCP host".to_string()); - } - - let tail = &rest[closing + 1..]; - let port = if tail.is_empty() { - DEFAULT_TCP_PORT - } else if let Some(port) = tail.strip_prefix(':') { - parse_tcp_port(port)? - } else { - return Err("unexpected text after bracketed TCP host".to_string()); - }; - - return Ok(format!("[{host}]:{port}")); - } - - let colon_count = endpoint.matches(':').count(); - match colon_count { - 0 => Ok(format!("{endpoint}:{DEFAULT_TCP_PORT}")), - 1 => { - let (host, port) = endpoint - .rsplit_once(':') - .expect("colon_count guarantees a separator"); - if host.is_empty() { - return Err("missing TCP host".to_string()); - } - Ok(format!("{host}:{}", parse_tcp_port(port)?)) - } - _ => Ok(format!("[{endpoint}]:{DEFAULT_TCP_PORT}")), - } -} - -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -fn parse_tcp_port(port: &str) -> Result { - if port.is_empty() { - return Err("missing TCP port".to_string()); - } - port.parse::() - .map_err(|_| format!("invalid TCP port: {port}")) -} - -/// A unified sync I/O stream for either a serial port or a TCP socket. -/// -/// Both variants support `Read + Write + Send + 'static` so the existing -/// `spawn_blocking` read/write loops require minimal changes. -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -pub enum RNodeStream { - #[cfg(feature = "serial")] - Serial(Box), - Tcp(std::net::TcpStream), -} - -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -impl RNodeStream { - /// Open a serial port. - #[cfg(feature = "serial")] - pub fn open_serial(path: &str, baud: u32) -> std::io::Result { - let port = serialport::new(path, baud) - .timeout(Duration::from_millis(RNODE_READ_TIMEOUT_MS)) - .open() - .map_err(std::io::Error::other)?; - Ok(Self::Serial(port)) - } - - /// Connect to a TCP socket (blocking). - pub fn connect_tcp(addr: &str) -> std::io::Result { - Self::connect_tcp_with_timeout(addr, Duration::from_secs(RNODE_TCP_CONNECT_TIMEOUT_SECS)) - } - - fn connect_tcp_with_timeout(addr: &str, timeout: Duration) -> std::io::Result { - use std::net::ToSocketAddrs; - - let mut last_error = None; - for socket_addr in addr.to_socket_addrs()? { - match std::net::TcpStream::connect_timeout(&socket_addr, timeout) { - Ok(stream) => return Self::from_tcp_stream(stream), - Err(e) => last_error = Some(e), - } - } - - Err(last_error.unwrap_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::AddrNotAvailable, - format!("no socket addresses resolved for {addr}"), - ) - })) - } - - fn from_tcp_stream(stream: std::net::TcpStream) -> std::io::Result { - // Mirror the serial timeout so the read loop doesn't block forever. - stream.set_read_timeout(Some(Duration::from_millis(RNODE_READ_TIMEOUT_MS)))?; - stream.set_nodelay(true)?; - crate::socket_tuning::set_keepalive_tuned( - &stream, - Duration::from_secs(RNODE_TCP_KEEPIDLE_SECS), - Duration::from_secs(RNODE_TCP_KEEPINTVL_SECS), - RNODE_TCP_KEEPCNT, - Duration::from_secs(RNODE_TCP_USER_TIMEOUT_SECS), - ); - crate::socket_tuning::set_socket_buffers(&stream, RNODE_TCP_BUFFER_BYTES); - Ok(Self::Tcp(stream)) - } - - /// Shallow-clone the stream for the write half. - /// - /// - Serial: uses `SerialPort::try_clone`. - /// - TCP: uses `TcpStream::try_clone` (both halves share the same fd). - pub fn try_clone(&self) -> std::io::Result { - match self { - #[cfg(feature = "serial")] - Self::Serial(p) => Ok(Self::Serial(p.try_clone().map_err(std::io::Error::other)?)), - Self::Tcp(s) => Ok(Self::Tcp(s.try_clone()?)), - } - } - - /// Human-readable description for log messages. - pub fn description(&self) -> String { - match self { - #[cfg(feature = "serial")] - Self::Serial(p) => p.name().unwrap_or_else(|| "".to_string()), - Self::Tcp(s) => s - .peer_addr() - .map(|a| a.to_string()) - .unwrap_or_else(|_| "".to_string()), - } - } - - fn is_tcp(&self) -> bool { - matches!(self, Self::Tcp(_)) - } -} - -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -impl std::io::Read for RNodeStream { - fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - match self { - #[cfg(feature = "serial")] - Self::Serial(p) => p.read(buf), - Self::Tcp(s) => s.read(buf), - } - } -} - -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -impl std::io::Write for RNodeStream { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - match self { - #[cfg(feature = "serial")] - Self::Serial(p) => p.write(buf), - Self::Tcp(s) => s.write(buf), - } - } - - fn flush(&mut self) -> std::io::Result<()> { - match self { - #[cfg(feature = "serial")] - Self::Serial(p) => p.flush(), - Self::Tcp(s) => s.flush(), - } - } -} - -#[cfg(any(feature = "serial", feature = "rnode-tcp"))] -fn read_rnode_stream( - mut stream: RNodeStream, - mut buf: [u8; 1024], -) -> Result<(RNodeStream, [u8; 1024], usize), (RNodeStream, std::io::Error)> { - use std::io::Read; - - match stream.read(&mut buf) { - Ok(0) if stream.is_tcp() => Err(( - stream, - std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "RNode TCP socket closed"), - )), - Ok(n) => Ok((stream, buf, n)), - // Serial returns TimedOut; TCP returns WouldBlock on non-blocking - // or TimedOut on a read-timeout. Treat both as "no data yet". - Err(e) - if e.kind() == std::io::ErrorKind::TimedOut - || e.kind() == std::io::ErrorKind::WouldBlock => - { - Ok((stream, buf, 0)) - } - Err(e) => Err((stream, e)), - } -} +use crate::serial_tcp_stream::read_stream as read_rnode_stream; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] async fn open_configured_rnode_stream( @@ -526,16 +277,7 @@ async fn open_configured_rnode_stream( } #[cfg(any(feature = "serial", feature = "rnode-tcp"))] -fn reconnect_delay() -> Duration { - #[cfg(test)] - { - Duration::from_millis(100) - } - #[cfg(not(test))] - { - Duration::from_secs(RECONNECT_WAIT) - } -} +use crate::serial_tcp_stream::reconnect_delay; #[derive(Debug, Clone)] pub struct RNodeConfig { diff --git a/crates/rns-interface/src/serial_tcp_stream.rs b/crates/rns-interface/src/serial_tcp_stream.rs new file mode 100644 index 00000000..210d08a7 --- /dev/null +++ b/crates/rns-interface/src/serial_tcp_stream.rs @@ -0,0 +1,425 @@ +//! Shared serial/TCP transport for the KISS-family interfaces (KISS, RNode). +//! +//! A [`SerialTcpStream`] is either a local serial port or a TCP socket +//! behind one blocking `Read + Write` interface, so callers drive both the +//! same way: through `spawn_blocking` shuttles (see [`crate::serial_io`]) +//! for the data path, and [`read_stream`] for a single bounded read that +//! treats an idle timeout uniformly across both transports. +//! +//! Port selection is driven by the config string: +//! - `/dev/ttyUSB0`, `COM3`, etc. -> serial (feature `serial` required) +//! - `tcp://192.168.1.1` -> TCP, [`DEFAULT_TCP_PORT`] +//! - `tcp://192.168.1.1:9000` -> TCP, explicit port + +use std::time::Duration; + +/// Read timeout applied to both serial ports and TCP sockets so the caller's +/// read loop can periodically re-check its exit/online flag. +pub const READ_TIMEOUT_MS: u64 = 100; + +const TCP_CONNECT_TIMEOUT_SECS: u64 = 5; +const TCP_KEEPIDLE_SECS: u64 = 5; +const TCP_KEEPINTVL_SECS: u64 = 2; +const TCP_KEEPCNT: u32 = 12; +const TCP_USER_TIMEOUT_SECS: u64 = 24; +const TCP_BUFFER_BYTES: usize = 131_072; + +/// Delay between reconnect attempts after a transport drop. +pub const RECONNECT_WAIT_SECS: u64 = 5; + +/// Backoff before a driver retries opening its transport after a failure. +/// Shortened under `#[cfg(test)]` so reconnect tests don't sit idle. +pub fn reconnect_delay() -> Duration { + #[cfg(test)] + { + Duration::from_millis(100) + } + #[cfg(not(test))] + { + Duration::from_secs(RECONNECT_WAIT_SECS) + } +} + +/// Parsed representation of a `port` config field. +#[derive(Debug, Clone)] +pub enum PortConfig { + /// A local serial device path, e.g. `/dev/ttyUSB0` or `COM3`. + #[cfg(feature = "serial")] + Serial { path: String, baud: u32 }, + /// A TCP endpoint, e.g. `tcp://192.168.1.1` or `tcp://192.168.1.1:9000`. + Tcp { addr: String }, +} + +impl PortConfig { + pub fn parse(port: &str, baud: u32) -> Result { + #[cfg(not(feature = "serial"))] + let _ = baud; + + if let Some(rest) = strip_tcp_scheme(port) { + let addr = parse_tcp_endpoint(rest)?; + Ok(Self::Tcp { addr }) + } else { + #[cfg(feature = "serial")] + { + Ok(Self::Serial { + path: port.to_string(), + baud, + }) + } + #[cfg(not(feature = "serial"))] + Err( + "serial ports require the 'serial' feature; use tcp://host[:port] for TCP" + .to_string(), + ) + } + } +} + +fn strip_tcp_scheme(port: &str) -> Option<&str> { + const TCP_SCHEME: &str = "tcp://"; + port.get(..TCP_SCHEME.len()) + .filter(|prefix| prefix.eq_ignore_ascii_case(TCP_SCHEME)) + .and_then(|_| port.get(TCP_SCHEME.len()..)) +} + +fn parse_tcp_endpoint(endpoint: &str) -> Result { + if endpoint.is_empty() { + return Err("missing TCP host".to_string()); + } + + if let Some(rest) = endpoint.strip_prefix('[') { + let Some(closing) = rest.find(']') else { + return Err("missing closing ']' in IPv6 TCP host".to_string()); + }; + let host = &rest[..closing]; + if host.is_empty() { + return Err("missing TCP host".to_string()); + } + + let tail = &rest[closing + 1..]; + let port = if tail.is_empty() { + DEFAULT_TCP_PORT + } else if let Some(port) = tail.strip_prefix(':') { + parse_tcp_port(port)? + } else { + return Err("unexpected text after bracketed TCP host".to_string()); + }; + + return Ok(format!("[{host}]:{port}")); + } + + let colon_count = endpoint.matches(':').count(); + match colon_count { + 0 => Ok(format!("{endpoint}:{DEFAULT_TCP_PORT}")), + 1 => { + let (host, port) = endpoint + .rsplit_once(':') + .expect("colon_count guarantees a separator"); + if host.is_empty() { + return Err("missing TCP host".to_string()); + } + Ok(format!("{host}:{}", parse_tcp_port(port)?)) + } + _ => Ok(format!("[{endpoint}]:{DEFAULT_TCP_PORT}")), + } +} + +fn parse_tcp_port(port: &str) -> Result { + if port.is_empty() { + return Err("missing TCP port".to_string()); + } + port.parse::() + .map_err(|_| format!("invalid TCP port: {port}")) +} + +/// Default TCP port for KISS/RNode-over-IP. +pub const DEFAULT_TCP_PORT: u16 = 7633; + +/// A unified sync I/O stream for either a serial port or a TCP socket. +/// +/// Both variants support `Read + Write + Send + 'static` so blocking +/// read/write shuttles (`spawn_blocking`) require no per-transport branching +/// at the call site. +pub enum SerialTcpStream { + #[cfg(feature = "serial")] + Serial(Box), + Tcp(std::net::TcpStream), +} + +impl SerialTcpStream { + /// Open a serial port. + #[cfg(feature = "serial")] + pub fn open_serial(path: &str, baud: u32) -> std::io::Result { + let port = serialport::new(path, baud) + .timeout(Duration::from_millis(READ_TIMEOUT_MS)) + .open() + .map_err(std::io::Error::other)?; + Ok(Self::Serial(port)) + } + + /// Connect to a TCP socket (blocking) with the default connect timeout. + pub fn connect_tcp(addr: &str) -> std::io::Result { + Self::connect_tcp_with_timeout(addr, Duration::from_secs(TCP_CONNECT_TIMEOUT_SECS)) + } + + pub fn connect_tcp_with_timeout(addr: &str, timeout: Duration) -> std::io::Result { + use std::net::ToSocketAddrs; + + let mut last_error = None; + for socket_addr in addr.to_socket_addrs()? { + match std::net::TcpStream::connect_timeout(&socket_addr, timeout) { + Ok(stream) => return Self::from_tcp_stream(stream), + Err(e) => last_error = Some(e), + } + } + + Err(last_error.unwrap_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::AddrNotAvailable, + format!("no socket addresses resolved for {addr}"), + ) + })) + } + + fn from_tcp_stream(stream: std::net::TcpStream) -> std::io::Result { + // Mirror the serial timeout so the read loop doesn't block forever. + stream.set_read_timeout(Some(Duration::from_millis(READ_TIMEOUT_MS)))?; + stream.set_nodelay(true)?; + crate::socket_tuning::set_keepalive_tuned( + &stream, + Duration::from_secs(TCP_KEEPIDLE_SECS), + Duration::from_secs(TCP_KEEPINTVL_SECS), + TCP_KEEPCNT, + Duration::from_secs(TCP_USER_TIMEOUT_SECS), + ); + crate::socket_tuning::set_socket_buffers(&stream, TCP_BUFFER_BYTES); + Ok(Self::Tcp(stream)) + } + + /// Shallow-clone the stream for the write half. + /// + /// - Serial: uses `SerialPort::try_clone`. + /// - TCP: uses `TcpStream::try_clone` (both halves share the same fd). + pub fn try_clone(&self) -> std::io::Result { + match self { + #[cfg(feature = "serial")] + Self::Serial(p) => Ok(Self::Serial(p.try_clone().map_err(std::io::Error::other)?)), + Self::Tcp(s) => Ok(Self::Tcp(s.try_clone()?)), + } + } + + /// Human-readable description for log messages. + pub fn description(&self) -> String { + match self { + #[cfg(feature = "serial")] + Self::Serial(p) => p.name().unwrap_or_else(|| "".to_string()), + Self::Tcp(s) => s + .peer_addr() + .map(|a| a.to_string()) + .unwrap_or_else(|_| "".to_string()), + } + } + + pub fn is_tcp(&self) -> bool { + matches!(self, Self::Tcp(_)) + } +} + +impl std::io::Read for SerialTcpStream { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + match self { + #[cfg(feature = "serial")] + Self::Serial(p) => p.read(buf), + Self::Tcp(s) => s.read(buf), + } + } +} + +impl std::io::Write for SerialTcpStream { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + match self { + #[cfg(feature = "serial")] + Self::Serial(p) => p.write(buf), + Self::Tcp(s) => s.write(buf), + } + } + + fn flush(&mut self) -> std::io::Result<()> { + match self { + #[cfg(feature = "serial")] + Self::Serial(p) => p.flush(), + Self::Tcp(s) => s.flush(), + } + } +} + +/// One bounded synchronous read, meant to be run inside `spawn_blocking`. +/// +/// Serial idle timeouts and TCP `WouldBlock`/`TimedOut` both fold into +/// "no data yet" (`n == 0`, stream returned for reuse). A TCP `Ok(0)` means +/// the peer closed the connection and surfaces as a real error — unlike +/// serial, where `Ok(0)` is a normal empty read. +pub fn read_stream( + mut stream: SerialTcpStream, + mut buf: [u8; 1024], +) -> Result<(SerialTcpStream, [u8; 1024], usize), (SerialTcpStream, std::io::Error)> { + use std::io::Read; + + match stream.read(&mut buf) { + Ok(0) if stream.is_tcp() => Err(( + stream, + std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "TCP socket closed"), + )), + Ok(n) => Ok((stream, buf, n)), + Err(e) + if e.kind() == std::io::ErrorKind::TimedOut + || e.kind() == std::io::ErrorKind::WouldBlock => + { + Ok((stream, buf, 0)) + } + Err(e) => Err((stream, e)), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(feature = "serial")] + #[test] + fn test_port_config_serial() { + let cfg = PortConfig::parse("/dev/ttyUSB0", 115200).unwrap(); + assert!(matches!(cfg, PortConfig::Serial { path, .. } if path == "/dev/ttyUSB0")); + } + + #[test] + fn test_port_config_tcp_default_port() { + let cfg = PortConfig::parse("tcp://192.168.1.1", 115200).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "192.168.1.1:7633"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_port_config_tcp_explicit_port() { + let cfg = PortConfig::parse("tcp://192.168.1.1:9000", 115200).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "192.168.1.1:9000"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_port_config_tcp_hostname() { + let cfg = PortConfig::parse("tcp://rnode.local", 115200).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "rnode.local:7633"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_port_config_tcp_case_insensitive_scheme() { + let cfg = PortConfig::parse("TCP://rnode.local", 115200).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "rnode.local:7633"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_port_config_tcp_empty_host_rejected() { + let err = PortConfig::parse("tcp://", 115200).unwrap_err(); + assert!(err.contains("missing TCP host")); + } + + #[test] + fn test_port_config_tcp_invalid_port_rejected() { + let err = PortConfig::parse("tcp://rnode.local:notaport", 115200).unwrap_err(); + assert!(err.contains("invalid TCP port")); + } + + #[test] + fn test_port_config_tcp_missing_port_rejected() { + let err = PortConfig::parse("tcp://rnode.local:", 115200).unwrap_err(); + assert!(err.contains("missing TCP port")); + } + + #[test] + fn test_port_config_tcp_bracketed_ipv6_default_port() { + let cfg = PortConfig::parse("tcp://[2001:db8::1]", 115200).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "[2001:db8::1]:7633"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_port_config_tcp_bracketed_ipv6_explicit_port() { + let cfg = PortConfig::parse("tcp://[2001:db8::1]:9000", 115200).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "[2001:db8::1]:9000"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_port_config_tcp_unbracketed_ipv6_default_port() { + let cfg = PortConfig::parse("tcp://2001:db8::1", 115200).unwrap(); + match cfg { + PortConfig::Tcp { addr } => assert_eq!(addr, "[2001:db8::1]:7633"), + #[cfg(feature = "serial")] + _ => panic!("expected Tcp variant"), + } + } + + #[test] + fn test_port_config_tcp_malformed_bracketed_ipv6_rejected() { + let err = PortConfig::parse("tcp://[2001:db8::1", 115200).unwrap_err(); + assert!(err.contains("missing closing")); + } + + #[test] + fn test_tcp_eof_is_read_error() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let accept = std::thread::spawn(move || { + let (_stream, _) = listener.accept().unwrap(); + }); + + let stream = SerialTcpStream::connect_tcp(&addr.to_string()).unwrap(); + let _clone = stream.try_clone().unwrap(); + accept.join().unwrap(); + + match read_stream(stream, [0u8; 1024]) { + Ok(_) => panic!("closed TCP socket should be EOF"), + Err((_stream, err)) => assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof), + } + } + + #[test] + fn test_tcp_connect_accepts_timeout() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let accept = std::thread::spawn(move || { + let (_stream, _) = listener.accept().unwrap(); + }); + + let stream = SerialTcpStream::connect_tcp_with_timeout( + &addr.to_string(), + Duration::from_millis(500), + ) + .unwrap(); + assert!(stream.is_tcp()); + + drop(stream); + accept.join().unwrap(); + } +}