canboat 9.1.0

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

//! `RawFrame` streams: a sync, pull-based [`FrameReader`] /
//! [`FrameWriter`] pair over the canonical [`RawFrame`].
//!
//! Every canboat-rs tool that isn't a pure byte-shunt has the same
//! spine: pull [`RawFrame`]s from some source, optionally transform
//! them, push them to some sink. Historically each binary re-wired
//! that spine by hand — a line loop here, an `mpsc::Receiver` drain
//! there. These two traits name the spine so the `convert` and
//! `interface` subcommands can wire an arbitrary source to an
//! arbitrary sink without caring which concrete pair they hold.
//!
//! The traits are deliberately blocking: nothing in the workspace is
//! async (the device bridges get their concurrency from
//! `std::thread` + `std::sync::mpsc`), so a `read_frame` that blocks
//! until the next frame — or `Ok(None)` at end-of-stream — matches
//! every existing source, including a live CAN interface drained
//! through [`crate::io::device::DeviceHandle`].

use std::collections::VecDeque;
use std::io::{self, BufRead, Read, Write};

use crate::engine::BusProtocol;
use crate::engine::RawFrame;
use crate::engine::format::actisense_ascii::write_line as write_actisense;
use crate::engine::format::bst::{BdtpDecoder, MessageAssembler};
use crate::engine::format::bst_d0::BstD0Decoder;
use crate::engine::format::ebl::{
    EBL_FORMAT_VERSION, EblDecoder, EblEvent, encode_frame as encode_ebl, encode_preamble,
    frame_unix_ms,
};
use crate::engine::format::plain::write_line as write_plain;
use crate::engine::format::ydwg02::write_line as write_ydwg02;
use crate::engine::format::{
    InputFormat, detect, header_implies_coalesced, is_undecided, parse_checked, parse_format_header,
};

/// Pull-based source of [`RawFrame`]s — the engine
/// [`FrameSource`](crate::engine::FrameSource) trait, re-exported here
/// under its historical name. Concrete readers in this crate
/// ([`LineFrameReader`], [`EblReader`]) implement it; an
/// [`mpsc::Receiver<RawFrame>`](std::sync::mpsc::Receiver) does too (the
/// impl lives in the engine).
pub use crate::engine::FrameSource as FrameReader;

/// Sink for [`RawFrame`]s.
pub trait FrameWriter {
    fn write_frame(&mut self, frame: &RawFrame) -> io::Result<()>;

    /// Flush any buffered output. The default is a no-op for sinks
    /// that don't buffer (e.g. a device channel).
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

/// Drain every frame from `reader` into `writer`, then flush.
/// Returns the number of frames copied. This is the whole body of a
/// format→format `convert` once the endpoints are chosen.
pub fn copy<R: FrameReader + ?Sized, W: FrameWriter + ?Sized>(
    reader: &mut R,
    writer: &mut W,
) -> io::Result<u64> {
    let mut n = 0;
    while let Some(frame) = reader.read_frame()? {
        writer.write_frame(&frame)?;
        n += 1;
    }
    writer.flush()?;
    Ok(n)
}

/// A [`FrameReader`] over any canboat ASCII line format.
///
/// Owns the format-detection state machine that used to live inline
/// in the analyzer's replay loop: it honours `# format=<NAME>`
/// headers, auto-detects the format from the first content line when
/// none is forced, and parses each subsequent line via
/// [`parse_for`](crate::engine::format::parse_for). Blank lines, comment/header lines and
/// non-frame control sentences (iKonvert `$PDGY` status, Garmin CSV
/// headers) are swallowed internally so callers only ever see frames.
///
/// The wrapped source is any [`BufRead`], which includes a file, a
/// stdin lock, or the PLAIN [`BufRead`](crate::io::container::plain_reader)
/// that unwraps a `.nif` / `.pcap` capture container.
pub struct LineFrameReader<R: BufRead> {
    lines: crate::io::LineReader<R>,
    /// Forced format (`--format`) or, once detected, the active one.
    active: Option<InputFormat>,
    /// True once a `# format=<NAME>` header declared an
    /// already-coalesced format (FAST/ACTISENSE/…). Consumers that
    /// run reassembly downstream consult this to decide whether to
    /// bypass the reassembler.
    header_coalesced: bool,
    /// The bus the lines come from: which raw CAN frames count.
    protocol: crate::engine::BusProtocol,
    /// NMEA 0183 lines skipped while the format is still open.
    undecided: usize,
    /// Whether a `$PCDIN` or `$MXPGN` sentence's checksum is checked.
    checksum: bool,
    /// A MiniPlex's Address Claims and Product Information, which come
    /// whole; and the frames of one that is a fast-packet, still to read.
    psmd: crate::engine::format::mxpgn::Psmd,
    pending: std::collections::VecDeque<RawFrame>,
    psmd_seq: u8,
}

/// How many leading NMEA 0183 lines that hold no frame leave the format
/// open before it falls back to PLAIN.
const MAX_UNDECIDED: usize = 1000;

impl<R: BufRead> LineFrameReader<R> {
    /// Wrap `reader`, auto-detecting the format from its first content
    /// line.
    pub fn new(reader: R) -> Self {
        Self {
            lines: crate::io::LineReader::new(reader),
            active: None,
            header_coalesced: false,
            protocol: crate::engine::BusProtocol::Nmea2000,
            undecided: 0,
            checksum: true,
            psmd: Default::default(),
            pending: Default::default(),
            psmd_seq: 0,
        }
    }

