feat!: Protocol v2 client rewrite with Dongle/Session model
The full 1.0 client surface, matching the protocol v2 wire definitions from donglora-protocol 1.0.0.
* Public API is now async. The blocking Client<T: Transport> is replaced by Dongle (from donglora_client::connect / connect_with / ConnectOptions).
* src/client.rs and src/codec.rs are removed in favor of src/dongle.rs (high-level surface), src/session.rs (per-client tag/slot state), src/retry.rs (RetryPolicy + tx_with_retry), src/errors.rs (thiserror-based ClientError / ClientResult). src/protocol.rs is replaced by direct donglora-protocol imports.
* Transport is now tokio-serial for USB and tokio::net::{UnixStream, TcpStream} for mux, replacing the blocking serial + shim model.
* Dongle::tx_with_retry(data, &RetryPolicy) implements PROTOCOL.md §6.10 randomized backoff for CHANNEL_BUSY / EBUSY, returning a TxOutcome with per-attempt trace.
* Background keepalive task pings every 500 ms on host idle, tracking the 1 s inactivity timeout from §3.4 with 2x margin. Disable via ConnectOptions::keepalive(false).
* Auto-recovery from ERR(ENOTCONFIGURED): the Dongle caches the last SET_CONFIG and silently re-applies on session drop, retrying once.
* Error taxonomy switches from anyhow to thiserror. Transport-level conditions (Timeout, TransportClosed, ReaderExited, BadFrame) are first-class.
* Examples rewritten for the async API: examples/ping.rs, examples/tx.rs, examples/rx.rs all target the MeshCore US preset (910.525 MHz / BW 62.5 / SF 7 / CR 4/5 / sync 0x1424 / 20 dBm). examples/common/mod.rs holds the shared helper.
* tests/vectors.rs gains protocol-level roundtrip coverage anchored to the published donglora-protocol 1.0.0 types.
This commit is contained in:
19 files changed
+2897
-1605
No files matched your search
+142
-147
@@ -1,212 +1,207 @@
|
||||
//! Transport abstraction for byte-stream connections.
|
||||
//! Async transport abstractions.
|
||||
//!
|
||||
//! The [`Transport`] trait unifies serial ports and mux socket connections,
|
||||
//! allowing the [`Client`](crate::client::Client) to work over any of them.
|
||||
//! The [`Transport`] marker trait unifies tokio-based byte streams so the
|
||||
//! [`Session`](crate::session::Session) can drive any of them. Three
|
||||
//! concrete transports are provided:
|
||||
//!
|
||||
//! - [`SerialTransport`] — `tokio-serial` USB CDC-ACM.
|
||||
//! - [`UnixSocketTransport`] — mux daemon over `tokio::net::UnixStream`.
|
||||
//! - [`TcpTransport`] — mux daemon over `tokio::net::TcpStream`.
|
||||
//!
|
||||
//! Most callers use [`AnyTransport`] so the concrete variant is decided
|
||||
//! at `connect()` time without leaking type parameters into user code.
|
||||
|
||||
use std::io::{self, Read, Write};
|
||||
use std::io;
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
use std::time::Duration;
|
||||
|
||||
/// A byte-stream transport that the client can talk over.
|
||||
pub trait Transport: Read + Write + Send {
|
||||
/// Set the read timeout. A read that takes longer returns zero bytes.
|
||||
fn set_timeout(&mut self, timeout: Duration) -> anyhow::Result<()>;
|
||||
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
|
||||
use tokio::net::TcpStream;
|
||||
#[cfg(unix)]
|
||||
use tokio::net::UnixStream;
|
||||
use tokio_serial::{SerialPortBuilderExt, SerialStream};
|
||||
|
||||
/// Get the current read timeout.
|
||||
fn timeout(&self) -> Duration;
|
||||
}
|
||||
use crate::errors::{ClientError, ClientResult};
|
||||
|
||||
/// Marker for any async byte-stream we can carry DongLoRa Protocol frames over.
|
||||
///
|
||||
/// This is intentionally a blanket trait so the [`Session`](crate::session::Session)
|
||||
/// can accept `SerialStream`, `UnixStream`, `TcpStream`, or the
|
||||
/// type-erased [`AnyTransport`] uniformly.
|
||||
pub trait Transport: AsyncRead + AsyncWrite + Unpin + Send + 'static {}
|
||||
impl<T: AsyncRead + AsyncWrite + Unpin + Send + 'static> Transport for T {}
|
||||
|
||||
// ── Serial transport ───────────────────────────────────────────────
|
||||
|
||||
/// Direct USB serial port connection to a DongLoRa dongle.
|
||||
/// Direct USB serial port connection (tokio-serial).
|
||||
pub struct SerialTransport {
|
||||
port: Box<dyn serialport::SerialPort>,
|
||||
timeout: Duration,
|
||||
inner: SerialStream,
|
||||
}
|
||||
|
||||
impl SerialTransport {
|
||||
/// Open a serial port with the given timeout.
|
||||
///
|
||||
/// Baud rate is irrelevant for USB CDC-ACM but we pass a conventional value.
|
||||
pub fn open(path: &str, timeout: Duration) -> anyhow::Result<Self> {
|
||||
let port = serialport::new(path, 115_200)
|
||||
.timeout(timeout)
|
||||
.open()
|
||||
.map_err(|e| anyhow::anyhow!("failed to open serial port {path}: {e}"))?;
|
||||
Ok(Self { port, timeout })
|
||||
}
|
||||
|
||||
/// Clear the serial input buffer.
|
||||
pub fn reset_input_buffer(&self) -> anyhow::Result<()> {
|
||||
self.port
|
||||
.clear(serialport::ClearBuffer::Input)
|
||||
.map_err(|e| anyhow::anyhow!("failed to clear input buffer: {e}"))
|
||||
/// Open a serial port. Baud rate is irrelevant for USB CDC-ACM but
|
||||
/// `tokio-serial` still demands a value; 115200 is conventional.
|
||||
pub fn open(path: &str) -> ClientResult<Self> {
|
||||
let inner = tokio_serial::new(path, 115_200)
|
||||
.open_native_async()
|
||||
.map_err(|e| ClientError::Other(format!("failed to open serial port {path}: {e}")))?;
|
||||
Ok(Self { inner })
|
||||
}
|
||||
}
|
||||
|
||||
impl Read for SerialTransport {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
self.port.read(buf)
|
||||
impl AsyncRead for SerialTransport {
|
||||
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_read(cx, buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for SerialTransport {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
self.port.write(buf)
|
||||
impl AsyncWrite for SerialTransport {
|
||||
fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
|
||||
Pin::new(&mut self.inner).poll_write(cx, buf)
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
self.port.flush()
|
||||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_flush(cx)
|
||||
}
|
||||
|
||||
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_shutdown(cx)
|
||||
}
|
||||
}
|
||||
|
||||
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}"))?;
|
||||
self.timeout = timeout;
|
||||
Ok(())
|
||||
}
|
||||
// ── Unix socket transport ──────────────────────────────────────────
|
||||
|
||||
fn timeout(&self) -> Duration {
|
||||
self.timeout
|
||||
/// Mux daemon connection over a Unix domain socket.
|
||||
#[cfg(unix)]
|
||||
pub struct UnixSocketTransport {
|
||||
inner: UnixStream,
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
impl UnixSocketTransport {
|
||||
/// Connect to the mux daemon at `path`.
|
||||
pub async fn connect(path: &str) -> ClientResult<Self> {
|
||||
let inner = UnixStream::connect(path)
|
||||
.await
|
||||
.map_err(|e| ClientError::Other(format!("failed to connect to mux socket {path}: {e}")))?;
|
||||
Ok(Self { inner })
|
||||
}
|
||||
}
|
||||
|
||||
// ── Mux transport ──────────────────────────────────────────────────
|
||||
|
||||
/// Connection to the DongLoRa mux daemon via Unix socket or TCP.
|
||||
pub struct MuxTransport {
|
||||
stream: MuxStream,
|
||||
timeout: Duration,
|
||||
#[cfg(unix)]
|
||||
impl AsyncRead for UnixSocketTransport {
|
||||
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_read(cx, buf)
|
||||
}
|
||||
}
|
||||
|
||||
enum MuxStream {
|
||||
#[cfg(unix)]
|
||||
Unix(std::os::unix::net::UnixStream),
|
||||
Tcp(std::net::TcpStream),
|
||||
}
|
||||
|
||||
impl MuxTransport {
|
||||
/// Connect to the mux daemon via Unix domain socket.
|
||||
#[cfg(unix)]
|
||||
pub fn unix(path: &str, timeout: Duration) -> anyhow::Result<Self> {
|
||||
let stream = std::os::unix::net::UnixStream::connect(path)
|
||||
.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 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 })
|
||||
#[cfg(unix)]
|
||||
impl AsyncWrite for UnixSocketTransport {
|
||||
fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
|
||||
Pin::new(&mut self.inner).poll_write(cx, buf)
|
||||
}
|
||||
|
||||
/// Connect to the mux daemon via TCP.
|
||||
pub fn tcp(host: &str, port: u16, timeout: Duration) -> anyhow::Result<Self> {
|
||||
use std::net::ToSocketAddrs;
|
||||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_flush(cx)
|
||||
}
|
||||
|
||||
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_shutdown(cx)
|
||||
}
|
||||
}
|
||||
|
||||
// ── TCP transport ──────────────────────────────────────────────────
|
||||
|
||||
/// Mux daemon connection over TCP. Uses `TCP_NODELAY` because DongLoRa Protocol frames
|
||||
/// are small and latency-sensitive.
|
||||
pub struct TcpTransport {
|
||||
inner: TcpStream,
|
||||
}
|
||||
|
||||
impl TcpTransport {
|
||||
/// Connect to `host:port`. A failed connect attempt returns an error
|
||||
/// immediately; no internal retries (callers decide policy).
|
||||
pub async fn connect(host: &str, port: u16, timeout: Duration) -> ClientResult<Self> {
|
||||
let addr = format!("{host}:{port}");
|
||||
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 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 })
|
||||
let fut = TcpStream::connect(&addr);
|
||||
let stream = tokio::time::timeout(timeout, fut)
|
||||
.await
|
||||
.map_err(|_| ClientError::Timeout { what: "tcp connect" })?
|
||||
.map_err(|e| ClientError::Other(format!("failed to connect to {addr}: {e}")))?;
|
||||
stream.set_nodelay(true).map_err(|e| ClientError::Other(format!("failed to set TCP_NODELAY: {e}")))?;
|
||||
Ok(Self { inner: stream })
|
||||
}
|
||||
}
|
||||
|
||||
impl Read for MuxTransport {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
match &mut self.stream {
|
||||
#[cfg(unix)]
|
||||
MuxStream::Unix(s) => s.read(buf),
|
||||
MuxStream::Tcp(s) => s.read(buf),
|
||||
}
|
||||
impl AsyncRead for TcpTransport {
|
||||
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_read(cx, buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for MuxTransport {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
match &mut self.stream {
|
||||
#[cfg(unix)]
|
||||
MuxStream::Unix(s) => s.write(buf),
|
||||
MuxStream::Tcp(s) => s.write(buf),
|
||||
}
|
||||
impl AsyncWrite for TcpTransport {
|
||||
fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
|
||||
Pin::new(&mut self.inner).poll_write(cx, buf)
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
match &mut self.stream {
|
||||
#[cfg(unix)]
|
||||
MuxStream::Unix(s) => s.flush(),
|
||||
MuxStream::Tcp(s) => s.flush(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Transport for MuxTransport {
|
||||
fn set_timeout(&mut self, timeout: Duration) -> anyhow::Result<()> {
|
||||
self.timeout = timeout;
|
||||
let result = match &self.stream {
|
||||
#[cfg(unix)]
|
||||
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}"))
|
||||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_flush(cx)
|
||||
}
|
||||
|
||||
fn timeout(&self) -> Duration {
|
||||
self.timeout
|
||||
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Pin::new(&mut self.inner).poll_shutdown(cx)
|
||||
}
|
||||
}
|
||||
|
||||
// ── AnyTransport ───────────────────────────────────────────────────
|
||||
|
||||
/// Type-erased transport for the [`connect`](crate::connect::connect) return type.
|
||||
/// Type-erased transport returned by [`crate::connect`]. One of the three
|
||||
/// concrete variants depending on which transport the auto-discovery chain
|
||||
/// succeeded on.
|
||||
pub enum AnyTransport {
|
||||
Serial(SerialTransport),
|
||||
Mux(MuxTransport),
|
||||
#[cfg(unix)]
|
||||
Unix(UnixSocketTransport),
|
||||
Tcp(TcpTransport),
|
||||
}
|
||||
|
||||
impl Read for AnyTransport {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
match self {
|
||||
Self::Serial(t) => t.read(buf),
|
||||
Self::Mux(t) => t.read(buf),
|
||||
impl AsyncRead for AnyTransport {
|
||||
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
||||
match self.get_mut() {
|
||||
Self::Serial(t) => Pin::new(t).poll_read(cx, buf),
|
||||
#[cfg(unix)]
|
||||
Self::Unix(t) => Pin::new(t).poll_read(cx, buf),
|
||||
Self::Tcp(t) => Pin::new(t).poll_read(cx, buf),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for AnyTransport {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
match self {
|
||||
Self::Serial(t) => t.write(buf),
|
||||
Self::Mux(t) => t.write(buf),
|
||||
impl AsyncWrite for AnyTransport {
|
||||
fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
|
||||
match self.get_mut() {
|
||||
Self::Serial(t) => Pin::new(t).poll_write(cx, buf),
|
||||
#[cfg(unix)]
|
||||
Self::Unix(t) => Pin::new(t).poll_write(cx, buf),
|
||||
Self::Tcp(t) => Pin::new(t).poll_write(cx, buf),
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
match self {
|
||||
Self::Serial(t) => t.flush(),
|
||||
Self::Mux(t) => t.flush(),
|
||||
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
match self.get_mut() {
|
||||
Self::Serial(t) => Pin::new(t).poll_flush(cx),
|
||||
#[cfg(unix)]
|
||||
Self::Unix(t) => Pin::new(t).poll_flush(cx),
|
||||
Self::Tcp(t) => Pin::new(t).poll_flush(cx),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Transport for AnyTransport {
|
||||
fn set_timeout(&mut self, timeout: Duration) -> anyhow::Result<()> {
|
||||
match self {
|
||||
Self::Serial(t) => t.set_timeout(timeout),
|
||||
Self::Mux(t) => t.set_timeout(timeout),
|
||||
}
|
||||
}
|
||||
|
||||
fn timeout(&self) -> Duration {
|
||||
match self {
|
||||
Self::Serial(t) => t.timeout(),
|
||||
Self::Mux(t) => t.timeout(),
|
||||
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
match self.get_mut() {
|
||||
Self::Serial(t) => Pin::new(t).poll_shutdown(cx),
|
||||
#[cfg(unix)]
|
||||
Self::Unix(t) => Pin::new(t).poll_shutdown(cx),
|
||||
Self::Tcp(t) => Pin::new(t).poll_shutdown(cx),
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user