canboat 9.1.0

// (C) 2009-2026, Kees Verruijt, Harlingen, The Netherlands.

//! Device-level read/write abstraction shared by the standalone
//! reader binaries (actisense-serial, ikonvert-serial, maretron-ipg)
//! and the integrated `n2kd-pipeline` binary.
//!
//! Each supported device speaks its own wire protocol (NGT-1 binary,
//! iKonvert ASCII, Maretron 0xA5 TCP), but the threading shape is
//! identical: a reader thread drains device bytes through a sans-I/O
//! decoder, and a writer thread drains a command channel back out to
//! the device. Both threads are owned by the runner here; the
//! per-device codec lives behind two small traits ([`DeviceDecoder`]
//! and [`DeviceEncoder`]) so the runner stays oblivious to the
//! protocol. The gateways with a public sans-I/O
//! [`Codec`](crate::engine::codec::Codec) (NGT-1, iKonvert, Maretron)
//! run through [`run_codec`], which shares the one codec between both
//! threads and supplies the clock.
//!
//! A typical use, through the public [`Device`](open::Device), which
//! opens the port and calls the runner:
//!
//! ```no_run
//! use canboat::bus::{BusProtocol, Device, Kind};
//! let handle = Device::new(Kind::NGT1, "/dev/ttyUSB0").open(BusProtocol::Nmea2000)?;
//! while let Ok(frame) = handle.frames_rx.recv() {
//!     println!("got pgn {}", frame.pgn);
//! }
//! # Ok::<(), std::io::Error>(())
//! ```

// The `raw` feature alone builds the runner here and the raw links; the
// rest, NMEA 2000 gateways and nodes, is the `io` feature's.
#[cfg(feature = "io")]
pub mod bst95;
#[cfg(feature = "io")]
pub mod bst_d0;
#[cfg(feature = "io")]
mod can_link;
#[cfg(feature = "io")]
pub mod canalyst;
#[cfg(feature = "io")]
pub mod canboat_csv;
#[cfg(feature = "io")]
pub mod candump;
#[cfg(feature = "io")]
pub mod ikonvert;
#[cfg(feature = "io")]
mod iso11783_node;
pub mod kind;
#[cfg(feature = "io")]
pub mod line_gateway;
#[cfg(feature = "io")]
pub mod maretron;
#[cfg(feature = "io")]
pub mod ngt1;
pub mod open;
pub mod raw;
#[cfg(feature = "io")]
pub mod replay;
#[cfg(feature = "io")]
pub mod socketcan;
pub mod supervisor;
#[cfg(feature = "io")]
pub mod usb_can_a;

pub use supervisor::DeviceFactory;

use std::io::{self, Read, Write};
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::engine::RawFrame;
use crate::engine::codec::{Codec, Event};

/// Events emitted by a [`DeviceDecoder`] as bytes are pumped in.
#[derive(Debug)]
pub enum DeviceEvent<F = RawFrame> {
    /// A complete N2K frame received from the bus (or, on a raw link, a
    /// single CAN frame).
    Frame(F),
    /// The codec needs the runner to write these bytes back to the
    /// device. Used by codecs whose handshake response depends on
    /// device output (e.g. Maretron `SET_MODE BINARY` after seeing
    /// `CONNECTED`).
    SendBytes(Vec<u8>),
    /// Decoded a framing or protocol error. The runner logs it.
    Error(String),
    /// The device is no use (see [`Codec::failure`]): the runner stops
    /// reading, and [`DeviceHandle::failure`] tells why.
    Fatal(String),
}

/// Decoder half of a device codec: pushes received bytes and emits
/// [`DeviceEvent`]s.
pub trait DeviceDecoder<F = RawFrame>: Send + 'static {
    fn decode(&mut self, bytes: &[u8], events: &mut Vec<DeviceEvent<F>>);

    /// Called when a read times out with nothing received (a serial port
    /// does every 250 ms), so a decoder's own deadlines advance on a quiet
    /// line too. Nothing by default.
    fn tick(&mut self, _events: &mut Vec<DeviceEvent<F>>) {}
}