    /// Wrap `reader`, forcing `format` instead of auto-detecting
    /// (analyzer's `--format`).
    pub fn with_format(reader: R, format: InputFormat) -> Self {
        Self {
            lines: crate::io::LineReader::new(reader),
            active: Some(format),
            header_coalesced: false,
            protocol: crate::engine::BusProtocol::Nmea2000,
            undecided: 0,
            checksum: true,
            psmd: Default::default(),
            pending: Default::default(),
            psmd_seq: 0,
        }
    }

    /// Read the lines as coming from a bus of `protocol` (NMEA 2000 by
    /// default). It decides which raw CAN frames a candump line may hold:
    /// 11-bit ones on Quick, 29-bit ISO 11783 ones otherwise.
    pub fn protocol(mut self, protocol: crate::engine::BusProtocol) -> Self {
        self.protocol = protocol;
        self
    }

    /// Accept `$PCDIN` and `$MXPGN` sentences whose checksum is wrong.
    pub fn ignore_checksum(mut self, ignore: bool) -> Self {
        self.checksum = !ignore;
        self
    }

    /// The format currently in effect: the forced one, or whatever
    /// was detected once the first frame has been read.
    pub fn active_format(&self) -> Option<InputFormat> {
        self.active
    }

    /// True when a `# format=<NAME>` header has declared an
    /// already-coalesced format, so a downstream reassembler should
    /// be bypassed. Mirrors canboat's `header_implies_coalesced`.
    pub fn header_coalesced(&self) -> bool {
        self.header_coalesced
    }
}

impl<R: BufRead> FrameReader for LineFrameReader<R> {
    fn read_frame(&mut self) -> io::Result<Option<RawFrame>> {
        loop {
            if let Some(frame) = self.pending.pop_front() {
                return Ok(Some(frame));
            }
            let Some(line) = self.lines.next_line()? else {
                return Ok(None);
            };
            if line.is_empty() {
                continue;
            }
            // `# format=<NAME>` headers from the canboat reader
            // binaries: set the format if not already known and note
            // whether it implies coalesced frames.
            if line.starts_with('#') {
                if let Some(fmt) = parse_format_header(line) {
                    if self.active.is_none() {
                        self.active = Some(fmt);
                        log::info!("input format set by header: {fmt:?}");
                    }
                    if header_implies_coalesced(line) {
                        self.header_coalesced = true;
                        log::debug!("header declares coalesced format; skipping reassembly");
                    }
                }
                continue;
            }
            // Auto-detect on the first content line if not forced.
            // Canboat falls back to PLAIN when nothing else matches; an NMEA
            // 0183 sentence that holds no frame leaves it open, for a
            // while: a MiniPlex stream may start with `$GPRMC` before its
            // first `$MXPGN`.
            if self.active.is_none() {
                if is_undecided(line) && self.undecided < MAX_UNDECIDED {
                    self.undecided += 1;
                    continue;
                }
                self.active = detect(line).or(Some(InputFormat::Plain));
                log::debug!("input format: {:?}", self.active);
            }
            let format = self.active.expect("active format set above");
            match parse_checked(format, self.protocol, line, self.checksum) {
                Ok(Some(frame)) => return Ok(Some(frame)),
                // A MiniPlex's Address Claim or Product Information, among
                // its CAN frames: the latter goes in its fast-packet frames
                // too.
                Ok(None) if format == InputFormat::Mxpgn => {
                    match self.psmd.push(line, self.checksum) {
                        Ok(Some(message)) if message.data.len() > 8 => {
                            let seq = self.psmd_seq;
                            self.psmd_seq = (seq + 1) & 7;
                            for chunk in crate::engine::fastpacket::fragment(seq, &message.data)
                                .unwrap_or_default()
                            {
                                self.pending.push_back(RawFrame::new(
                                    None,
                                    message.prio,
                                    message.pgn,
                                    message.src,
                                    message.dst,
                                    chunk,
                                ));
                            }
                        }
                        Ok(Some(message)) => return Ok(Some(message)),
                        Ok(None) => {}
                        Err(e) => log::warn!("skipping malformed input line: {e}"),
                    }
                    continue;
                }
                // iKonvert control sentences, Garmin CSV headers, etc.
                Ok(None) => continue,
                Err(crate::engine::format::PlainError::Empty) => continue,
                Err(e) => {
                    log::warn!("skipping malformed input line: {e}");
                    continue;
                }
            }
        }
    }
}

/// A [`FrameWriter`] that serialises each frame as a canboat
/// PLAIN/FAST line to an underlying [`Write`].
pub struct PlainWriter<W: Write> {
    inner: W,
    buf: String,
}

impl<W: Write> PlainWriter<W> {
    pub fn new(inner: W) -> Self {
        Self {
            inner,
            buf: String::with_capacity(256),
        }
    }
}

impl<W: Write> FrameWriter for PlainWriter<W> {
    fn write_frame(&mut self, frame: &RawFrame) -> io::Result<()> {
        self.buf.clear();
        // Writing to a `String` is infallible; the only error path is
        // the real I/O below.
        write_plain(&mut self.buf, frame).expect("PLAIN write to String");
        self.buf.push('\n');
        self.inner.write_all(self.buf.as_bytes())
    }

