626 lines
24 KiB
Rust
626 lines
24 KiB
Rust
//! The user-facing [`Dongle`] — a connected, configured, ready-to-TX radio.
|
|
//!
|
|
//! Built on top of the internal [`crate::session::Session`]. Hides the
|
|
//! frame codec, tag plumbing, CRC, and keepalive details behind an async
|
|
//! method surface. Mirrors `client-py/donglora/dongle.py`.
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use donglora_protocol::{Info, Modulation, RxPayload, TxDonePayload};
|
|
use tokio::sync::Mutex;
|
|
use tokio::task::JoinHandle;
|
|
|
|
use crate::errors::{ClientError, ClientResult};
|
|
use crate::retry::{RetryPolicy, TxAttempt, TxOutcome};
|
|
use crate::session::Session;
|
|
|
|
/// Default keepalive cadence. Spec §3.4 sets the inactivity timer to
|
|
/// 1000 ms — we ping at 500 ms for 2x margin. Host commands reset the
|
|
/// firmware's timer naturally, so we only emit a PING when the host
|
|
/// has been quiet for a full keepalive interval.
|
|
pub const KEEPALIVE_INTERVAL: Duration = Duration::from_millis(500);
|
|
|
|
/// Live DongLoRa radio session.
|
|
///
|
|
/// Normally constructed via [`crate::connect`]. The happy path:
|
|
///
|
|
/// ```no_run
|
|
/// # async fn demo() -> Result<(), donglora_client::ClientError> {
|
|
/// use donglora_client as dl;
|
|
///
|
|
/// let dongle = dl::connect().await?;
|
|
/// dongle.tx(b"Hello").await?;
|
|
/// while let Some(pkt) = dongle.recv(std::time::Duration::from_secs(5)).await {
|
|
/// println!("{:.1} dBm {:?}", pkt.rssi_tenths_dbm as f32 / 10.0, pkt.data);
|
|
/// }
|
|
/// # Ok(()) }
|
|
/// ```
|
|
///
|
|
/// Thread-safe: multiple tasks can call `tx` / `recv` / `ping`
|
|
/// concurrently. Drop is best-effort — explicit `close().await` is
|
|
/// recommended when the caller can afford to await it.
|
|
/// Which transport `connect()` ended up using. Useful for logs / UIs
|
|
/// that want to distinguish "talking to the dongle directly" from
|
|
/// "talking to a mux that talks to the dongle".
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub enum TransportKind {
|
|
/// Direct USB serial. String is the device path (e.g. `"/dev/ttyACM0"`).
|
|
Serial(String),
|
|
/// `donglora-mux` daemon over a Unix domain socket. String is the
|
|
/// socket path.
|
|
MuxUnix(String),
|
|
/// `donglora-mux` daemon over TCP. String is `"host:port"`.
|
|
MuxTcp(String),
|
|
}
|
|
|
|
impl TransportKind {
|
|
/// Short human-readable label for this transport, suitable for a
|
|
/// status line (`"ttyACM0"`, `"mux:unix"`, `"mux:tcp"`).
|
|
#[must_use]
|
|
pub fn short_label(&self) -> String {
|
|
match self {
|
|
Self::Serial(p) => {
|
|
std::path::Path::new(p).file_name().map_or_else(|| p.clone(), |f| f.to_string_lossy().into_owned())
|
|
}
|
|
Self::MuxUnix(_) => "mux:unix".to_string(),
|
|
Self::MuxTcp(_) => "mux:tcp".to_string(),
|
|
}
|
|
}
|
|
|
|
/// True for any mux-backed transport.
|
|
#[must_use]
|
|
pub const fn is_mux(&self) -> bool {
|
|
matches!(self, Self::MuxUnix(_) | Self::MuxTcp(_))
|
|
}
|
|
}
|
|
|
|
pub struct Dongle {
|
|
inner: Arc<Inner>,
|
|
/// Keepalive task handle. `std::sync::Mutex` so `Drop` can abort
|
|
/// without an async context.
|
|
keepalive_handle: std::sync::Mutex<Option<JoinHandle<()>>>,
|
|
}
|
|
|
|
pub(crate) struct Inner {
|
|
pub(crate) session: Session,
|
|
info: Info,
|
|
transport_kind: TransportKind,
|
|
applied_config: Mutex<Option<Modulation>>,
|
|
rx_started: Mutex<bool>,
|
|
last_write_at: Mutex<tokio::time::Instant>,
|
|
closed: Mutex<bool>,
|
|
}
|
|
|
|
impl Dongle {
|
|
/// Construct a Dongle on an already-connected session. The caller is
|
|
/// responsible for running GET_INFO and (optionally) SET_CONFIG first;
|
|
/// [`crate::connect`] does both.
|
|
pub(crate) fn new(
|
|
session: Session,
|
|
info: Info,
|
|
transport_kind: TransportKind,
|
|
applied_config: Option<Modulation>,
|
|
keepalive: bool,
|
|
) -> Self {
|
|
let inner = Arc::new(Inner {
|
|
session,
|
|
info,
|
|
transport_kind,
|
|
applied_config: Mutex::new(applied_config),
|
|
rx_started: Mutex::new(false),
|
|
last_write_at: Mutex::new(tokio::time::Instant::now()),
|
|
closed: Mutex::new(false),
|
|
});
|
|
|
|
let keepalive_handle = if keepalive {
|
|
let weak = Arc::downgrade(&inner);
|
|
let handle = tokio::spawn(async move {
|
|
loop {
|
|
tokio::time::sleep(KEEPALIVE_INTERVAL).await;
|
|
let Some(inner) = weak.upgrade() else {
|
|
return;
|
|
};
|
|
if *inner.closed.lock().await {
|
|
return;
|
|
}
|
|
let last = *inner.last_write_at.lock().await;
|
|
if last.elapsed() < KEEPALIVE_INTERVAL {
|
|
continue;
|
|
}
|
|
match inner.session.ping(Duration::from_secs(2)).await {
|
|
Ok(()) => {
|
|
*inner.last_write_at.lock().await = tokio::time::Instant::now();
|
|
}
|
|
Err(ClientError::NotConfigured) => {
|
|
let cfg = *inner.applied_config.lock().await;
|
|
if let Some(m) = cfg {
|
|
match inner.session.set_config(m, Duration::from_secs(2)).await {
|
|
Ok(_) => {
|
|
*inner.last_write_at.lock().await = tokio::time::Instant::now();
|
|
}
|
|
Err(e) => {
|
|
tracing::debug!(?e, "keepalive recovery set_config failed");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
tracing::debug!(?e, "keepalive ping failed");
|
|
}
|
|
}
|
|
}
|
|
});
|
|
std::sync::Mutex::new(Some(handle))
|
|
} else {
|
|
std::sync::Mutex::new(None)
|
|
};
|
|
|
|
Self { inner, keepalive_handle }
|
|
}
|
|
|
|
/// Cached `GET_INFO` snapshot from connect-time.
|
|
pub fn info(&self) -> &Info {
|
|
&self.inner.info
|
|
}
|
|
|
|
/// Which transport this Dongle is backed by. Lets UIs distinguish
|
|
/// a direct USB connection from a mux-backed one.
|
|
pub fn transport_kind(&self) -> &TransportKind {
|
|
&self.inner.transport_kind
|
|
}
|
|
|
|
/// The most recently applied config (if any).
|
|
pub async fn config(&self) -> Option<Modulation> {
|
|
*self.inner.applied_config.lock().await
|
|
}
|
|
|
|
/// Transmit *data*. Blocks until `TX_DONE` arrives or an error
|
|
/// surfaces. Performs one auto-recovery retry on `NotConfigured`.
|
|
///
|
|
/// Use [`Dongle::tx_with_retry`] if you want CAD-busy backoff.
|
|
pub async fn tx(&self, data: &[u8]) -> ClientResult<TxDonePayload> {
|
|
self.tx_with_timeout(data, false, Duration::from_secs(10)).await
|
|
}
|
|
|
|
/// Transmit with a configurable timeout and the `skip_cad` flag.
|
|
pub async fn tx_with_timeout(&self, data: &[u8], skip_cad: bool, timeout: Duration) -> ClientResult<TxDonePayload> {
|
|
self.check_open().await?;
|
|
self.mark_write().await;
|
|
match self.inner.session.transmit(data, skip_cad, timeout).await {
|
|
Err(ClientError::NotConfigured) => self.recover_and_retry_tx(data, skip_cad, timeout).await,
|
|
other => other,
|
|
}
|
|
}
|
|
|
|
/// Transmit with automatic retry on `CHANNEL_BUSY` / `EBUSY` per
|
|
/// spec §6.10. Returns a [`TxOutcome`] capturing the full per-attempt
|
|
/// history (so callers — bridge's TUI especially — can surface retry
|
|
/// counts).
|
|
pub async fn tx_with_retry(&self, data: &[u8], policy: &RetryPolicy) -> ClientResult<TxOutcome> {
|
|
self.check_open().await?;
|
|
let mut attempts: Vec<TxAttempt> = Vec::with_capacity(policy.max_attempts as usize);
|
|
let mut delay_ms = policy.backoff_ms_min;
|
|
for attempt_num in 1..=policy.max_attempts {
|
|
self.mark_write().await;
|
|
let started = tokio::time::Instant::now();
|
|
let result = self.inner.session.transmit(data, policy.skip_cad, policy.per_attempt_timeout).await;
|
|
let elapsed = started.elapsed();
|
|
match result {
|
|
Ok(td) => {
|
|
attempts.push(TxAttempt { attempt: attempt_num, result: Ok(td), elapsed });
|
|
return Ok(TxOutcome { final_airtime_us: td.airtime_us, attempts });
|
|
}
|
|
Err(e) => {
|
|
let retryable = e.is_retryable();
|
|
let attempt_log = TxAttempt { attempt: attempt_num, result: Err(e.clone_kind()), elapsed };
|
|
attempts.push(attempt_log);
|
|
if !retryable || attempt_num == policy.max_attempts {
|
|
return Err(e);
|
|
}
|
|
let jitter_ms = policy.jitter_ms();
|
|
let wait = Duration::from_millis(u64::from(delay_ms) + u64::from(jitter_ms));
|
|
tokio::time::sleep(wait).await;
|
|
delay_ms = (delay_ms as f32 * policy.backoff_multiplier) as u32;
|
|
delay_ms = delay_ms.min(policy.backoff_cap_ms);
|
|
}
|
|
}
|
|
}
|
|
// Unreachable — loop either returns on success or errors on the
|
|
// final attempt. Surface a clean error just in case.
|
|
Err(ClientError::Other("tx_with_retry exited without result".into()))
|
|
}
|
|
|
|
/// Enter continuous receive mode if not already active.
|
|
pub async fn rx_start(&self) -> ClientResult<()> {
|
|
self.check_open().await?;
|
|
self.mark_write().await;
|
|
let mut started = self.inner.rx_started.lock().await;
|
|
if *started {
|
|
return Ok(());
|
|
}
|
|
match self.inner.session.rx_start(Duration::from_secs(2)).await {
|
|
Err(ClientError::NotConfigured) => {
|
|
drop(started);
|
|
self.reapply_config().await?;
|
|
self.inner.session.rx_start(Duration::from_secs(2)).await?;
|
|
*self.inner.rx_started.lock().await = true;
|
|
}
|
|
Err(e) => return Err(e),
|
|
Ok(()) => {
|
|
*started = true;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Exit continuous receive mode.
|
|
pub async fn rx_stop(&self) -> ClientResult<()> {
|
|
self.check_open().await?;
|
|
self.mark_write().await;
|
|
self.inner.session.rx_stop(Duration::from_secs(2)).await?;
|
|
*self.inner.rx_started.lock().await = false;
|
|
Ok(())
|
|
}
|
|
|
|
/// Wait up to *timeout* for the next RX event. Lazily starts continuous
|
|
/// RX on the first call. Returns `None` on timeout.
|
|
pub async fn recv(&self, timeout: Duration) -> Option<RxPayload> {
|
|
if self.check_open().await.is_err() {
|
|
return None;
|
|
}
|
|
if !*self.inner.rx_started.lock().await && self.rx_start().await.is_err() {
|
|
return None;
|
|
}
|
|
self.inner.session.recv_rx(timeout).await
|
|
}
|
|
|
|
/// Long-running listener: yields every RX event until the session
|
|
/// closes. Lazily starts continuous RX on first call.
|
|
pub async fn next_rx(&self) -> Option<RxPayload> {
|
|
if self.check_open().await.is_err() {
|
|
return None;
|
|
}
|
|
if !*self.inner.rx_started.lock().await && self.rx_start().await.is_err() {
|
|
return None;
|
|
}
|
|
self.inner.session.next_rx().await
|
|
}
|
|
|
|
/// Apply a new radio configuration (possibly a different modulation).
|
|
pub async fn set_config(&self, modulation: Modulation) -> ClientResult<()> {
|
|
self.check_open().await?;
|
|
self.mark_write().await;
|
|
let result = self.inner.session.set_config(modulation, Duration::from_secs(2)).await?;
|
|
// Cache the config for auto-recovery. We store what the firmware
|
|
// echoed back as `current` (authoritative), not what we sent.
|
|
*self.inner.applied_config.lock().await = Some(result.current);
|
|
// SET_CONFIG aborts any continuous RX on the device side.
|
|
*self.inner.rx_started.lock().await = false;
|
|
Ok(())
|
|
}
|
|
|
|
/// Send a `PING`. Rarely needed directly — the keepalive task covers
|
|
/// session liveness unless you constructed the Dongle with `keepalive = false`.
|
|
pub async fn ping(&self) -> ClientResult<()> {
|
|
self.check_open().await?;
|
|
self.mark_write().await;
|
|
self.inner.session.ping(Duration::from_secs(2)).await
|
|
}
|
|
|
|
/// Drain any async error frames (`ERR(EFRAME)`, `ERR(ERADIO)` with
|
|
/// tag 0, etc.) the reader observed since the last drain.
|
|
pub async fn drain_async_errors(&self) -> Vec<ClientError> {
|
|
self.inner.session.drain_async_errors().await
|
|
}
|
|
|
|
/// Close the session gracefully. Aborts the keepalive task and the
|
|
/// reader task. Idempotent.
|
|
pub async fn close(&self) {
|
|
let mut closed = self.inner.closed.lock().await;
|
|
if *closed {
|
|
return;
|
|
}
|
|
*closed = true;
|
|
drop(closed);
|
|
self.abort_keepalive();
|
|
self.inner.session.close().await;
|
|
}
|
|
|
|
fn abort_keepalive(&self) {
|
|
#[allow(clippy::unwrap_used)] // std::sync::Mutex poisoning is unrecoverable
|
|
let mut guard = self.keepalive_handle.lock().unwrap();
|
|
if let Some(h) = guard.take() {
|
|
h.abort();
|
|
}
|
|
}
|
|
|
|
// ── internals ─────────────────────────────────────────────────
|
|
|
|
async fn check_open(&self) -> ClientResult<()> {
|
|
if *self.inner.closed.lock().await {
|
|
return Err(ClientError::TransportClosed("dongle closed".into()));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn mark_write(&self) {
|
|
*self.inner.last_write_at.lock().await = tokio::time::Instant::now();
|
|
}
|
|
|
|
async fn reapply_config(&self) -> ClientResult<()> {
|
|
let cfg = (*self.inner.applied_config.lock().await).ok_or(ClientError::NotConfigured)?;
|
|
let result = self.inner.session.set_config(cfg, Duration::from_secs(2)).await?;
|
|
*self.inner.applied_config.lock().await = Some(result.current);
|
|
Ok(())
|
|
}
|
|
|
|
async fn recover_and_retry_tx(
|
|
&self,
|
|
data: &[u8],
|
|
skip_cad: bool,
|
|
timeout: Duration,
|
|
) -> ClientResult<TxDonePayload> {
|
|
tracing::info!("dongle: auto-recovering after inactivity timeout");
|
|
self.reapply_config().await?;
|
|
self.inner.session.transmit(data, skip_cad, timeout).await
|
|
}
|
|
}
|
|
|
|
impl Drop for Dongle {
|
|
fn drop(&mut self) {
|
|
// Best-effort cleanup. If the caller wanted a clean async
|
|
// shutdown they should have called `close().await` first; we
|
|
// can only abort the keepalive task here.
|
|
self.abort_keepalive();
|
|
}
|
|
}
|
|
|
|
// Give ClientError a cheap clone-like helper for recording it into a
|
|
// TxAttempt (which is owned), without requiring Clone on the whole
|
|
// error type (std::io::Error isn't Clone).
|
|
impl ClientError {
|
|
pub(crate) fn clone_kind(&self) -> ClientError {
|
|
match self {
|
|
Self::NotConfigured => Self::NotConfigured,
|
|
Self::Busy => Self::Busy,
|
|
Self::Param => Self::Param,
|
|
Self::Length => Self::Length,
|
|
Self::Modulation => Self::Modulation,
|
|
Self::UnknownCmd => Self::UnknownCmd,
|
|
Self::Radio => Self::Radio,
|
|
Self::Frame => Self::Frame,
|
|
Self::Internal => Self::Internal,
|
|
Self::UnknownCode(c) => Self::UnknownCode(*c),
|
|
Self::ChannelBusy => Self::ChannelBusy,
|
|
Self::Cancelled => Self::Cancelled,
|
|
Self::Timeout { what } => Self::Timeout { what },
|
|
Self::TransportClosed(s) => Self::TransportClosed(s.clone()),
|
|
Self::ReaderExited => Self::ReaderExited,
|
|
Self::BadFrame(s) => Self::BadFrame(s.clone()),
|
|
Self::EncodeFailed(s) => Self::EncodeFailed(s.clone()),
|
|
Self::Io(e) => Self::Other(format!("io: {e}")),
|
|
Self::ConfigNotSupported { reason } => Self::ConfigNotSupported { reason: reason.clone() },
|
|
Self::Other(s) => Self::Other(s.clone()),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[allow(clippy::unwrap_used, clippy::panic)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::retry::RetryPolicy;
|
|
use donglora_protocol::{
|
|
FrameDecoder, FrameResult, LoRaBandwidth, LoRaCodingRate, LoRaHeaderMode, Owner, RadioChipId, SetConfigResult,
|
|
SetConfigResultCode, TxResult, commands, encode_frame, events,
|
|
};
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
|
|
|
|
fn lora_cfg() -> Modulation {
|
|
Modulation::LoRa(donglora_protocol::LoRaConfig {
|
|
freq_hz: 910_525_000,
|
|
sf: 7,
|
|
bw: LoRaBandwidth::Khz62,
|
|
cr: LoRaCodingRate::Cr4_5,
|
|
preamble_len: 16,
|
|
sync_word: 0x1424,
|
|
tx_power_dbm: 20,
|
|
header_mode: LoRaHeaderMode::Explicit,
|
|
payload_crc: true,
|
|
iq_invert: false,
|
|
})
|
|
}
|
|
|
|
fn sample_info() -> Info {
|
|
Info {
|
|
proto_major: 1,
|
|
proto_minor: 0,
|
|
fw_major: 0,
|
|
fw_minor: 1,
|
|
fw_patch: 0,
|
|
radio_chip_id: RadioChipId::Sx1262.as_u16(),
|
|
capability_bitmap: donglora_protocol::cap::LORA | donglora_protocol::cap::CAD_BEFORE_TX,
|
|
supported_sf_bitmap: 0x1FE0,
|
|
supported_bw_bitmap: 0x03FF,
|
|
max_payload_bytes: 255,
|
|
rx_queue_capacity: 64,
|
|
tx_queue_capacity: 16,
|
|
freq_min_hz: 150_000_000,
|
|
freq_max_hz: 960_000_000,
|
|
tx_power_min_dbm: -9,
|
|
tx_power_max_dbm: 22,
|
|
mcu_uid_len: 0,
|
|
mcu_uid: [0u8; donglora_protocol::MAX_MCU_UID_LEN],
|
|
radio_uid_len: 0,
|
|
radio_uid: [0u8; donglora_protocol::MAX_RADIO_UID_LEN],
|
|
}
|
|
}
|
|
|
|
/// Spawn a fake device task on one side of a tokio duplex and return
|
|
/// both (a) a [`Dongle`] wired to the other side and (b) a handle to
|
|
/// the device so the test can drive responses on cue.
|
|
async fn drain_one_frame(device: &mut DuplexStream) -> (u8, u16, Vec<u8>) {
|
|
let mut decoder = FrameDecoder::new();
|
|
let mut buf = [0u8; 256];
|
|
loop {
|
|
let n = device.read(&mut buf).await.unwrap();
|
|
let mut out: Option<(u8, u16, Vec<u8>)> = None;
|
|
decoder.feed(&buf[..n], |res| match res {
|
|
FrameResult::Ok { type_id, tag, payload } => {
|
|
out = Some((type_id, tag, payload.to_vec()));
|
|
}
|
|
FrameResult::Err(e) => panic!("frame decode: {e:?}"),
|
|
});
|
|
if let Some(t) = out {
|
|
return t;
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn send_frame(device: &mut DuplexStream, type_id: u8, tag: u16, payload: &[u8]) {
|
|
let mut wire = [0u8; donglora_protocol::MAX_WIRE_FRAME];
|
|
let n = encode_frame(type_id, tag, payload, &mut wire).unwrap();
|
|
device.write_all(&wire[..n]).await.unwrap();
|
|
device.flush().await.unwrap();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn tx_success_path() {
|
|
let (host, mut device) = tokio::io::duplex(1024);
|
|
let session = Session::spawn(host);
|
|
let dongle =
|
|
Dongle::new(session, sample_info(), TransportKind::Serial("/dev/test".into()), Some(lora_cfg()), false);
|
|
|
|
let tx_task = tokio::spawn({
|
|
let dongle = dongle;
|
|
async move {
|
|
let td = dongle.tx(b"hello").await;
|
|
dongle.close().await;
|
|
td
|
|
}
|
|
});
|
|
|
|
let (type_id, tag, _p) = drain_one_frame(&mut device).await;
|
|
assert_eq!(type_id, commands::TYPE_TX);
|
|
send_frame(&mut device, events::TYPE_OK, tag, &[]).await;
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
|
|
let td = donglora_protocol::TxDonePayload { result: TxResult::Transmitted, airtime_us: 7_777 };
|
|
let mut td_buf = [0u8; donglora_protocol::TxDonePayload::WIRE_SIZE];
|
|
td.encode(&mut td_buf).unwrap();
|
|
send_frame(&mut device, events::TYPE_TX_DONE, tag, &td_buf).await;
|
|
|
|
let result = tx_task.await.unwrap().unwrap();
|
|
assert_eq!(result.airtime_us, 7_777);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn tx_with_retry_succeeds_after_channel_busy() {
|
|
let (host, mut device) = tokio::io::duplex(2048);
|
|
let session = Session::spawn(host);
|
|
let dongle =
|
|
Dongle::new(session, sample_info(), TransportKind::Serial("/dev/test".into()), Some(lora_cfg()), false);
|
|
|
|
let policy = RetryPolicy {
|
|
max_attempts: 3,
|
|
backoff_ms_min: 1,
|
|
backoff_ms_max: 1,
|
|
backoff_multiplier: 1.0,
|
|
backoff_cap_ms: 10,
|
|
per_attempt_timeout: Duration::from_secs(1),
|
|
skip_cad: false,
|
|
};
|
|
|
|
let tx_task = tokio::spawn({
|
|
let dongle = dongle;
|
|
let p = policy.clone();
|
|
async move {
|
|
let outcome = dongle.tx_with_retry(b"hi", &p).await;
|
|
dongle.close().await;
|
|
outcome
|
|
}
|
|
});
|
|
|
|
// Attempt 1 — CHANNEL_BUSY.
|
|
let (_t1, tag1, _) = drain_one_frame(&mut device).await;
|
|
send_frame(&mut device, events::TYPE_OK, tag1, &[]).await;
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
let mut td_buf = [0u8; donglora_protocol::TxDonePayload::WIRE_SIZE];
|
|
donglora_protocol::TxDonePayload { result: TxResult::ChannelBusy, airtime_us: 0 }.encode(&mut td_buf).unwrap();
|
|
send_frame(&mut device, events::TYPE_TX_DONE, tag1, &td_buf).await;
|
|
|
|
// Attempt 2 — TRANSMITTED.
|
|
let (_t2, tag2, _) = drain_one_frame(&mut device).await;
|
|
assert_ne!(tag1, tag2, "retry should use a fresh tag");
|
|
send_frame(&mut device, events::TYPE_OK, tag2, &[]).await;
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
donglora_protocol::TxDonePayload { result: TxResult::Transmitted, airtime_us: 9_000 }
|
|
.encode(&mut td_buf)
|
|
.unwrap();
|
|
send_frame(&mut device, events::TYPE_TX_DONE, tag2, &td_buf).await;
|
|
|
|
let outcome = tx_task.await.unwrap().unwrap();
|
|
assert_eq!(outcome.attempts_used(), 2);
|
|
assert!(outcome.had_retries());
|
|
assert_eq!(outcome.final_airtime_us, 9_000);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn tx_auto_recovers_from_not_configured() {
|
|
let (host, mut device) = tokio::io::duplex(2048);
|
|
let session = Session::spawn(host);
|
|
let cfg = lora_cfg();
|
|
let dongle = Dongle::new(session, sample_info(), TransportKind::Serial("/dev/test".into()), Some(cfg), false);
|
|
|
|
let task = tokio::spawn({
|
|
let dongle = dongle;
|
|
async move {
|
|
let r = dongle.tx(b"abc").await;
|
|
dongle.close().await;
|
|
r
|
|
}
|
|
});
|
|
|
|
// First TX → ERR(ENOTCONFIGURED).
|
|
let (_t, tag1, _) = drain_one_frame(&mut device).await;
|
|
let mut err_buf = [0u8; 2];
|
|
donglora_protocol::events::encode_err_payload(donglora_protocol::ErrorCode::ENotConfigured, &mut err_buf)
|
|
.unwrap();
|
|
send_frame(&mut device, events::TYPE_ERR, tag1, &err_buf).await;
|
|
|
|
// Auto-recovery: SET_CONFIG goes out. Reply APPLIED/MINE.
|
|
let (set_type, tag_set, _) = drain_one_frame(&mut device).await;
|
|
assert_eq!(set_type, commands::TYPE_SET_CONFIG);
|
|
let result = SetConfigResult { result: SetConfigResultCode::Applied, owner: Owner::Mine, current: cfg };
|
|
let mut rbuf = [0u8; donglora_protocol::MAX_SETCONFIG_OK_PAYLOAD];
|
|
let n = result.encode(&mut rbuf).unwrap();
|
|
send_frame(&mut device, events::TYPE_OK, tag_set, &rbuf[..n]).await;
|
|
|
|
// Retry TX → succeed.
|
|
let (_t, tag2, _) = drain_one_frame(&mut device).await;
|
|
send_frame(&mut device, events::TYPE_OK, tag2, &[]).await;
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
let mut td_buf = [0u8; donglora_protocol::TxDonePayload::WIRE_SIZE];
|
|
donglora_protocol::TxDonePayload { result: TxResult::Transmitted, airtime_us: 1234 }
|
|
.encode(&mut td_buf)
|
|
.unwrap();
|
|
send_frame(&mut device, events::TYPE_TX_DONE, tag2, &td_buf).await;
|
|
|
|
let td = task.await.unwrap().unwrap();
|
|
assert_eq!(td.airtime_us, 1234);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn drop_aborts_keepalive() {
|
|
let (host, _device) = tokio::io::duplex(512);
|
|
let session = Session::spawn(host);
|
|
let dongle = Dongle::new(session, sample_info(), TransportKind::Serial("/dev/test".into()), None, true);
|
|
// Keepalive is running. Dropping the Dongle must abort it.
|
|
assert!(dongle.keepalive_handle.lock().unwrap().is_some());
|
|
drop(dongle);
|
|
// Give the abort a tick to land.
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
}
|