/// Encoder half: turns app-side [`RawFrame`]s into device bytes, plus
/// optional init / keepalive sequences.
pub trait DeviceEncoder<F = RawFrame>: Send + Sync + 'static {
    /// Bytes to write as soon as the connection opens (NGT-1 startup
    /// ping, iKonvert init handshake, Maretron CONNECT). Empty when
    /// the device needs no handshake.
    fn init_bytes(&self) -> Vec<u8> {
        Vec::new()
    }

    /// Optional periodic keepalive: `(interval, payload)`. `None`
    /// means the writer thread never wakes on a timer.
    fn keepalive(&self) -> Option<(Duration, Vec<u8>)> {
        None
    }

    /// Encode a [`RawFrame`] for transmission. `None` means "silently
    /// drop" — used to skip synthetic PGNs that never hit the bus.
    fn encode_frame(&self, frame: &F) -> Option<Vec<u8>>;

    /// Bytes to write when the application closes the device on purpose
    /// ([`DeviceHandle::close`]), e.g. to take a gateway off the bus.
    /// Empty when the device needs nothing.
    fn shutdown_bytes(&self) -> Vec<u8> {
        Vec::new()
    }

    /// The least time from one frame written to the next, for a device
    /// that loses frames written faster than it sends them. Zero by
    /// default.
    fn write_gap(&self) -> Duration {
        Duration::ZERO
    }

    /// When [`Self::poll`] next has bytes to write, in Unix milliseconds
    /// (the host clock): the next packet of an ISO TP transfer. The
    /// writer wakes for it. `None` by default.
    fn next_deadline(&self) -> Option<u64> {
        None
    }

    /// The bytes due at `now_ms`.
    fn poll(&self, _now_ms: u64) -> Vec<u8> {
        Vec::new()
    }
}

/// Commands accepted by the writer thread.
pub(crate) enum WriterCmd<F = RawFrame> {
    Bytes(Vec<u8>),
    Frame(F),
    /// Write the encoder's [`DeviceEncoder::shutdown_bytes`] and stop,
    /// then tell the sender, if any, whether that succeeded.
    Shutdown(Option<mpsc::Sender<bool>>),
}

/// Returned by [`run`]. Owns the reader/writer threads' join handles
/// and the channels used to talk to them. `F` is what goes in and out:
/// whole NMEA 2000 messages, or single CAN frames on a raw link.
pub struct DeviceHandle<F = RawFrame> {
    /// Frames received from the device. The reader thread drops the
    /// sender on EOF / fatal error, which closes this channel.
    pub frames_rx: mpsc::Receiver<F>,
    cmd_tx: mpsc::Sender<WriterCmd<F>>,
    joins: Vec<JoinHandle<()>>,
    /// The writer thread, joined apart from the others by [`Self::close`].
    /// `None` for a codec that has no separate writer (SocketCAN).
    writer: Option<JoinHandle<()>>,
    /// Writes the writer thread has completed, so a close can tell a
    /// writer still draining a backlog from a stuck one.
    progress: Arc<AtomicU64>,
    /// Why the device was given up on, when it was.
    failure: DeviceFailure,
    /// Why the device's frames stopped, when the device went away.
    lost: DeviceFailure,
    /// A supervised device's: set on close, join or drop, so that it stops
    /// reconnecting.
    stop: Option<StopOnDrop>,
    /// The address canboat's node on the device holds, [`CLAIM_UNCLAIMED`]
    /// while it has none; `None` where canboat runs no node.
    claim: Option<Arc<AtomicU8>>,
}

/// What [`DeviceHandle::claimed_address`] reads while canboat's node holds
/// no address.
pub(crate) const CLAIM_UNCLAIMED: u8 = 254;

/// Sets its flag when dropped: a supervised [`DeviceHandle`]'s, so that
/// letting go of the handle stops the reconnecting too.
struct StopOnDrop(Arc<AtomicBool>);

impl Drop for StopOnDrop {
    fn drop(&mut self) {
        self.0.store(true, Ordering::Relaxed);
    }
}

/// Why a device stopped: see [`DeviceHandle::failure`] and
/// [`DeviceHandle::lost`]. Clone it out of a [`DeviceHandle`] to ask once
/// the frames have stopped.
#[derive(Debug, Clone, Default)]
pub struct DeviceFailure(Arc<std::sync::OnceLock<String>>);

impl DeviceFailure {
    /// The reason, once there is one.
    pub fn get(&self) -> Option<&str> {
        self.0.get().map(String::as_str)
    }

    /// Record `why`, unless a reason is already there.
    pub(crate) fn set(&self, why: &str) {
        let _ = self.0.set(why.to_owned());
    }
}

/// Returned by [`DeviceHandle::send_frame`] / [`FrameSender::send_frame`]
/// when the writer thread has gone away (either the device closed or
/// the runner shut down).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DeviceWriterGone;

impl std::fmt::Display for DeviceWriterGone {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("device writer is no longer accepting frames")
    }
}

impl std::error::Error for DeviceWriterGone {}