    fn flush(&mut self) -> io::Result<()> {
        self.inner.flush()
    }
}

/// A [`FrameWriter`] of ShipModul `$MXPGN` sentences, one CAN frame each,
/// as a MiniPlex sends what it receives: a message longer than a frame
/// goes out in its fast-packet frames, one sentence each. [`Self::actisense`]
/// writes the data in bus order, as an Actisense gateway does.
pub struct MxpgnWriter<W: Write> {
    inner: W,
    protocol: BusProtocol,
    order: crate::engine::format::mxpgn::DataOrder,
    /// Per-(pgn, src) fast-packet sequence counters, mod 8.
    seq: std::collections::HashMap<(u32, u8), u8>,
    buf: String,
}

impl<W: Write> MxpgnWriter<W> {
    /// Sentences for a bus carrying `protocol`, which decides which
    /// messages are fast-packets.
    pub fn new(inner: W, protocol: BusProtocol) -> Self {
        Self {
            inner,
            protocol,
            order: Default::default(),
            seq: Default::default(),
            buf: String::with_capacity(64),
        }
    }

    /// [`Self::new`], the data in bus order, as an Actisense gateway
    /// sends it.
    pub fn actisense(inner: W, protocol: BusProtocol) -> Self {
        Self {
            order: crate::engine::format::mxpgn::DataOrder::Bus,
            ..Self::new(inner, protocol)
        }
    }

