diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..cef926a --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,27 @@ +# Changelog + +## 0.1.0 — 2026-04-06 + +Initial release. + +### Features + +- High-level `Client` with send/recv and command helpers (ping, set_config, + start_rx, stop_rx, transmit, get_config, get_mac, display_on, display_off) +- COBS wire framing via `ucobs` (matches firmware implementation) +- Auto-detection connection: TCP mux, Unix socket mux, direct USB serial +- Bounded RX packet buffering (256 packets, FIFO eviction) +- `FrameReader` accumulator for streaming byte sources +- USB device discovery by VID:PID with blocking wait + +### Resilience + +- Cross-platform timeout handling: `TimedOut` (Windows) and `WouldBlock` + (Linux/macOS) both treated as clean timeouts in `read_frame` +- `EINTR`/`Interrupted` signals retried automatically in `read_frame` +- `drain_rx` always restores the original timeout, even on I/O errors +- TCP mux connections use `connect_timeout` (bounded by caller's timeout) +- Mux sockets set both read and write timeouts +- `SerialTransport` tracks timeout accurately for save/restore +- Unexpected unsolicited frames logged via `tracing::warn` instead of silently + discarded diff --git a/src/client.rs b/src/client.rs index 9bb0d90..c9142b9 100644 --- a/src/client.rs +++ b/src/client.rs @@ -46,7 +46,10 @@ impl Client { let data = read_frame(&mut self.transport)? .ok_or_else(|| anyhow::anyhow!("timeout waiting for response"))?; let resp = Response::from_bytes(&data) - .ok_or_else(|| anyhow::anyhow!("malformed response ({} bytes)", data.len()))?; + .ok_or_else(|| { + let hex: Vec = data.iter().map(|b| format!("{b:02x}")).collect(); + anyhow::anyhow!("malformed response ({} bytes): [{}]", data.len(), hex.join(" ")) + })?; if resp.is_rx_packet() { self.buffer_rx(resp); continue; @@ -71,7 +74,7 @@ impl Client { if resp.is_rx_packet() { Ok(Some(resp)) } else { - // Non-RxPacket unsolicited response — shouldn't happen, discard + tracing::warn!("recv: discarding unexpected unsolicited response: {resp:?}"); Ok(None) } } @@ -87,18 +90,22 @@ impl Client { self.transport .set_timeout(std::time::Duration::from_millis(10))?; - loop { - let Some(data) = read_frame(&mut self.transport)? else { - break; - }; - if let Some(resp) = Response::from_bytes(&data) - && resp.is_rx_packet() - { - packets.push(resp); + let result: anyhow::Result<()> = (|| { + loop { + let Some(data) = read_frame(&mut self.transport)? else { + return Ok(()); + }; + if let Some(resp) = Response::from_bytes(&data) + && resp.is_rx_packet() + { + packets.push(resp); + } } - } + })(); + // Always restore timeout, even if the loop errored self.transport.set_timeout(old_timeout)?; + result?; Ok(packets) } diff --git a/src/codec.rs b/src/codec.rs index 9ad5e4d..fcdcfde 100644 --- a/src/codec.rs +++ b/src/codec.rs @@ -107,9 +107,15 @@ pub fn read_frame(reader: &mut dyn std::io::Read) -> anyhow::Result { + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock + ) => + { return Ok(None); } + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, Err(e) => return Err(e.into()), } } @@ -274,4 +280,51 @@ mod tests { let decoded = decode_frame(&encoded[..encoded.len() - 1]); assert_eq!(decoded, Some(data)); } + + #[test] + fn config_response_cobs_roundtrip() { + // Exact Config response bytes: tag=1, freq=910525000, bw=62.5k, sf=7, + // cr=5, sync=0x3444, tx=22dBm, preamble=16, cad=1 + // Note: preamble high byte is 0x00 — tests COBS with embedded zero. + let data = vec![ + 0x01, 0x48, 0x82, 0x45, 0x36, 0x06, 0x07, 0x05, + 0x44, 0x34, 0x16, 0x10, 0x00, 0x01, + ]; + assert_eq!(data.len(), 14); + let encoded = encode_frame(&data); + let decoded = decode_frame(&encoded[..encoded.len() - 1]); + assert_eq!(decoded, Some(data)); + } + + #[test] + fn config_response_cobs_roundtrip_cad_zero() { + // Same but with cad=0 — two consecutive 0x00 bytes at end. + let data = vec![ + 0x01, 0x48, 0x82, 0x45, 0x36, 0x06, 0x07, 0x05, + 0x44, 0x34, 0x16, 0x10, 0x00, 0x00, + ]; + assert_eq!(data.len(), 14); + let encoded = encode_frame(&data); + let decoded = decode_frame(&encoded[..encoded.len() - 1]); + assert_eq!(decoded, Some(data)); + } + + #[cfg(unix)] + #[test] + fn read_frame_socket_timeout_returns_none() { + use std::os::unix::net::UnixStream; + use std::time::Duration; + + let (mut a, _b) = UnixStream::pair().unwrap(); + a.set_read_timeout(Some(Duration::from_millis(10))) + .unwrap(); + + // No data written to _b, so read on `a` will time out with WouldBlock on Linux. + let result = read_frame(&mut a); + assert!( + result.is_ok(), + "socket timeout should be Ok(None), got: {result:?}" + ); + assert!(result.unwrap().is_none()); + } } diff --git a/src/connect.rs b/src/connect.rs index 4af999d..2b56b86 100644 --- a/src/connect.rs +++ b/src/connect.rs @@ -131,6 +131,36 @@ pub fn connect_default() -> anyhow::Result> { connect(None, DEFAULT_TIMEOUT) } +/// Connect to a mux daemon only (TCP via env var, then Unix socket). +/// +/// Unlike [`connect`], this **never** falls back to direct USB serial. +/// Returns an error if no mux is reachable — the caller can retry with backoff. +/// +/// This is the Rust equivalent of the Python client's "sticky mux" behavior: +/// once you decide to use the mux, you stay on the mux. +pub fn connect_mux_auto(timeout: Duration) -> anyhow::Result> { + // Try TCP mux via environment variable. + if let Ok(tcp) = std::env::var("DONGLORA_MUX_TCP") + && let Some(transport) = try_tcp_mux(&tcp, timeout) + { + debug!("connected to TCP mux at {tcp}"); + return Ok(Client::new(AnyTransport::Mux(transport))); + } + + // Try Unix socket mux. + #[cfg(unix)] + { + let path = find_mux_socket() + .ok_or_else(|| anyhow::anyhow!("no mux socket found"))?; + let transport = MuxTransport::unix(&path, timeout)?; + debug!("connected to mux socket at {path}"); + Ok(Client::new(AnyTransport::Mux(transport))) + } + + #[cfg(not(unix))] + anyhow::bail!("mux-only mode requires Unix socket support or DONGLORA_MUX_TCP") +} + fn try_tcp_mux(addr: &str, timeout: Duration) -> Option { let (host, port) = if let Some((h, p)) = addr.rsplit_once(':') { let host = if h.is_empty() { "localhost" } else { h }; diff --git a/src/lib.rs b/src/lib.rs index 91c3be9..ec67a35 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -31,7 +31,7 @@ pub mod transport; // Flat re-exports for convenience pub use client::Client; pub use codec::{decode_frame, encode_frame, read_frame, FrameReader}; -pub use connect::{connect, connect_default, default_socket_path}; +pub use connect::{connect, connect_default, connect_mux_auto, default_socket_path}; pub use discovery::{find_port, wait_for_device, USB_PID, USB_VID}; pub use protocol::{ Bandwidth, Command, ErrorCode, RadioConfig, Response, CMD_TAG_SET_CONFIG, CMD_TAG_START_RX, diff --git a/src/transport.rs b/src/transport.rs index 101dca5..735f8cc 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -20,6 +20,7 @@ pub trait Transport: Read + Write + Send { /// Direct USB serial port connection to a DongLoRa dongle. pub struct SerialTransport { port: Box, + timeout: Duration, } impl SerialTransport { @@ -31,7 +32,7 @@ impl SerialTransport { .timeout(timeout) .open() .map_err(|e| anyhow::anyhow!("failed to open serial port {path}: {e}"))?; - Ok(Self { port }) + Ok(Self { port, timeout }) } /// Clear the serial input buffer. @@ -62,13 +63,13 @@ impl Transport for SerialTransport { fn set_timeout(&mut self, timeout: Duration) -> anyhow::Result<()> { self.port .set_timeout(timeout) - .map_err(|e| anyhow::anyhow!("failed to set serial timeout: {e}")) + .map_err(|e| anyhow::anyhow!("failed to set serial timeout: {e}"))?; + self.timeout = timeout; + Ok(()) } fn timeout(&self) -> Duration { - // serialport doesn't have a getter, so we'd need to track it ourselves. - // For now, return a sensible default. The actual timeout is set on the port. - Duration::from_secs(2) + self.timeout } } @@ -94,7 +95,10 @@ impl MuxTransport { .map_err(|e| anyhow::anyhow!("failed to connect to mux socket {path}: {e}"))?; stream .set_read_timeout(Some(timeout)) - .map_err(|e| anyhow::anyhow!("failed to set socket timeout: {e}"))?; + .map_err(|e| anyhow::anyhow!("failed to set socket read timeout: {e}"))?; + stream + .set_write_timeout(Some(timeout)) + .map_err(|e| anyhow::anyhow!("failed to set socket write timeout: {e}"))?; Ok(Self { stream: MuxStream::Unix(stream), timeout, @@ -103,12 +107,22 @@ impl MuxTransport { /// Connect to the mux daemon via TCP. pub fn tcp(host: &str, port: u16, timeout: Duration) -> anyhow::Result { + use std::net::ToSocketAddrs; + let addr = format!("{host}:{port}"); - let stream = std::net::TcpStream::connect(&addr) + let sock_addr = addr + .to_socket_addrs() + .map_err(|e| anyhow::anyhow!("failed to resolve mux address {addr}: {e}"))? + .next() + .ok_or_else(|| anyhow::anyhow!("no addresses found for {addr}"))?; + let stream = std::net::TcpStream::connect_timeout(&sock_addr, timeout) .map_err(|e| anyhow::anyhow!("failed to connect to mux at {addr}: {e}"))?; stream .set_read_timeout(Some(timeout)) - .map_err(|e| anyhow::anyhow!("failed to set TCP timeout: {e}"))?; + .map_err(|e| anyhow::anyhow!("failed to set TCP read timeout: {e}"))?; + stream + .set_write_timeout(Some(timeout)) + .map_err(|e| anyhow::anyhow!("failed to set TCP write timeout: {e}"))?; Ok(Self { stream: MuxStream::Tcp(stream), timeout, @@ -149,8 +163,12 @@ impl Transport for MuxTransport { self.timeout = timeout; let result = match &self.stream { #[cfg(unix)] - MuxStream::Unix(s) => s.set_read_timeout(Some(timeout)), - MuxStream::Tcp(s) => s.set_read_timeout(Some(timeout)), + MuxStream::Unix(s) => s + .set_read_timeout(Some(timeout)) + .and_then(|()| s.set_write_timeout(Some(timeout))), + MuxStream::Tcp(s) => s + .set_read_timeout(Some(timeout)) + .and_then(|()| s.set_write_timeout(Some(timeout))), }; result.map_err(|e| anyhow::anyhow!("failed to set timeout: {e}")) }