impl<F: Send + 'static> DeviceHandle<F> {
    /// Why the device was given up on, if it is: its frames stop then.
    /// Clone it to ask once they have.
    pub fn failure(&self) -> DeviceFailure {
        self.failure.clone()
    }

    /// Why the device's frames stopped, when the device went away without
    /// being closed: a read error (a gateway that reset the connection), its
    /// end of stream, or the device's own `--timeout`. Unlike a
    /// [`failure`](Self::failure), reconnecting may mend it.
    pub fn lost(&self) -> DeviceFailure {
        self.lost.clone()
    }

    /// The source address canboat's node on the device holds, if it runs
    /// one and has claimed an address.
    pub fn claimed_address(&self) -> Option<u8> {
        self.claim
            .as_ref()
            .map(|claim| claim.load(Ordering::Relaxed))
            .filter(|&address| address < CLAIM_UNCLAIMED)
    }

    /// The node's live claim address, for the parts of `canboat server`
    /// that stamp it on what they send.
    #[cfg_attr(not(feature = "bridge"), allow(dead_code))]
    pub(crate) fn claim_atom(&self) -> Option<Arc<AtomicU8>> {
        self.claim.clone()
    }

    /// Have [`claimed_address`](Self::claimed_address) read `claim`.
    #[cfg_attr(not(feature = "io"), allow(dead_code))]
    pub(crate) fn with_claim(mut self, claim: Arc<AtomicU8>) -> Self {
        self.claim = Some(claim);
        self
    }

    /// Queue a frame to be encoded and written to the device.
    pub fn send_frame(&self, frame: F) -> Result<(), DeviceWriterGone> {
        self.cmd_tx
            .send(WriterCmd::Frame(frame))
            .map_err(|_| DeviceWriterGone)
    }

    /// Clone-able sender, for handing off to a producer thread.
    pub fn frame_sender(&self) -> FrameSender<F> {
        FrameSender {
            cmd_tx: self.cmd_tx.clone(),
        }
    }

    /// Signal shutdown to the writer thread and wait for both threads
    /// to exit.
    pub fn join(self) {
        // A supervised device stops reconnecting.
        drop(self.stop);
        // Nothing receives any more: a reader that only stops when its
        // send fails (replay) must see that, or the join waits for ever.
        drop(self.frames_rx);
        drop(self.cmd_tx);
        for j in self.joins.into_iter().chain(self.writer) {
            let _ = j.join();
        }
    }

    /// Close the device on purpose: the writer finishes what is queued,
    /// sends the encoder's [`DeviceEncoder::shutdown_bytes`] (an iKonvert
    /// goes off the bus) and stops. See [`DeviceCloser::close`] for the
    /// wait; a writer that stops making progress is left behind rather
    /// than hanging shutdown. The reader thread is not joined: it may be
    /// blocked reading the device, and ends with it.
    pub fn close(self) -> Closed {
        let closed = self.closer().close(CLOSE_TIMEOUT);
        match closed {
            Closed::Confirmed | Closed::Gone => {
                if let Some(writer) = self.writer {
                    let _ = writer.join();
                }
            }
            Closed::Unconfirmed => {
                log::warn!("the device did not confirm closing; leaving it");
            }
        }
        closed
    }

    /// A handle another thread can [`close`](DeviceCloser::close) the
    /// device with — for a signal handler while this thread is busy
    /// reading.
    pub fn closer(&self) -> DeviceCloser<F> {
        DeviceCloser {
            cmd_tx: self.cmd_tx.clone(),
            progress: self.progress.clone(),
            stop: self.stop.as_ref().map(|stop| stop.0.clone()),
        }
    }
}

/// How long [`DeviceHandle::close`] waits for the device to confirm, or —
/// while it is still draining a backlog — for its next completed write.
pub const CLOSE_TIMEOUT: Duration = Duration::from_secs(2);

/// How a [`close`](DeviceHandle::close) went.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Closed {
    /// Everything queued and the goodbye (if any) were written and
    /// flushed.
    Confirmed,
    /// The device had already gone: its writer had stopped (the device
    /// was unplugged, or timed out), so there was nothing left to close.
    Gone,
    /// The goodbye could not be written, or the writer stopped making
    /// progress: a gateway may still be on the bus.
    Unconfirmed,
}

/// Closes a device from another thread; see [`DeviceHandle::closer`].
pub struct DeviceCloser<F = RawFrame> {
    cmd_tx: mpsc::Sender<WriterCmd<F>>,
    progress: Arc<AtomicU64>,
    /// A supervised device's stop flag.
    stop: Option<Arc<AtomicBool>>,
}

impl<F> Clone for DeviceCloser<F> {
    fn clone(&self) -> Self {
        Self {
            cmd_tx: self.cmd_tx.clone(),
            progress: self.progress.clone(),
            stop: self.stop.clone(),
        }
    }
}