    fn write_one(&mut self, frame: &RawFrame, data: &[u8]) -> io::Result<()> {
        let mut one = RawFrame::new(
            None,
            frame.prio,
            frame.pgn,
            frame.src,
            frame.dst,
            data.iter().copied(),
        );
        one.timestamp.clone_from(&frame.timestamp);
        self.buf.clear();
        crate::engine::format::mxpgn::write_ordered(&mut self.buf, &one, false, self.order)
            .expect("$MXPGN write to String");
        self.buf.push_str("\r\n");
        self.inner.write_all(self.buf.as_bytes())
    }
}

impl<W: Write> FrameWriter for MxpgnWriter<W> {
    fn write_frame(&mut self, frame: &RawFrame) -> io::Result<()> {
        use crate::engine::{FramePacketType, fastpacket};
        // canboat's own records never went over the bus, and Quick's
        // 11-bit frames have no PGN.
        if frame.pgn >= fastpacket::CANBOAT_PGN_START || self.protocol.standard_frames() {
            return Ok(());
        }
        if self.protocol.packet_type(frame.pgn) == FramePacketType::Fast {
            let e = self.seq.entry((frame.pgn, frame.src)).or_insert(0);
            let seq = *e;
            *e = (seq + 1) & 7;
            let Some(chunks) = fastpacket::fragment(seq, &frame.data) else {
                log::warn!(
                    "not writing PGN {}: {} bytes exceed a fast-packet",
                    frame.pgn,
                    frame.data.len()
                );
                return Ok(());
            };
            for chunk in chunks {
                self.write_one(frame, &chunk)?;
            }
        } else if frame.data.len() <= 8 {
            self.write_one(frame, &frame.data.clone())?;
        } else {
            log::warn!(
                "not writing PGN {}: {} bytes do not fit one frame",
                frame.pgn,
                frame.data.len()
            );
        }
        Ok(())
    }

    fn flush(&mut self) -> io::Result<()> {
        self.inner.flush()
    }
}

/// A [`FrameWriter`] emitting one line per frame via a core text
/// encoder (`fn(&mut String, &RawFrame)`), used for the YDWG02,
/// Actisense-ASCII and SeaSmart output formats. Mirrors [`PlainWriter`].
pub struct TextLineWriter<W: Write> {
    inner: W,
    buf: String,
    encode: fn(&mut String, &RawFrame) -> std::fmt::Result,
}

impl<W: Write> TextLineWriter<W> {
    /// YDWG02 / YDEN received lines (`<time> R <canid> <hex…>`).
    pub fn ydwg02(inner: W) -> Self {
        Self::new(inner, write_ydwg02)
    }

    /// Actisense N2K-ASCII lines (`A<time> <sdp> <pgn> <hex>`).
    pub fn actisense(inner: W) -> Self {
        Self::new(inner, write_actisense)
    }

    /// SeaSmart `$PCDIN` sentences, one whole message each, ending in
    /// CR LF as NMEA 0183 does. A message longer than a fast-packet, which
    /// a reader would cut short, is skipped with a warning.
    pub fn seasmart(inner: W) -> Self {
        Self::new(inner, |w, frame| {
            if frame.data.len() > crate::engine::frame::FASTPACKET_MAX_SIZE {
                log::warn!(
                    "not writing PGN {}: {} bytes exceed a fast-packet",
                    frame.pgn,
                    frame.data.len()
                );
                return Ok(());
            }
            crate::engine::format::seasmart::write_line(w, frame)?;
            w.push('\r');
            Ok(())
        })
    }

    fn new(inner: W, encode: fn(&mut String, &RawFrame) -> std::fmt::Result) -> Self {
        Self {
            inner,
            buf: String::with_capacity(256),
            encode,
        }
    }
}

impl<W: Write> FrameWriter for TextLineWriter<W> {
    fn write_frame(&mut self, frame: &RawFrame) -> io::Result<()> {
        self.buf.clear();
        (self.encode)(&mut self.buf, frame).expect("text write to String");
        if self.buf.is_empty() {
            // Skipped by the encoder.
            return Ok(());
        }
        self.buf.push('\n');
        self.inner.write_all(self.buf.as_bytes())
    }

    fn flush(&mut self) -> io::Result<()> {
        self.inner.flush()
    }
}

/// A [`FrameWriter`] emitting each frame as an Actisense `.ebl` binary
/// record pair (`TimeUTC` metatag + `N2K_MSG_RECEIVED`), after the
/// `TimeUTC` and `Version` metatags an `.ebl` file starts with. No line
/// terminator — the framing is self-delimiting.
pub struct EblWriter<W: Write> {
    inner: W,
    buf: Vec<u8>,
    started: bool,
}

impl<W: Write> EblWriter<W> {
    pub fn new(inner: W) -> Self {
        Self {
            inner,
            buf: Vec::with_capacity(256),
            started: false,
        }
    }
}

impl<W: Write> FrameWriter for EblWriter<W> {
    fn write_frame(&mut self, frame: &RawFrame) -> io::Result<()> {
        self.buf.clear();
        if !self.started {
            self.started = true;
            encode_preamble(frame_unix_ms(frame), &mut self.buf);
        }
        encode_ebl(frame, &mut self.buf);
        self.inner.write_all(&self.buf)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.inner.flush()
    }
}

/// A [`FrameReader`] over an Actisense `.ebl` byte stream — the inverse
/// of [`EblWriter`], and a reader for the logs of other Actisense gateways
/// and tools. It hands back the frame of every BST-93 (NGT-1), BST-95 (raw
/// CAN frame) and BST-D0 (W2K-1) message, whether in the stream or in a
/// `BSTRawFrame` metatag, dated by the `TimeUTC` metatag before it. A
/// BST-95 message is one CAN frame; its fast-packets are joined into whole
/// messages.
///
/// Binary, not line-based, so it implements [`FrameReader`] directly
/// rather than going through [`LineFrameReader`].
pub struct EblReader<R: Read> {
    inner: R,
    decoder: EblDecoder,
    messages: MessageAssembler,
    events: Vec<EblEvent>,
    queue: VecDeque<RawFrame>,
    /// The latest `TimeUTC`, formatted, for the frames after it.
    time: Option<String>,
    buf: Box<[u8; 8192]>,
    eof: bool,
}

impl<R: Read> EblReader<R> {
    /// A reader of an NMEA 2000 log.
    pub fn new(inner: R) -> Self {
        Self::with_protocol(inner, BusProtocol::Nmea2000)
    }

    /// A reader of a log of a bus carrying `protocol`, which decides which
    /// BST-95 frames are parts of a longer message.
    pub fn with_protocol(inner: R, protocol: BusProtocol) -> Self {
        Self {
            inner,
            decoder: EblDecoder::new(),
            messages: MessageAssembler::new(protocol),
            events: Vec::new(),
            queue: VecDeque::new(),
            time: None,
            buf: Box::new([0u8; 8192]),
            eof: false,
        }
    }

    fn take_events(&mut self) {
        for ev in self.events.drain(..) {
            match ev {
                EblEvent::Time(ms) => self.time = Some(crate::engine::format_iso_ms(ms)),
                EblEvent::Message { bytes, raw } => {
                    if let Some(mut frame) = self.messages.push(&bytes, raw) {
                        frame.timestamp = self.time.clone();
                        self.queue.push_back(frame);
                    }
                }
                EblEvent::Version(v) if v > EBL_FORMAT_VERSION => log::warn!(
                    "ebl: the file is EBL v{}.{:03}, newer than the v1.002 canboat knows; reading it anyway",
                    v / 1000,
                    v % 1000
                ),
                EblEvent::Warning(why) => log::warn!("ebl: {why}"),
                EblEvent::Version(_)
                | EblEvent::Direction(_)
                | EblEvent::Description(_)
                | EblEvent::Unknown { .. } => {}
            }
        }
    }
}

impl<R: Read> FrameReader for EblReader<R> {
    fn read_frame(&mut self) -> io::Result<Option<RawFrame>> {
        loop {
            if let Some(frame) = self.queue.pop_front() {
                return Ok(Some(frame));
            }
            if self.eof {
                return Ok(None);
            }
            let n = self.inner.read(&mut self.buf[..])?;
            if n == 0 {
                self.eof = true;
                self.decoder.finish(&mut self.events);
            } else {
                self.decoder.push_bytes(&self.buf[..n], &mut self.events);
            }
            self.take_events();
        }
    }
}

/// A [`FrameReader`] over a stream of Actisense BST-D0 messages: what a
/// W2K-1 or PRO-NDC-1E2K sends over TCP in its Actisense data mode
/// (#993). The messages carry only the device's own clock, so the frames
/// have no timestamp.
///
/// Binary, not line-based, so it implements [`FrameReader`] directly
/// rather than going through [`LineFrameReader`].
pub struct BstD0Reader<R: Read> {
    inner: R,
    decoder: BstD0Decoder,
    queue: VecDeque<RawFrame>,
    buf: Box<[u8; 8192]>,
    eof: bool,
}

impl<R: Read> BstD0Reader<R> {
    pub fn new(inner: R) -> Self {
        Self {
            inner,
            decoder: BstD0Decoder::new(),
            queue: VecDeque::new(),
            buf: Box::new([0u8; 8192]),
            eof: false,
        }
    }
}

impl<R: Read> FrameReader for BstD0Reader<R> {
    fn read_frame(&mut self) -> io::Result<Option<RawFrame>> {
        loop {
            if let Some(frame) = self.queue.pop_front() {
                return Ok(Some(frame));
            }
            if self.eof {
                return Ok(None);
            }
            let n = self.inner.read(&mut self.buf[..])?;
            if n == 0 {
                self.eof = true;
                continue;
            }
            self.queue.extend(self.decoder.push_bytes(&self.buf[..n]));
        }
    }
}

/// A [`FrameReader`] over a stream of Actisense BST messages in BDTP
/// framing: a capture of a PRO-NDC-1E2K or W2K-1 in its "CAN Actisense"
/// mode, which sends BST-95 raw CAN frames. It hands back the frame of every
/// BST-93, BST-94, BST-95 and BST-D0 message, and skips other messages and
/// damaged frames. A BST-95 message is one CAN frame; its fast-packets are
/// joined into whole messages. The messages carry only the device's own
/// clock, so the frames have no timestamp.
///
/// Binary, not line-based, so it implements [`FrameReader`] directly
/// rather than going through [`LineFrameReader`].
pub struct BstReader<R: Read> {
    inner: R,
    decoder: BdtpDecoder,
    messages: MessageAssembler,
    queue: VecDeque<RawFrame>,
    buf: Box<[u8; 8192]>,
    eof: bool,
}

impl<R: Read> BstReader<R> {
    /// A reader of an NMEA 2000 capture.
    pub fn new(inner: R) -> Self {
        Self::with_protocol(inner, BusProtocol::Nmea2000)
    }