impl<F> DeviceCloser<F> {
    /// [`DeviceHandle::close`] from elsewhere: the writer finishes what is
    /// queued, sends the encoder's shutdown bytes and stops. Waits for that
    /// as long as the writer keeps completing writes, giving up only after
    /// `timeout` without one — a long backlog still drains, a stuck writer
    /// does not hang the caller.
    pub fn close(&self, timeout: Duration) -> Closed {
        // A supervised device: no reconnecting after this.
        if let Some(stop) = &self.stop {
            stop.store(true, Ordering::Relaxed);
        }
        let (done_tx, done_rx) = mpsc::channel();
        if self
            .cmd_tx
            .send(WriterCmd::Shutdown(Some(done_tx)))
            .is_err()
        {
            return Closed::Gone;
        }
        loop {
            let before = self.progress.load(Ordering::Relaxed);
            match done_rx.recv_timeout(timeout) {
                Ok(true) => return Closed::Confirmed,
                Ok(false) => return Closed::Unconfirmed,
                // The writer ended without getting to the shutdown: it
                // stopped on a failed write, the device is gone.
                Err(mpsc::RecvTimeoutError::Disconnected) => return Closed::Gone,
                Err(mpsc::RecvTimeoutError::Timeout)
                    if self.progress.load(Ordering::Relaxed) != before => {}
                Err(mpsc::RecvTimeoutError::Timeout) => return Closed::Unconfirmed,
            }
        }
    }
}

/// Clone-able sender for [`DeviceHandle::send_frame`].
pub struct FrameSender<F = RawFrame> {
    cmd_tx: mpsc::Sender<WriterCmd<F>>,
}

impl<F> Clone for FrameSender<F> {
    fn clone(&self) -> Self {
        Self {
            cmd_tx: self.cmd_tx.clone(),
        }
    }
}

impl<F> FrameSender<F> {
    pub fn send_frame(&self, frame: F) -> Result<(), DeviceWriterGone> {
        self.cmd_tx
            .send(WriterCmd::Frame(frame))
            .map_err(|_| DeviceWriterGone)
    }
}

#[cfg(test)]
impl<F> FrameSender<F> {
    /// A sender whose frames arrive on the receiver, with no device behind it.
    pub(crate) fn for_test() -> (Self, mpsc::Receiver<WriterCmd<F>>) {
        let (cmd_tx, cmd_rx) = mpsc::channel();
        (Self { cmd_tx }, cmd_rx)
    }
}

/// Build a [`DeviceHandle`] from already-wired channels and thread
/// handles. Used by codecs whose I/O model isn't byte-stream-shaped
/// (e.g. `socketcan`) and can't use the generic [`run`] runner below.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
///
/// `progress` counts the device's completed writes, as the generic
/// writer's does, so a close can tell a draining backlog from a stuck one.
/// `lost` is the codec's to set when the device goes away.
pub(crate) fn from_parts<F>(
    frames_rx: mpsc::Receiver<F>,
    cmd_tx: mpsc::Sender<WriterCmd<F>>,
    joins: Vec<JoinHandle<()>>,
    progress: Arc<AtomicU64>,
    lost: DeviceFailure,
) -> DeviceHandle<F> {
    DeviceHandle {
        frames_rx,
        cmd_tx,
        joins,
        writer: None,
        progress,
        failure: DeviceFailure::default(),
        lost,
        stop: None,
        claim: None,
    }
}

/// Generic two-thread runner. Each codec module wraps this with its
/// own [`DeviceDecoder`] / [`DeviceEncoder`] impls.
pub fn run<F: Send + 'static, D: DeviceDecoder<F>, E: DeviceEncoder<F>>(
    decoder: D,
    encoder: E,
    mut reader: Box<dyn Read + Send>,
    mut writer: Box<dyn Write + Send>,
) -> DeviceHandle<F> {
    let (frames_tx, frames_rx) = mpsc::channel::<F>();
    let (cmd_tx, cmd_rx) = mpsc::channel::<WriterCmd<F>>();
    let cmd_tx_for_reader = cmd_tx.clone();

    let init = encoder.init_bytes();
    let keepalive = encoder.keepalive();
    let progress = Arc::new(AtomicU64::new(0));
    let writer_progress = progress.clone();

    let writer_join = thread::Builder::new()
        .name("device-writer".into())
        .spawn(move || {
            writer_thread(
                &mut *writer,
                encoder,
                init,
                keepalive,
                cmd_rx,
                &writer_progress,
            )
        })
        .expect("spawn device writer");

    let failure = DeviceFailure::default();
    let reader_failure = failure.clone();
    let lost = DeviceFailure::default();
    let reader_lost = lost.clone();
    let reader_join = thread::Builder::new()
        .name("device-reader".into())
        .spawn(move || {
            reader_thread(
                &mut *reader,
                decoder,
                frames_tx,
                cmd_tx_for_reader,
                &reader_failure,
                &reader_lost,
            )
        })
        .expect("spawn device reader");

    DeviceHandle {
        frames_rx,
        cmd_tx,
        joins: vec![reader_join],
        writer: Some(writer_join),
        progress,
        failure,
        lost,
        stop: None,
        claim: None,
    }
}