    /// A reader of a capture of a bus carrying `protocol`, which decides
    /// which BST-95 frames are parts of a longer message.
    pub fn with_protocol(inner: R, protocol: BusProtocol) -> Self {
        Self {
            inner,
            decoder: BdtpDecoder::default(),
            messages: MessageAssembler::new(protocol),
            queue: VecDeque::new(),
            buf: Box::new([0u8; 8192]),
            eof: false,
        }
    }
}

impl<R: Read> FrameReader for BstReader<R> {
    fn read_frame(&mut self) -> io::Result<Option<RawFrame>> {
        loop {
            if let Some(frame) = self.queue.pop_front() {
                return Ok(Some(frame));
            }
            if self.eof {
                return Ok(None);
            }
            let n = self.inner.read(&mut self.buf[..])?;
            if n == 0 {
                self.eof = true;
                continue;
            }
            for &b in &self.buf[..n] {
                if let Some(m) = self.decoder.push_byte(b) {
                    self.queue.extend(self.messages.push(&m, false));
                }
            }
        }
    }
}

/// [`FrameReader`] over the bytes a Waveshare USB-CAN-A sends on its
/// serial port (`cat /dev/ttyUSB0 > capture.bin` once canboat has set it up):
/// 20-byte frames, with their fast-packets joined. A lost byte costs the
/// frames it falls in, not the rest. The frames carry no time, so neither
/// do these.
pub struct UsbCanAReader<R: Read> {
    inner: R,
    frames: crate::engine::codec::usb_can_a::Frames,
    protocol: BusProtocol,
    reassembler: crate::engine::Reassembler,
    queue: VecDeque<RawFrame>,
    buf: Box<[u8; 8192]>,
    eof: bool,
}

impl<R: Read> UsbCanAReader<R> {
    /// A reader of an NMEA 2000 capture.
    pub fn new(inner: R) -> Self {
        Self::with_protocol(inner, BusProtocol::Nmea2000)
    }