/// Run a sans-I/O [`Codec`] under [`run`]: the reader thread feeds it
/// the bytes it reads, the writer thread has it encode frames, and both
/// pass it the host clock. `fixtime`, when set, replaces the timestamp of
/// every frame it produces (canboat's `-fixtime`, for reproducible output).
pub fn run_codec<C: Codec + Send + 'static>(
    codec: C,
    fixtime: Option<String>,
    reader: Box<dyn Read + Send>,
    writer: Box<dyn Write + Send>,
) -> DeviceHandle {
    let codec = Arc::new(Mutex::new(codec));
    let stray = Arc::new(Mutex::new(Vec::new()));
    let decoder = CodecDecoder {
        codec: codec.clone(),
        fixtime,
        events: Vec::with_capacity(8),
        stray: stray.clone(),
    };
    run(decoder, CodecEncoder { codec, stray }, reader, writer)
}

/// The host clock in Unix milliseconds, as the codecs take it.
pub(crate) fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// The locked codec; a panic in the other thread does not lose it.
fn lock<C>(codec: &Mutex<C>) -> std::sync::MutexGuard<'_, C> {
    codec
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// The reader thread's view of a shared [`Codec`].
struct CodecDecoder<C> {
    codec: Arc<Mutex<C>>,
    fixtime: Option<String>,
    events: Vec<Event>,
    /// What a tick on the writer's deadline produced that is not bytes to
    /// write, for the reader to deliver.
    stray: Arc<Mutex<Vec<Event>>>,
}

impl<C> CodecDecoder<C> {
    fn deliver(&mut self, out: &mut Vec<DeviceEvent>) {
        let mut stray = std::mem::take(&mut *lock(&self.stray));
        stray.append(&mut self.events);
        out.extend(stray.into_iter().map(|ev| match ev {
            Event::Frame(mut f) => {
                if let Some(ts) = &self.fixtime {
                    f.timestamp = Some(ts.clone());
                }
                DeviceEvent::Frame(f)
            }
            Event::Send(b) => DeviceEvent::SendBytes(b),
            Event::Error(e) => DeviceEvent::Error(e),
        }));
    }
}

impl<C: Codec + Send + 'static> DeviceDecoder for CodecDecoder<C> {
    fn decode(&mut self, bytes: &[u8], events: &mut Vec<DeviceEvent>) {
        let failure = {
            let mut codec = lock(&self.codec);
            codec.receive(bytes, now_ms(), &mut self.events);
            codec.failure().map(str::to_owned)
        };
        self.deliver(events);
        events.extend(failure.map(DeviceEvent::Fatal));
    }

    fn tick(&mut self, events: &mut Vec<DeviceEvent>) {
        let failure = {
            let mut codec = lock(&self.codec);
            codec.tick(now_ms(), &mut self.events);
            codec.failure().map(str::to_owned)
        };
        self.deliver(events);
        events.extend(failure.map(DeviceEvent::Fatal));
    }
}

/// The writer thread's view of a shared [`Codec`].
struct CodecEncoder<C> {
    codec: Arc<Mutex<C>>,
    stray: Arc<Mutex<Vec<Event>>>,
}

impl<C: Codec + Send + 'static> DeviceEncoder for CodecEncoder<C> {
    fn init_bytes(&self) -> Vec<u8> {
        lock(&self.codec).open()
    }

    fn keepalive(&self) -> Option<(Duration, Vec<u8>)> {
        lock(&self.codec).keepalive()
    }

    fn encode_frame(&self, frame: &RawFrame) -> Option<Vec<u8>> {
        match lock(&self.codec).send_at(frame, now_ms()) {
            Ok(bytes) => Some(bytes),
            Err(why) => {
                log::debug!("not sending PGN {}: {why}", frame.pgn);
                None
            }
        }
    }

    fn shutdown_bytes(&self) -> Vec<u8> {
        lock(&self.codec).close()
    }

    fn next_deadline(&self) -> Option<u64> {
        lock(&self.codec).next_deadline()
    }

    /// Tick the codec at its deadline: the bytes it sends are written, and
    /// anything else it has is left for the reader.
    fn poll(&self, now_ms: u64) -> Vec<u8> {
        let mut events = Vec::new();
        lock(&self.codec).tick(now_ms, &mut events);
        let mut bytes = Vec::new();
        let mut stray = lock(&self.stray);
        for ev in events {
            match ev {
                Event::Send(b) => bytes.extend(b),
                ev => stray.push(ev),
            }
        }
        bytes
    }
}

fn reader_thread<F, D: DeviceDecoder<F>>(
    reader: &mut dyn Read,
    mut decoder: D,
    frames_tx: mpsc::Sender<F>,
    cmd_tx: mpsc::Sender<WriterCmd<F>>,
    failure: &DeviceFailure,
    lost: &DeviceFailure,
) {
    let mut buf = [0u8; 4096];
    let mut events: Vec<DeviceEvent<F>> = Vec::with_capacity(8);
    // Hand the decoder's events on; `false` once a channel is gone.
    let deliver = |events: &mut Vec<DeviceEvent<F>>| -> bool {
        for ev in events.drain(..) {
            match ev {
                DeviceEvent::Frame(f) => {
                    if frames_tx.send(f).is_err() {
                        return false;
                    }
                }
                DeviceEvent::SendBytes(b) => {
                    if cmd_tx.send(WriterCmd::Bytes(b)).is_err() {
                        return false;
                    }
                }
                DeviceEvent::Error(e) => {
                    log::warn!("device decode error: {e}");
                }
                DeviceEvent::Fatal(e) => {
                    log::error!("{e}");
                    failure.set(&e);
                    return false;
                }
            }
        }
        true
    };
    loop {
        match reader.read(&mut buf) {
            Ok(0) => {
                lost.set("the device ended the connection");
                return;
            }
            Ok(n) => {
                decoder.decode(&buf[..n], &mut events);
                if !deliver(&mut events) {
                    return;
                }
            }
            Err(e) if is_read_timeout(&e) => {
                decoder.tick(&mut events);
                if !deliver(&mut events) {
                    return;
                }
            }
            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
            Err(e) => {
                let why = format!("device read error: {e}");
                log::error!("{why}");
                lost.set(&why);
                return;
            }
        }
    }
}

/// Whether a read error is only the read timing out, to read on after.
/// A serial port's timeout is TimedOut, made up by the serialport crate;
/// a socket's read timeout is WouldBlock on Unix, TimedOut on Windows.
/// On Unix an ETIMEDOUT from the OS is the connection timing out instead:
/// its TCP keepalive got no answer from the gateway (canboat#1039).
fn is_read_timeout(e: &io::Error) -> bool {
    match e.kind() {
        io::ErrorKind::WouldBlock => true,
        io::ErrorKind::TimedOut => cfg!(windows) || e.raw_os_error().is_none(),
        _ => false,
    }
}