    /// A reader of a capture of a bus carrying `protocol`, which decides
    /// which frames are parts of a longer message, and whether the 11-bit
    /// or the 29-bit frames are the bus's.
    pub fn with_protocol(inner: R, protocol: BusProtocol) -> Self {
        Self {
            inner,
            frames: Default::default(),
            protocol,
            reassembler: crate::engine::Reassembler::new(),
            queue: VecDeque::new(),
            buf: Box::new([0u8; 8192]),
            eof: false,
        }
    }
}

impl<R: Read> FrameReader for UsbCanAReader<R> {
    fn read_frame(&mut self) -> io::Result<Option<RawFrame>> {
        use crate::engine::codec::{Event, raw_can};
        loop {
            if let Some(frame) = self.queue.pop_front() {
                return Ok(Some(frame));
            }
            if self.eof {
                return Ok(None);
            }
            let n = self.inner.read(&mut self.buf[..])?;
            if n == 0 {
                self.eof = true;
                continue;
            }
            let (protocol, reassembler) = (self.protocol, &mut self.reassembler);
            let mut events = Vec::new();
            self.frames.push(&self.buf[..n], |f| {
                raw_can::receive(
                    protocol,
                    reassembler,
                    f.id,
                    f.extended,
                    &f.data,
                    None,
                    &mut events,
                )
            });
            for event in events {
                match event {
                    Event::Frame(f) => self.queue.push_back(f),
                    Event::Error(e) => log::warn!("usb-can-a: {e}"),
                    Event::Send(_) => {}
                }
            }
        }
    }
}

/// The receiving end of a device's frame channel is a blocking
/// A device's [`FrameSender`](crate::io::device::FrameSender) is a
/// [`FrameWriter`]: each frame is queued to the device's writer
/// thread, which encodes it for the wire. A gone writer thread
/// (device closed) surfaces as a broken-pipe I/O error.
impl FrameWriter for crate::io::device::FrameSender {
    fn write_frame(&mut self, frame: &RawFrame) -> io::Result<()> {
        self.send_frame(frame.clone())
            .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, e))
    }
}

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

    #[test]
    fn line_reader_yields_frames_and_skips_noise() {
        // A comment/header, a blank line, then two PLAIN frames.
        let input = "\
# format=PLAIN

2022-09-10T12:10:16.614Z,6,60928,5,255,8,fb,9b,70,22,00,9b,50,c0
2022-09-10T12:10:16.700Z,6,60928,6,255,8,01,02,03,04,05,06,07,08
";
        let mut r = LineFrameReader::new(Cursor::new(input));
        let f1 = r.read_frame().unwrap().expect("first frame");
        assert_eq!(f1.pgn, 60928);
        assert_eq!(f1.src, 5);
        let f2 = r.read_frame().unwrap().expect("second frame");
        assert_eq!(f2.src, 6);
        assert!(r.read_frame().unwrap().is_none());
        assert_eq!(r.active_format(), Some(InputFormat::Plain));
    }

    #[test]
    fn plain_round_trips_through_copy() {
        let input = "2022-09-10T12:10:16.614Z,6,60928,5,255,8,fb,9b,70,22,00,9b,50,c0\n";
        let mut reader = LineFrameReader::new(Cursor::new(input));
        let mut out: Vec<u8> = Vec::new();
        let n = copy(&mut reader, &mut PlainWriter::new(&mut out)).unwrap();
        assert_eq!(n, 1);
        // Re-parsing the emitted line yields an identical frame.
        let mut back = LineFrameReader::new(Cursor::new(out));
        let f = back.read_frame().unwrap().unwrap();
        assert_eq!(f.pgn, 60928);
        assert_eq!(f.src, 5);
        assert_eq!(f.dst, 255);
        assert_eq!(
            &f.data[..],
            &[0xfb, 0x9b, 0x70, 0x22, 0x00, 0x9b, 0x50, 0xc0]
        );
    }

    #[test]
    fn header_coalesced_is_reported() {
        let input = "# format=FAST\n";
        let mut r = LineFrameReader::new(Cursor::new(input));
        assert!(r.read_frame().unwrap().is_none());
        assert!(r.header_coalesced());
    }

    /// SeaSmart: one sentence per message; one longer than a fast-packet,
    /// which a reader would cut short, is skipped.
    #[test]
    fn seasmart_writes_whole_messages_and_skips_oversized_ones() {
        let mut out = Vec::new();
        {
            let mut w = TextLineWriter::seasmart(&mut out);
            let long = RawFrame::new(None, 6, 0xef00, 1, 2, std::iter::repeat_n(0u8, 300));
            w.write_frame(&long).unwrap();
            let attitude = RawFrame::new(
                None,
                3,
                127257,
                15,
                255,
                [0x2a, 0xaf, 0x00, 0xd1, 0x06, 0x74, 0x14, 0xff],
            );
            w.write_frame(&attitude).unwrap();
        }
        assert_eq!(
            String::from_utf8(out).unwrap(),
            "$PCDIN,01F119,00000000,0F,2AAF00D1067414FF*59\r\n"
        );
    }
}