fn writer_thread<F, E: DeviceEncoder<F>>(
    writer: &mut dyn Write,
    encoder: E,
    init: Vec<u8>,
    keepalive: Option<(Duration, Vec<u8>)>,
    rx: mpsc::Receiver<WriterCmd<F>>,
    progress: &AtomicU64,
) {
    let gap = encoder.write_gap();
    let mut last_frame: Option<Instant> = None;
    if !init.is_empty() {
        if let Err(e) = writer.write_all(&init) {
            log::error!("device init write failed: {e}");
            return;
        }
        let _ = writer.flush();
    }

    let mut next_keepalive = keepalive.as_ref().map(|(d, _)| Instant::now() + *d);

    loop {
        let now = Instant::now();
        // The encoder's own deadline (ISO TP pacing), on the host clock.
        let due = encoder
            .next_deadline()
            .map(|at| now + Duration::from_millis(at.saturating_sub(now_ms())));
        let wake = match (due, next_keepalive) {
            (Some(d), Some(k)) => Some(d.min(k)),
            (d, k) => d.or(k),
        };
        let cmd = match wake {
            Some(wake) => match rx.recv_timeout(wake.saturating_duration_since(now)) {
                Ok(cmd) => cmd,
                Err(mpsc::RecvTimeoutError::Timeout) if due == Some(wake) => {
                    match encoder.poll(now_ms()) {
                        b if b.is_empty() => continue,
                        b => WriterCmd::Bytes(b),
                    }
                }
                Err(mpsc::RecvTimeoutError::Timeout) => {
                    let Some((interval, payload)) = &keepalive else {
                        continue;
                    };
                    if let Err(e) = writer.write_all(payload) {
                        log::warn!("device keepalive failed: {e}");
                        return;
                    }
                    let _ = writer.flush();
                    next_keepalive = Some(Instant::now() + *interval);
                    continue;
                }
                Err(mpsc::RecvTimeoutError::Disconnected) => return,
            },
            None => match rx.recv() {
                Ok(cmd) => cmd,
                Err(_) => return,
            },
        };

        let bytes = match cmd {
            WriterCmd::Bytes(b) => b,
            WriterCmd::Frame(f) => match encoder.encode_frame(&f) {
                Some(b) => {
                    // Space frames for a device that loses them otherwise.
                    if let Some(last) = last_frame {
                        let wait = (last + gap).saturating_duration_since(Instant::now());
                        if !wait.is_zero() {
                            thread::sleep(wait);
                        }
                    }
                    if !gap.is_zero() {
                        last_frame = Some(Instant::now());
                    }
                    b
                }
                None => continue,
            },
            WriterCmd::Shutdown(done) => {
                let bytes = encoder.shutdown_bytes();
                let sent = bytes.is_empty()
                    || match writer.write_all(&bytes).and_then(|()| writer.flush()) {
                        Ok(()) => true,
                        Err(e) => {
                            log::warn!("device shutdown write failed: {e}");
                            false
                        }
                    };
                if let Some(done) = done {
                    let _ = done.send(sent);
                }
                return;
            }
        };
        if let Err(e) = writer.write_all(&bytes) {
            log::error!("device write failed: {e}");
            return;
        }
        let _ = writer.flush();
        progress.fetch_add(1, Ordering::Relaxed);
        // Push the keepalive deadline forward — every real write
        // counts as activity.
        if let Some((interval, _)) = keepalive.as_ref() {
            next_keepalive = Some(Instant::now() + *interval);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A line that stays quiet: every read times out, then it closes.
    struct QuietThenEof(u32);
    impl Read for QuietThenEof {
        fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
            if self.0 == 0 {
                return Ok(0);
            }
            self.0 -= 1;
            // Both kinds of read timeout: a serial port's, and a TCP
            // socket's on Unix.
            if self.0.is_multiple_of(2) {
                Err(io::ErrorKind::WouldBlock.into())
            } else {
                Err(io::ErrorKind::TimedOut.into())
            }
        }
    }

    /// Emits one frame per tick.
    struct Ticker;
    impl DeviceDecoder for Ticker {
        fn decode(&mut self, _bytes: &[u8], _events: &mut Vec<DeviceEvent>) {}
        fn tick(&mut self, events: &mut Vec<DeviceEvent>) {
            events.push(DeviceEvent::Frame(RawFrame::new(
                None,
                0,
                1,
                0,
                0,
                std::iter::empty(),
            )));
        }
    }

    struct NoEncoder;
    impl DeviceEncoder for NoEncoder {
        fn encode_frame(&self, _frame: &RawFrame) -> Option<Vec<u8>> {
            None
        }
    }

    /// Collects what the writer thread writes.
    #[derive(Clone, Default)]
    struct Sink(Arc<Mutex<Vec<u8>>>);
    impl Write for Sink {
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    /// A codec under the threaded runner: `open` goes out first, the
    /// handshake reply it asks for reaches the writer, frames come out
    /// with `fixtime` stamped on, sends are encoded by the same codec, and
    /// closing writes its goodbye.
    #[test]
    fn run_codec_drives_a_codec_on_threads() {
        use crate::engine::codec::maretron::{Config, Maretron};
        use crate::engine::format::maretron_ipg::{
            F1_SYNC_BIT, FRAME_SYNC, build_connect, build_frame, build_set_mode_binary,
        };

        let mut input = b"CONNECTED\t1234567\0".to_vec();
        input.extend([
            FRAME_SYNC,
            F1_SYNC_BIT | (6 << 4) | (1 << 1),
            0xF1,
            0x01,
            0x17,
            1,
            0x42,
        ]);
        let sink = Sink::default();
        let handle = run_codec(
            Maretron::new(Config {
                password: "pw".into(),
            }),
            Some("fixed".into()),
            Box::new(io::Cursor::new(input)),
            Box::new(sink.clone()),
        );

        let frame = handle.frames_rx.recv().expect("a frame");
        assert_eq!(frame.pgn, 0xF101);
        assert_eq!(frame.timestamp.as_deref(), Some("fixed"));

        let out = RawFrame::new(None, 3, 127508, 17, 255, [1u8, 2, 3, 4, 5, 6, 7, 8]);
        handle.send_frame(out.clone()).unwrap();
        assert_eq!(handle.close(), Closed::Confirmed);

        let mut expected = build_connect("pw");
        expected.extend(build_set_mode_binary());
        expected.extend(build_frame(out.pgn, out.prio, out.dst, &out.data).unwrap());
        assert_eq!(*sink.0.lock().unwrap(), expected);
    }

    /// A BAM's data packets go out at the codec's deadlines, 50 ms apart,
    /// though the reader only ticks every 250 ms of a quiet line.
    #[test]
    fn the_writer_sends_at_the_codecs_deadlines() {
        use crate::engine::BusProtocol;
        use crate::engine::codec::bst95::Bst95;
        use crate::engine::format::bst::{BdtpDecoder, bst95_message};

        /// A quiet line, for a second.
        struct Quiet(u32);
        impl Read for Quiet {
            fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
                if self.0 == 0 {
                    return Ok(0);
                }
                self.0 -= 1;
                thread::sleep(Duration::from_millis(250));
                Err(io::ErrorKind::WouldBlock.into())
            }
        }
        let sink = Sink::default();
        let handle = run_codec(
            Bst95::new(BusProtocol::J1939),
            None,
            Box::new(Quiet(4)),
            Box::new(sink.clone()),
        );
        let started = Instant::now();
        let bam = RawFrame::new(None, 6, 65260, 0x21, 255, [7u8; 20]);
        handle.send_frame(bam).unwrap();
        let packets = |sink: &Sink| {
            let mut d = BdtpDecoder::default();
            sink.0
                .lock()
                .unwrap()
                .iter()
                .filter_map(|&b| d.push_byte(b))
                .filter_map(|m| bst95_message(&m))
                .count()
        };
        while packets(&sink) < 4 && started.elapsed() < Duration::from_secs(2) {
            thread::sleep(Duration::from_millis(5));
        }
        let took = started.elapsed();
        assert_eq!(packets(&sink), 4, "the announcement and three packets");
        assert!(took >= Duration::from_millis(150), "{took:?}");
        assert!(
            took < Duration::from_millis(600),
            "not paced by the tick: {took:?}"
        );
        handle.close();
    }

    /// A codec that finds the device is the wrong one stops the reader:
    /// the frames end, and the handle says why. The bytes after the one
    /// that told it are not read.
    #[test]
    fn a_codec_failure_ends_the_device() {
        struct Picky(bool);
        impl Codec for Picky {
            fn receive(&mut self, bytes: &[u8], _now_ms: u64, events: &mut Vec<Event>) {
                for &b in bytes {
                    if b == b'!' {
                        self.0 = true;
                    } else if !self.0 {
                        events.push(Event::Frame(RawFrame::new(None, 6, 127508, 1, 255, [b])));
                    }
                }
            }
            fn send(
                &mut self,
                _frame: &RawFrame,
            ) -> Result<Vec<u8>, crate::engine::codec::Refused> {
                Ok(Vec::new())
            }
            fn failure(&self) -> Option<&str> {
                self.0.then_some("not the device")
            }
        }
        let handle = run_codec(
            Picky(false),
            None,
            Box::new(io::Cursor::new(b"a!".to_vec())),
            Box::new(Sink::default()),
        );
        let failure = handle.failure();
        let frames: Vec<RawFrame> = handle.frames_rx.iter().collect();
        assert_eq!(frames.len(), 1);
        assert_eq!(failure.get(), Some("not the device"));
    }

    /// A decoder's deadlines advance on a quiet line: each read timeout
    /// ticks it, and what the tick produces is delivered.
    #[test]
    fn read_timeouts_tick_the_decoder() {
        let handle = run(
            Ticker,
            NoEncoder,
            Box::new(QuietThenEof(3)),
            Box::new(io::sink()),
        );
        let ticks = handle.frames_rx.iter().count();
        assert_eq!(ticks, 3);
    }

    /// A gateway that reboots resets the connection: the frames end and
    /// the handle says why, so `interface` can exit with an error.
    #[test]
    fn a_read_error_loses_the_device() {
        struct Reset;
        impl Read for Reset {
            fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
                Err(io::ErrorKind::ConnectionReset.into())
            }
        }
        let handle = run(Ticker, NoEncoder, Box::new(Reset), Box::new(io::sink()));
        let (failure, lost) = (handle.failure(), handle.lost());
        assert_eq!(handle.frames_rx.iter().count(), 0);
        assert_eq!(failure.get(), None);
        assert_eq!(lost.get(), Some("device read error: connection reset"));
    }

    /// A TCP keepalive that got no answer ends the read with ETIMEDOUT:
    /// the gateway is gone, though the error is TimedOut like a serial
    /// port's read timeout (canboat#1039).
    #[cfg(all(unix, feature = "io"))]
    #[test]
    fn a_connection_timing_out_loses_the_device() {
        struct KeepaliveGaveUp;
        impl Read for KeepaliveGaveUp {
            fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
                Err(io::Error::from_raw_os_error(libc::ETIMEDOUT))
            }
        }
        let handle = run(
            Ticker,
            NoEncoder,
            Box::new(KeepaliveGaveUp),
            Box::new(io::sink()),
        );
        let lost = handle.lost();
        assert_eq!(handle.frames_rx.iter().count(), 0);
        let why = lost.get().expect("the device is lost");
        assert!(why.starts_with("device read error: "), "{why}");
        assert_ne!(why, "the device ended the connection");
    }

    #[test]
    fn only_a_read_timeout_is_read_on_after() {
        assert!(is_read_timeout(&io::ErrorKind::WouldBlock.into()));
        // The serialport crate's: no OS error.
        assert!(is_read_timeout(&io::Error::new(
            io::ErrorKind::TimedOut,
            "Operation timed out"
        )));
        assert!(!is_read_timeout(&io::ErrorKind::ConnectionReset.into()));
        #[cfg(all(unix, feature = "io"))]
        assert!(!is_read_timeout(&io::Error::from_raw_os_error(
            libc::ETIMEDOUT
        )));
    }

    /// A device that ends its stream is gone too.
    #[test]
    fn the_end_of_the_stream_loses_the_device() {
        let handle = run(
            Ticker,
            NoEncoder,
            Box::new(QuietThenEof(0)),
            Box::new(io::sink()),
        );
        let lost = handle.lost();
        assert_eq!(handle.frames_rx.iter().count(), 0);
        assert_eq!(lost.get(), Some("the device ended the connection"));
    }
}