canboat 9.1.0

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

//! Line-based network gateway codecs: Yacht Devices RAW (YDWG-02 /
//! YDEN, also the NavLink2's RAW port), Actisense N2K ASCII (W2K-1),
//! ShipModul `$MXPGN` (MiniPlex), SeaSmart `$PCDIN` and SocketCAN
//! `candump` (a SensESP N2K gateway).
//!
//! They speak newline-delimited text over a TCP or UDP socket, or a
//! serial port. The decoder half accumulates a partial line
//! across reads, parses each complete line with the matching
//! [`crate::engine::format`] parser and — for RAW, whose lines are
//! individual CAN frames — reassembles fast-packets before emitting.
//! The encoder half writes the device's transmit shape: complete
//! `A…` lines for N2K ASCII, bare `<CANID> <bytes…>` lines for RAW
//! with fast-packet payloads fragmented per ISO 11783-3. On RAW and
//! candump a longer message that is not a fast-packet goes by ISO TP
//! ([`crate::engine::iso_tp`]), paced by the writer at the encoder's
//! deadlines; the decoder hands it the receiver's Clear To Send.

use std::collections::{HashMap, VecDeque};
use std::fmt::Write as _;
use std::io::{Read, Write};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use crate::engine::format::{actisense_ascii, candump, iso11783_compose, mxpgn, seasmart, ydwg02};
use crate::engine::frame::FASTPACKET_MAX_SIZE;
use crate::engine::iso_tp::{PGN_TP_CM, TP_MAX_SIZE, TpSender};
use crate::engine::reassembly::{Reassembled, Reassembler};
use crate::engine::startup::format_iso_ms;
use crate::engine::{BusProtocol, FramePacketType, RawFrame};

use super::{DeviceDecoder, DeviceEncoder, DeviceEvent, DeviceHandle};
use crate::engine::fastpacket;

/// Which line format the socket speaks.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LineFormat {
    /// Yacht Devices RAW: `HH:MM:SS.mmm R <CANID> <bytes…>` per CAN
    /// frame received; transmit lines are `<CANID> <bytes…>`.
    YdwgRaw,
    /// Actisense N2K ASCII: `A…` lines carrying complete (coalesced)
    /// messages in both directions.
    N2kAscii,
    /// ShipModul MiniPlex `$MXPGN` sentences, one CAN frame each, among
    /// other NMEA 0183 sentences, which carry no frame. Sent in the send
    /// form: the MiniPlex sends from its own address.
    Mxpgn,
    /// `$MXPGN` as an Actisense gateway sends and takes it: the data in bus
    /// order (see [`mxpgn::DataOrder`]); otherwise as [`Self::Mxpgn`].
    ActisenseMxpgn,
    /// SeaSmart `$PCDIN` sentences, one whole message each, among other
    /// NMEA 0183 sentences, which carry no message. Sent the same way,
    /// with no priority or destination: the sentence has neither.
    SeaSmart,
    /// SocketCAN `candump` text, one CAN frame per line, in any of its
    /// shapes; sent in the log shape. 11-bit frames on a Quick bus.
    Candump,
}

impl LineFormat {
    /// How its `$MXPGN` orders the data.
    fn mxpgn_order(self) -> mxpgn::DataOrder {
        if self == LineFormat::ActisenseMxpgn {
            mxpgn::DataOrder::Bus
        } else {
            mxpgn::DataOrder::Reversed
        }
    }
}

/// The CAN frames written to a RAW gateway, to know their `T` echoes
/// from another client's transmits: the gateway echoes every transmit on
/// every connection, with its own source address in place of ours.
#[derive(Clone, Default)]
pub struct Sent(Arc<Mutex<VecDeque<SentFrame>>>);

/// A CAN id without its source address, the data, and when it was sent.
type SentFrame = (u32, Vec<u8>, Instant);

/// How long an echo of a frame written is waited for.
const ECHO_WAIT: Duration = Duration::from_secs(2);

impl Sent {
    /// The CAN id without its source address.
    fn key(canid: u32) -> u32 {
        canid & !0xff
    }

    fn note(&self, canid: u32, data: &[u8]) {
        let mut sent = self.0.lock().expect("sent mutex");
        let now = Instant::now();
        while sent.front().is_some_and(|(_, _, at)| now - *at > ECHO_WAIT) {
            sent.pop_front();
        }
        sent.push_back((Self::key(canid), data.to_vec(), now));
    }

    /// Whether `canid` and `data` echo a frame written, which is then
    /// forgotten.
    fn is_echo(&self, canid: u32, data: &[u8]) -> bool {
        let mut sent = self.0.lock().expect("sent mutex");
        let now = Instant::now();
        let key = Self::key(canid);
        match sent
            .iter()
            .position(|(k, d, at)| *k == key && d == data && now - *at <= ECHO_WAIT)
        {
            Some(i) => {
                sent.remove(i);
                true
            }
            None => false,
        }
    }
}

pub struct Decoder {
    format: LineFormat,
    /// Decides which RAW frames are fast-packet fragments.
    protocol: BusProtocol,
    acc: String,
    reassembler: Reassembler,
    sent: Sent,
    /// Whether a `$PCDIN` or `$MXPGN` sentence's checksum is checked.
    checksum: bool,
    /// A MiniPlex's Address Claims and Product Information.
    psmd: mxpgn::Psmd,
    /// The encoder, to hand the ISO TP answers the bus delivers.
    encoder: Option<Arc<Encoder>>,
}

impl Decoder {
    #[cfg(test)]
    pub fn new(format: LineFormat) -> Self {
        Self::with_protocol(format, BusProtocol::Nmea2000)
    }

    pub fn with_protocol(format: LineFormat, protocol: BusProtocol) -> Self {
        Self {
            format,
            protocol,
            acc: String::with_capacity(256),
            reassembler: Reassembler::new(),
            sent: Sent::default(),
            checksum: true,
            psmd: mxpgn::Psmd::default(),
            encoder: None,
        }
    }

    /// Hand `encoder` the Clear To Send, End Of Message and Abort frames
    /// for the ISO TP transfers it sends, and the gateway's address.
    pub fn with_encoder(mut self, encoder: Arc<Encoder>) -> Self {
        self.encoder = Some(encoder);
        self
    }

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

    /// Know the echoes of what `sent` records.
    pub fn with_sent(mut self, sent: Sent) -> Self {
        self.sent = sent;
        self
    }

    /// Pass on a CAN frame received at `now`, joining fast-packets.
    fn push_frame(&mut self, mut frame: RawFrame, now: &str, events: &mut Vec<DeviceEvent>) {
        if frame.pgn == PGN_TP_CM
            && let Some(encoder) = &self.encoder
        {
            let answer = encoder.tp_frame(super::now_ms(), &frame);
            if !answer.is_empty() {
                events.push(DeviceEvent::SendBytes(answer));
            }
        }
        frame.timestamp = Some(now.to_string());
        let pt = self.protocol.packet_type(frame.pgn);
        match self.reassembler.push(frame, pt) {
            Reassembled::PassThrough(f) | Reassembled::Complete(f) => {
                events.push(DeviceEvent::Frame(f))
            }
            Reassembled::Partial => {}
            Reassembled::Error(e) => events.push(DeviceEvent::Error(e.to_string())),
        }
    }
}

impl DeviceDecoder for Decoder {
    /// A quiet line still gets its minute's summary of incomplete
    /// fast-packets.
    fn tick(&mut self, events: &mut Vec<DeviceEvent>) {
        self.reassembler.tick(super::now_ms());
        self.poll_tp(events);
    }

    /// Frames get the host's time, as from every other gateway: the
    /// gateways' own is a time of day without a date, and a YDEN's runs
    /// behind the host's, and now and then backwards (canboat#1039).
    fn decode(&mut self, bytes: &[u8], events: &mut Vec<DeviceEvent>) {
        let now_ms = super::now_ms();
        self.reassembler.tick(now_ms);
        let now = format_iso_ms(now_ms);
        self.acc.push_str(&String::from_utf8_lossy(bytes));
        while let Some(pos) = self.acc.find('\n') {
            let line: String = self.acc[..pos].trim().to_string();
            self.acc.drain(..=pos);
            if line.is_empty() {
                continue;
            }
            match self.format {
                LineFormat::YdwgRaw => {
                    // A transmit, marked `T`, is echoed on every
                    // connection: ours is skipped, as no gateway reports
                    // what canboat sent, but another client's is bus
                    // traffic like any other.
                    let transmit = line.split_whitespace().nth(1) == Some("T");
                    match ydwg02::parse_line(&line) {
                        Ok(frame)
                            if transmit
                                && self.sent.is_echo(
                                    iso11783_compose(frame.prio, frame.pgn, frame.src, frame.dst),
                                    &frame.data,
                                ) =>
                        {
                            // Sent from the gateway's own address: where the
                            // answers to our ISO TP come.
                            if let Some(encoder) = &self.encoder {
                                encoder.gateway_is(frame.src);
                            }
                        }
                        Ok(frame) => self.push_frame(frame, &now, events),
                        Err(e) => events.push(DeviceEvent::Error(format!("{line}: {e}"))),
                    }
                }
                LineFormat::Mxpgn | LineFormat::ActisenseMxpgn => {
                    match mxpgn::parse_ordered(&line, self.checksum, self.format.mxpgn_order()) {
                        Ok(Some(frame)) => self.push_frame(frame, &now, events),
                        // The NMEA 0183 a MiniPlex mixes in, among which its
                        // Address Claims and Product Information: whole
                        // messages, which need no joining.
                        Ok(None) => match self.psmd.push(&line, self.checksum) {
                            Ok(Some(mut message)) => {
                                message.timestamp = Some(now.clone());
                                events.push(DeviceEvent::Frame(message));
                            }
                            Ok(None) => {}
                            Err(e) => events.push(DeviceEvent::Error(format!("{line}: {e}"))),
                        },
                        Err(e) => events.push(DeviceEvent::Error(format!("{line}: {e}"))),
                    }
                }
                LineFormat::SeaSmart => match seasmart::parse_sentence(&line, self.checksum) {
                    Ok(Some(mut frame)) => {
                        frame.timestamp = Some(now.clone());
                        events.push(DeviceEvent::Frame(frame))
                    }
                    // The NMEA 0183 a gateway mixes in.
                    Ok(None) => {}
                    Err(e) => events.push(DeviceEvent::Error(format!("{line}: {e}"))),
                },
                LineFormat::Candump => {
                    match candump::parse_line_for(&line, self.protocol.standard_frames()) {
                        Ok(Some(frame)) => self.push_frame(frame, &now, events),
                        // A frame of the other kind (11-bit on an NMEA 2000
                        // bus), or the Angstrom banner.
                        Ok(None) => {}
                        Err(e) => events.push(DeviceEvent::Error(format!("{line}: {e}"))),
                    }
                }
                LineFormat::N2kAscii => match actisense_ascii::parse_line(&line) {
                    Ok(mut frame) => {
                        frame.timestamp = Some(now.clone());
                        events.push(DeviceEvent::Frame(frame))
                    }
                    Err(e) => events.push(DeviceEvent::Error(format!("{line}: {e}"))),
                },
            }
        }
        // The window a Clear To Send asked for.
        self.poll_tp(events);
    }
}

impl Decoder {
    fn poll_tp(&mut self, events: &mut Vec<DeviceEvent>) {
        if let Some(encoder) = &self.encoder {
            let due = encoder.poll(super::now_ms());
            if !due.is_empty() {
                events.push(DeviceEvent::SendBytes(due));
            }
        }
    }
}

/// The ISO TP transfers an [`Encoder`] sends.
#[derive(Default)]
struct IsoTp {
    sender: TpSender,
    /// The gateway puts its own source address on what it sends, so the
    /// transfers are its, and the answers come to it.
    rewrites_source: bool,
    /// Its address, once an echo has shown it.
    gateway: Option<u8>,
}

/// The source the transfers through a gateway that rewrites it are kept
/// under, whatever the application gave.
const GATEWAY_SOURCE: u8 = crate::engine::ADDR_NULL;

pub struct Encoder {
    format: LineFormat,
    /// Decides which RAW transmissions are fast-packet framed.
    protocol: BusProtocol,
    /// Per-(pgn, src) fast-packet TX sequence counters, mod 8.
    seq: Mutex<HashMap<(u32, u8), u8>>,
    /// Bytes to write as soon as the link opens.
    init: Vec<u8>,
    sent: Sent,
    tp: Mutex<IsoTp>,
}

impl Encoder {
    #[cfg(test)]
    pub fn new(format: LineFormat) -> Self {
        Self::with_protocol(format, BusProtocol::Nmea2000)
    }

    pub fn with_protocol(format: LineFormat, protocol: BusProtocol) -> Self {
        Self {
            format,
            protocol,
            seq: Mutex::new(HashMap::new()),
            init: Vec::new(),
            sent: Sent::default(),
            tp: Mutex::new(IsoTp::default()),
        }
    }

    /// The gateway puts its own source address on what it transmits: the
    /// answers to an ISO TP transfer come to it, not to the source the
    /// application gave.
    pub fn rewrites_source(self, rewrites: bool) -> Self {
        self.tp.lock().expect("tp mutex").rewrites_source = rewrites;
        self
    }

    /// Whether a long message goes by ISO TP: a 29-bit CAN frame per line.
    fn sends_iso_tp(&self) -> bool {
        matches!(self.format, LineFormat::YdwgRaw | LineFormat::Candump)
            && !self.protocol.standard_frames()
    }

    /// The gateway's address, from an echo of what it sent for us.
    fn gateway_is(&self, address: u8) {
        self.tp.lock().expect("tp mutex").gateway = Some(address);
    }

    /// Feed a TP.CM from the bus to the transfers; the bytes it answers.
    fn tp_frame(&self, now_ms: u64, frame: &RawFrame) -> Vec<u8> {
        let mut tp = self.tp.lock().expect("tp mutex");
        let answer = if tp.rewrites_source {
            // Ours are the gateway's; until an echo shows its address, an
            // answer to anyone may be one.
            if tp.gateway.is_some_and(|gw| gw != frame.dst) {
                return Vec::new();
            }
            let mut to_us = frame.clone();
            to_us.dst = GATEWAY_SOURCE;
            tp.sender.on_frame(now_ms, &to_us)
        } else {
            tp.sender.on_frame(now_ms, frame)
        };
        drop(tp);
        self.write_tp(&answer)
    }

    /// The lines for ISO TP `frames`.
    fn write_tp(&self, frames: &[RawFrame]) -> Vec<u8> {
        let mut out = String::new();
        for f in frames {
            self.write_can(&mut out, f.prio, f.pgn, f.src, f.dst, &f.data);
        }
        out.into_bytes()
    }

    /// One CAN frame, as a line.
    fn write_can(&self, out: &mut String, prio: u8, pgn: u32, src: u8, dst: u8, data: &[u8]) {
        // An 11-bit frame's identifier is its message type, its pgn.
        let extended = !self.protocol.standard_frames();
        let canid = if extended {
            iso11783_compose(prio, pgn, src, dst)
        } else {
            pgn
        };
        match self.format {
            LineFormat::Candump => {
                candump::write_log(out, canid, extended, data, super::now_ms());
                return;
            }
            LineFormat::YdwgRaw => {
                self.sent.note(canid, data);
                let _ = write!(out, "{canid:08X}");
                for b in data {
                    let _ = write!(out, " {b:02X}");
                }
            }
            _ => {
                let one = RawFrame::new(None, prio, pgn, src, dst, data.iter().copied());
                let _ = mxpgn::write_ordered(out, &one, true, self.format.mxpgn_order());
            }
        }
        out.push_str("\r\n");
    }

    /// Record in `sent` the RAW frames written.
    pub fn with_sent(mut self, sent: Sent) -> Self {
        self.sent = sent;
        self
    }

    /// Write `init` as soon as the link opens.
    pub fn with_init(mut self, init: &[u8]) -> Self {
        self.init = init.to_vec();
        self
    }

    fn next_seq(&self, pgn: u32, src: u8) -> u8 {
        let mut m = self.seq.lock().expect("seq mutex");
        let e = m.entry((pgn, src)).or_insert(0);
        let s = *e;
        *e = (*e + 1) & 0x07;
        s
    }
}

impl DeviceEncoder for Encoder {
    fn init_bytes(&self) -> Vec<u8> {
        self.init.clone()
    }

    fn encode_frame(&self, frame: &RawFrame) -> Option<Vec<u8>> {
        // Synthetic canboat-internal PGNs never go on the wire.
        if frame.pgn >= fastpacket::CANBOAT_PGN_START {
            return None;
        }
        let mut out = String::with_capacity(64);
        if self.format == LineFormat::Mxpgn
            && let Some(pgn) = miniplex_request(frame)
        {
            // Its answers come as `$PSMDNC` / `$PSMDNP` only for a request
            // the MiniPlex made itself.
            return Some(mxpgn::request(pgn, frame.dst).into_bytes());
        }
        match self.format {
            LineFormat::N2kAscii => {
                actisense_ascii::write_line(&mut out, frame).ok()?;
                out.push_str("\r\n");
            }
            LineFormat::SeaSmart => {
                if frame.data.len() > FASTPACKET_MAX_SIZE {
                    log::warn!(
                        "not sending PGN {}: {} bytes exceed a fast-packet",
                        frame.pgn,
                        frame.data.len()
                    );
                    return None;
                }
                if frame.dst != 255 && (frame.pgn >> 8) & 0xff < 240 {
                    log::warn!(
                        "PGN {} for {} goes out in $PCDIN, which has no destination: \
                         the gateway decides where it goes",
                        frame.pgn,
                        frame.dst
                    );
                }
                seasmart::write_line(&mut out, frame).ok()?;
                out.push_str("\r\n");
            }
            LineFormat::YdwgRaw
            | LineFormat::Mxpgn
            | LineFormat::ActisenseMxpgn
            | LineFormat::Candump => {
                // An 11-bit frame's identifier is its message type, its pgn.
                if self.protocol.standard_frames() && frame.pgn > 0x7FF {
                    log::warn!(
                        "not sending message type {:#X}: an 11-bit identifier is at most 0x7FF",
                        frame.pgn
                    );
                    return None;
                }
                let write_one = |out: &mut String, data: &[u8]| {
                    self.write_can(out, frame.prio, frame.pgn, frame.src, frame.dst, data)
                };
                if self.protocol.packet_type(frame.pgn) == FramePacketType::Fast {
                    let seq = self.next_seq(frame.pgn, frame.src);
                    let Some(chunks) = fastpacket::fragment(seq, &frame.data) else {
                        log::warn!(
                            "not sending PGN {}: {} bytes exceed one fast-packet",
                            frame.pgn,
                            frame.data.len()
                        );
                        return None;
                    };
                    for chunk in chunks {
                        write_one(&mut out, &chunk);
                    }
                } else if frame.data.len() > 8
                    && frame.data.len() <= TP_MAX_SIZE
                    && self.sends_iso_tp()
                {
                    // One line is one CAN frame: a longer message goes by ISO
                    // TP, its announcement now and its data at the deadlines.
                    let mut tp = self.tp.lock().expect("tp mutex");
                    let src = if tp.rewrites_source {
                        GATEWAY_SOURCE
                    } else {
                        frame.src
                    };
                    let announce = match tp.sender.send(
                        super::now_ms(),
                        frame.pgn,
                        src,
                        frame.dst,
                        &frame.data,
                    ) {
                        Ok(announce) => announce,
                        Err(e) => {
                            log::warn!("not sending PGN {}: {e}", frame.pgn);
                            return None;
                        }
                    };
                    drop(tp);
                    return Some(self.write_tp(&announce));
                } else if frame.data.len() > 8 {
                    // One line is one CAN frame, and this gateway gets no ISO
                    // TP from canboat.
                    log::warn!(
                        "not sending PGN {}: {} bytes do not fit one CAN frame",
                        frame.pgn,
                        frame.data.len()
                    );
                    return None;
                } else {
                    write_one(&mut out, &frame.data);
                }
            }
        }
        Some(out.into_bytes())
    }

    fn next_deadline(&self) -> Option<u64> {
        self.tp.lock().expect("tp mutex").sender.next_deadline()
    }

    fn poll(&self, now_ms: u64) -> Vec<u8> {
        let due = self.tp.lock().expect("tp mutex").sender.poll(now_ms);
        self.write_tp(&due)
    }
}

/// An encoder its decoder shares.
impl DeviceEncoder for Arc<Encoder> {
    fn init_bytes(&self) -> Vec<u8> {
        (**self).init_bytes()
    }

    fn encode_frame(&self, frame: &RawFrame) -> Option<Vec<u8>> {
        (**self).encode_frame(frame)
    }

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

    fn poll(&self, now_ms: u64) -> Vec<u8> {
        (**self).poll(now_ms)
    }
}

pub use crate::engine::format::mxpgn::MINIPLEX_MXPGN_ALL;

/// [`MINIPLEX_MXPGN_ALL`], then a request for every device's Address Claim
/// and Product Information (`$PSMDNR`), so that the devices on the bus are
/// known from the start. The MiniPlex passes on Product Information only
/// for a request it made.
pub fn miniplex_init() -> Vec<u8> {
    let mut init = MINIPLEX_MXPGN_ALL.to_vec();
    for pgn in MINIPLEX_REQUESTED {
        init.extend_from_slice(mxpgn::request(pgn, 255).as_bytes());
    }
    init
}

/// The PGNs a MiniPlex answers requests for with its own sentences.
const MINIPLEX_REQUESTED: [u32; 2] = [60928, 126996];

/// The PGN `frame` requests, if it is an ISO Request (59904) for one that
/// a MiniPlex must request itself ([`MINIPLEX_REQUESTED`]).
fn miniplex_request(frame: &RawFrame) -> Option<u32> {
    let [a, b, c, ..] = frame.data[..] else {
        return None;
    };
    let pgn = u32::from_le_bytes([a, b, c, 0]);
    (frame.pgn == 59904 && MINIPLEX_REQUESTED.contains(&pgn)).then_some(pgn)
}

pub use crate::engine::format::ydwg02::YDNU_RAW_MODE;

/// Start the gateway reader/writer threads over any byte transport,
/// writing `init` as soon as the link opens. `ignore_checksum` accepts
/// `$PCDIN` and `$MXPGN` sentences whose checksum is wrong.
pub fn run(
    reader: Box<dyn Read + Send>,
    writer: Box<dyn Write + Send>,
    format: LineFormat,
    protocol: BusProtocol,
    init: &[u8],
    ignore_checksum: bool,
    rewrites_source: bool,
) -> DeviceHandle {
    let sent = Sent::default();
    let encoder = Arc::new(
        Encoder::with_protocol(format, protocol)
            .with_init(init)
            .with_sent(sent.clone())
            .rewrites_source(rewrites_source),
    );
    super::run(
        Decoder::with_protocol(format, protocol)
            .with_sent(sent)
            .ignore_checksum(ignore_checksum)
            .with_encoder(encoder.clone()),
        encoder,
        reader,
        writer,
    )
}

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

    #[test]
    fn an_11_bit_identifier_too_large_is_not_sent() {
        let enc = Encoder::with_protocol(LineFormat::Candump, BusProtocol::Quick);
        assert!(enc.encode_frame(&frame(0x800, &[1])).is_none());
        assert!(enc.encode_frame(&frame(0x7FF, &[1])).is_some());
    }

    #[test]
    fn candump_lines_go_both_ways() {
        let enc = Encoder::new(LineFormat::Candump);
        // A fast-packet PGN goes out as its frames, each a log line.
        let out = enc.encode_frame(&frame(129029, &[0u8; 20])).expect("sent");
        let out = String::from_utf8(out).unwrap();
        assert_eq!(out.lines().count(), 3, "{out}");
        assert!(out.lines().all(|l| l.contains(" can0 0DF80507#")), "{out}");
        // And what it wrote reads back whole.
        let mut dec = Decoder::new(LineFormat::Candump);
        let got = decode_all(&mut dec, &out);
        assert_eq!(got.len(), 1);
        assert_eq!((got[0].pgn, got[0].src, got[0].data.len()), (129029, 7, 20));
        // Every shape is read; the timestamp is the host's.
        let got = decode_all(
            &mut dec,
            "  can0  09F80115   [8]  7E 47 2B 1D B8 39 05 03\n\
             <0x09f80115> [8] 7e 47 2b 1d b8 39 05 03\n",
        );
        assert_eq!(got.len(), 2);
        assert!(got.iter().all(|f| f.pgn == 129025 && f.timestamp.is_some()));
    }

    #[test]
    fn candump_on_a_quick_bus_is_11_bit() {
        let enc = Encoder::with_protocol(LineFormat::Candump, BusProtocol::Quick);
        let quick = RawFrame::new(None, 0, 0x123, 0, 255, [1, 2]);
        let out = String::from_utf8(enc.encode_frame(&quick).unwrap()).unwrap();
        assert!(out.ends_with(" can0 123#0102\n"), "{out}");
        let mut dec = Decoder::with_protocol(LineFormat::Candump, BusProtocol::Quick);
        let got = decode_all(&mut dec, &out);
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].pgn, 0x123);
        // A 29-bit frame is not of a Quick bus.
        assert!(decode_all(&mut dec, "(1.0) can0 09F80115#7E\n").is_empty());
    }

    fn frame(pgn: u32, data: &[u8]) -> RawFrame {
        RawFrame::new(None, 3, pgn, 7, 255, data.iter().copied())
    }

    fn decode_all(dec: &mut Decoder, text: &str) -> Vec<RawFrame> {
        let mut events = Vec::new();
        dec.decode(text.as_bytes(), &mut events);
        events
            .into_iter()
            .filter_map(|e| match e {
                DeviceEvent::Frame(f) => Some(f),
                _ => None,
            })
            .collect()
    }

    #[test]
    fn raw_single_frame_round_trips() {
        let f = frame(130306, &[0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08]);
        let enc = Encoder::new(LineFormat::YdwgRaw);
        let tx = String::from_utf8(enc.encode_frame(&f).unwrap()).unwrap();
        // TX shape: bare `<CANID> <bytes…>`; the gateway's RX lines add
        // a time-of-day and direction, which the decoder expects.
        let rx: String = tx
            .lines()
            .map(|l| format!("00:00:01.000 R {l}\r\n"))
            .collect();
        let mut dec = Decoder::new(LineFormat::YdwgRaw);
        let out = decode_all(&mut dec, &rx);
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].pgn, 130306);
        assert_eq!(out[0].data.as_slice(), f.data.as_slice());
    }

    #[test]
    fn raw_rejects_a_line_with_more_than_eight_bytes() {
        // The head of one line glued to the tail of the next (#1001): an
        // error and no frame, and the gateway goes on with the next line.
        let mut dec = Decoder::new(LineFormat::YdwgRaw);
        let mut events = Vec::new();
        dec.decode(
            b"17:31:22.999 R 0DF80514 60 2F FC 25 C0 8A 5C 84 41 6A\r\n",
            &mut events,
        );
        assert!(matches!(events.as_slice(), [DeviceEvent::Error(_)]));
        let out = decode_all(
            &mut dec,
            "17:31:23.000 R 09F80114 2B 49 79 21 D0 34 AD 01\r\n",
        );
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].pgn, 129025);
    }

    #[test]
    fn raw_fast_packet_fragments_and_reassembles() {
        // 129029 GNSS Position Data: 43-byte fast-packet payload.
        let payload: Vec<u8> = (0u8..43).collect();
        let f = frame(129029, &payload);
        let enc = Encoder::new(LineFormat::YdwgRaw);
        let tx = String::from_utf8(enc.encode_frame(&f).unwrap()).unwrap();
        assert_eq!(tx.lines().count(), 7, "43 bytes -> 1x6 + 6x7 chunk frames");
        let rx: String = tx
            .lines()
            .map(|l| format!("00:00:01.000 R {l}\r\n"))
            .collect();
        let mut dec = Decoder::new(LineFormat::YdwgRaw);
        let out = decode_all(&mut dec, &rx);
        assert_eq!(out.len(), 1, "fragments must reassemble to one frame");
        assert_eq!(out[0].pgn, 129029);
        assert_eq!(out[0].data.as_slice(), payload.as_slice());
    }

    /// canboat#1039: the echo of a frame canboat wrote is skipped, with
    /// the gateway's source address in place of ours; another client's
    /// transmit is passed on.
    /// The bytes the decoder asks to write.
    fn answers(dec: &mut Decoder, text: &str) -> String {
        let mut events = Vec::new();
        dec.decode(text.as_bytes(), &mut events);
        events
            .into_iter()
            .filter_map(|e| match e {
                DeviceEvent::SendBytes(b) => Some(String::from_utf8(b).unwrap()),
                _ => None,
            })
            .collect()
    }

    /// `text`'s lines, as a YDWG echoes them: marked `T`, from `gateway`.
    fn echoed(text: &str, gateway: u8) -> String {
        text.lines()
            .map(|l| {
                let (canid, data) = l.split_once(' ').unwrap();
                format!("00:00:01.000 T {}{gateway:02X} {data}\r\n", &canid[..6])
            })
            .collect()
    }

    #[test]
    fn a_long_j1939_message_goes_out_as_a_bam_at_the_deadlines() {
        let enc = Encoder::with_protocol(LineFormat::YdwgRaw, BusProtocol::J1939);
        let vin: Vec<u8> = (1..=20).collect();
        let cm = String::from_utf8(enc.encode_frame(&frame(65260, &vin)).unwrap()).unwrap();
        assert_eq!(cm, "1CECFF07 20 14 00 03 FF EC FE 00\r\n");
        let mut packets = String::new();
        let mut at = Vec::new();
        while let Some(due) = enc.next_deadline() {
            at.push(due);
            packets.push_str(&String::from_utf8(enc.poll(due)).unwrap());
        }
        assert_eq!(packets.lines().count(), 3, "{packets}");
        assert!(packets.lines().all(|l| l.starts_with("1CEBFF07 ")));
        assert!(packets.ends_with("1CEBFF07 03 0F 10 11 12 13 14 FF\r\n"));
        assert!(
            at.windows(2)
                .all(|w| w[1] >= w[0] + crate::engine::iso_tp::BAM_GAP_MS)
        );
    }

    /// A YDWG sends from its own address, so the receiver's Clear To Send
    /// comes to that, which its echo shows.
    #[test]
    fn a_clear_to_send_to_the_gateway_releases_the_window() {
        let sent = Sent::default();
        let enc = Arc::new(
            Encoder::with_protocol(LineFormat::YdwgRaw, BusProtocol::J1939)
                .with_sent(sent.clone())
                .rewrites_source(true),
        );
        let mut dec = Decoder::with_protocol(LineFormat::YdwgRaw, BusProtocol::J1939)
            .with_sent(sent)
            .with_encoder(enc.clone());
        let mut to_peer = frame(65260, &[7; 20]);
        to_peer.dst = 0x30;
        let rts = String::from_utf8(enc.encode_frame(&to_peer).unwrap()).unwrap();
        assert!(rts.starts_with("1CEC30FE 10 14 00 03 FF EC FE 00"), "{rts}");
        assert!(decode_all(&mut dec, &echoed(&rts, 0xE2)).is_empty());

        // A CTS to another address is not for us.
        let cts = |to: u8| format!("00:00:01.010 R 1CEC{to:02X}30 11 03 01 FF FF EC FE 00\r\n");
        assert_eq!(answers(&mut dec, &cts(0xE3)), "");
        let window = answers(&mut dec, &cts(0xE2));
        assert_eq!(window.lines().count(), 3, "{window}");
        assert!(
            window.lines().all(|l| l.starts_with("1CEB30FE ")),
            "{window}"
        );

        // The echoes of the window are not the bus's.
        assert!(decode_all(&mut dec, &echoed(&window, 0xE2)).is_empty());
        let eoma = "00:00:01.020 R 1CECE230 13 14 00 03 FF EC FE 00\r\n";
        decode_all(&mut dec, eoma);
        assert_eq!(enc.next_deadline(), None, "the transfer is done");
    }

    #[test]
    fn candump_sends_iso_tp_from_the_source_given() {
        let enc = Encoder::with_protocol(LineFormat::Candump, BusProtocol::J1939);
        let cm = String::from_utf8(enc.encode_frame(&frame(65260, &[7; 20])).unwrap()).unwrap();
        assert!(cm.ends_with(" can0 1CECFF07#20140003FFECFE00\n"), "{cm}");
        assert!(enc.next_deadline().is_some());
    }

    #[test]
    fn mxpgn_gets_no_iso_tp() {
        let enc = Encoder::with_protocol(LineFormat::Mxpgn, BusProtocol::J1939);
        assert!(enc.encode_frame(&frame(65260, &[7; 20])).is_none());
        assert_eq!(enc.next_deadline(), None);
    }

    #[test]
    fn raw_skips_only_the_echoes_of_our_own_transmits() {
        let sent = Sent::default();
        let enc = Encoder::new(LineFormat::YdwgRaw).with_sent(sent.clone());
        let mut dec = Decoder::new(LineFormat::YdwgRaw).with_sent(sent);
        let ours = RawFrame::new(None, 6, 59904, 0, 0x0a, [0x14, 0xf0, 0x01]);
        let tx = String::from_utf8(enc.encode_frame(&ours).unwrap()).unwrap();
        assert_eq!(tx, "18EA0A00 14 F0 01\r\n");
        // Echoed from the gateway's address, 0xE2, twice: the second is
        // another client sending the same.
        let echo = "00:00:01.000 T 18EA0AE2 14 F0 01\r\n";
        assert!(decode_all(&mut dec, echo).is_empty());
        let other = decode_all(&mut dec, echo);
        assert_eq!(other.len(), 1);
        assert_eq!((other[0].pgn, other[0].src), (59904, 0xe2));
    }

    /// Not the gateway's time of day, which can run backwards.
    #[test]
    fn frames_get_the_host_time() {
        let before = format_iso_ms(super::super::now_ms());
        let mut dec = Decoder::new(LineFormat::YdwgRaw);
        let out = decode_all(
            &mut dec,
            "00:00:01.010 R 09F11223 01 02 03 04 05 06 07 08\r\n",
        );
        let ts = out[0].timestamp.clone().unwrap();
        assert!(ts >= before && !ts.contains("T00:00:01.010"), "{ts}");
    }

    #[test]
    fn raw_buffers_partial_lines() {
        let mut dec = Decoder::new(LineFormat::YdwgRaw);
        // A split line is buffered across reads and completes on the
        // second push.
        let out = decode_all(&mut dec, "00:00:01.010 R 09F1");
        assert!(out.is_empty());
        let out = decode_all(&mut dec, "1223 01 02 03 04 05 06 07 08\r\n");
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].pgn, 127250);
    }

    /// A MiniPlex stream: `$MXPGN` frames among other NMEA 0183, a
    /// fast-packet in its frames; what canboat sends goes in the send form.
    #[test]
    fn mxpgn_frames_are_joined_and_sent_in_the_send_form() {
        let mut dec = Decoder::new(LineFormat::Mxpgn);
        let rmc = "$GPRMC,123519,A,4807.038,N,01131.000,E,022.4,084.4,230394,003.1,W*6A\r\n";
        let mut text = String::from(rmc);
        let sent = frame(129029, &(0u8..20).collect::<Vec<_>>());
        let mut wire = String::new();
        for chunk in crate::engine::fastpacket::fragment(0, &sent.data).unwrap() {
            let one = RawFrame::new(None, 3, 129029, 7, 255, chunk);
            crate::engine::format::mxpgn::write_line(&mut wire, &one, false).unwrap();
            wire.push_str("\r\n");
        }
        text.push_str(&wire);
        text.push_str(rmc);
        let out = decode_all(&mut dec, &text);
        assert_eq!(out.len(), 1);
        assert_eq!((out[0].pgn, out[0].src), (129029, 7));
        assert_eq!(out[0].data.as_slice(), sent.data.as_slice());

        let enc = Encoder::new(LineFormat::Mxpgn);
        let request = RawFrame::new(None, 6, 59904, 0, 9, [0x16, 0xf0, 0x01]);
        let bytes = enc.encode_frame(&request).unwrap();
        assert_eq!(bytes, b"$MXPGN,00EA00,E309,01F016*6B\r\n");
        let lines = enc.encode_frame(&sent).unwrap();
        let lines = String::from_utf8(lines).unwrap();
        assert_eq!(lines.lines().count(), 3);
        assert!(lines.lines().all(|l| l.starts_with("$MXPGN,01F805,B800,")));
    }

    /// A SeaSmart stream: whole messages among other NMEA 0183, with the
    /// host's time; what canboat sends is one sentence per message.
    #[test]
    fn seasmart_messages_pass_whole_both_ways() {
        let payload: Vec<u8> = (0u8..43).collect();
        let f = frame(129029, &payload);
        let enc = Encoder::new(LineFormat::SeaSmart);
        let tx = String::from_utf8(enc.encode_frame(&f).unwrap()).unwrap();
        assert_eq!(tx.lines().count(), 1);
        assert!(tx.starts_with("$PCDIN,01F805,00000000,07,"), "{tx}");

        let rmc = "$GPRMC,123519,A,4807.038,N,01131.000,E,022.4,084.4,230394,003.1,W*6A\r\n";
        let mut dec = Decoder::new(LineFormat::SeaSmart);
        let mut events = Vec::new();
        dec.decode(format!("{rmc}{tx}{rmc}").as_bytes(), &mut events);
        let [DeviceEvent::Frame(out)] = events.as_slice() else {
            panic!("{events:?}");
        };
        assert_eq!((out.pgn, out.src), (129029, 7));
        assert_eq!(out.data.as_slice(), payload.as_slice());
        assert!(!out.timestamp.as_deref().unwrap().starts_with("1970"));

        assert!(enc.encode_frame(&frame(129029, &[0; 224])).is_none());
    }

    /// An Actisense stream (a W2K-2): `$MXPGN` with the data in bus order,
    /// a fast-packet in its frames; what canboat sends goes the same way.
    #[test]
    fn actisense_mxpgn_is_in_bus_order_both_ways() {
        let mut dec = Decoder::new(LineFormat::ActisenseMxpgn);
        let w2k2 = "$MXPGN,01F805,3806,402BFF04510158F4*1D\r\n\
                    $MXPGN,01F805,3806,41284097D2A5B553*61\r\n\
                    $MXPGN,01F805,3806,426107F08DE6544A*1C\r\n\
                    $MXPGN,01F805,3806,43DEC00070FF6B06*17\r\n\
                    $MXPGN,01F805,3806,440000000012FC04*15\r\n\
                    $MXPGN,01F805,3806,456A0272029F1100*1B\r\n\
                    $MXPGN,01F805,3806,460000FFFFFFFFFF*15\r\n";
        let out = decode_all(&mut dec, w2k2);
        assert_eq!(out.len(), 1);
        assert_eq!((out[0].pgn, out[0].src, out[0].data.len()), (129029, 6, 43));
        // The SID, then days since 1970: 2026-10-14.
        assert_eq!(&out[0].data[..3], &[0xff, 0x04, 0x51]);

        let enc = Encoder::new(LineFormat::ActisenseMxpgn);
        let request = RawFrame::new(None, 6, 59904, 0, 255, [0x14, 0xf0, 0x01]);
        let bytes = enc.encode_frame(&request).unwrap();
        assert_eq!(bytes, b"$MXPGN,00EA00,E3FF,14F001*60\r\n");
    }

    /// A wrong checksum is an error, unless it is to be ignored.
    #[test]
    fn a_wrong_checksum_is_ignored_only_when_asked() {
        let wrong = "$PCDIN,01F119,00000000,0F,2AAF00D1067414FF*21\r\n\
                     $MXPGN,01F801,2801,C1308AC40C5DE343*18\r\n";
        for (format, pgn) in [(LineFormat::SeaSmart, 127257), (LineFormat::Mxpgn, 129025)] {
            let mut events = Vec::new();
            Decoder::new(format).decode(wrong.as_bytes(), &mut events);
            assert!(
                events.iter().all(|e| matches!(e, DeviceEvent::Error(_))),
                "{events:?}"
            );
            let mut dec = Decoder::new(format).ignore_checksum(true);
            let out = decode_all(&mut dec, wrong);
            assert_eq!(out.len(), 1, "{format:?}");
            assert_eq!(out[0].pgn, pgn);
        }
    }

    /// A MiniPlex's Address Claims and Product Information come through
    /// whole; canboat asks it for them on connect, and an ISO Request for
    /// one goes as the MiniPlex's own request.
    #[test]
    fn miniplex_devices_are_read_and_requested() {
        let mut dec = Decoder::new(LineFormat::Mxpgn);
        let out = decode_all(
            &mut dec,
            "$PSMDNC,R,C032870022220839,7*31\r\n\
             $PSMDNP,A,7,2.100,11369,2,1*70\r\n\
             $PSMDNP,B,NMEA 2000<->0183 Gateway (NGW-1),133177*61\r\n\
             $PSMDNP,C,1.100, 2.660,NGW-1-USB  hv1.03*5C\r\n",
        );
        let got: Vec<_> = out.iter().map(|f| (f.pgn, f.src, f.data.len())).collect();
        assert_eq!(got, [(60928, 7, 8), (126996, 7, 134)]);
        assert!(out.iter().all(|f| f.timestamp.is_some()));

        let init = String::from_utf8(miniplex_init()).unwrap();
        assert!(
            init.ends_with("$PSMDNR,60928*0F\r\n$PSMDNR,126996*39\r\n"),
            "{init}"
        );

        let enc = Encoder::new(LineFormat::Mxpgn);
        let ask = |pgn: u32, dst: u8| {
            let [a, b, c, _] = pgn.to_le_bytes();
            let request = RawFrame::new(None, 6, 59904, 0, dst, [a, b, c]);
            String::from_utf8(enc.encode_frame(&request).unwrap()).unwrap()
        };
        assert_eq!(ask(126996, 255), "$PSMDNR,126996*39\r\n");
        assert!(ask(60928, 7).starts_with("$PSMDNR,60928,7*"));
        // Any other request goes on the bus as it is.
        assert!(ask(126998, 7).starts_with("$MXPGN,00EA00,"));
    }

    #[test]
    fn ascii_lines_pass_complete_frames() {
        // Round-trip a coalesced frame through the N2K ASCII writer and
        // parser: W2K lines carry complete messages, no reassembly.
        let payload: Vec<u8> = (0u8..43).collect();
        let f = frame(129029, &payload);
        let enc = Encoder::new(LineFormat::N2kAscii);
        let tx = String::from_utf8(enc.encode_frame(&f).unwrap()).unwrap();
        assert_eq!(tx.lines().count(), 1);
        let mut dec = Decoder::new(LineFormat::N2kAscii);
        let out = decode_all(&mut dec, &tx);
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].pgn, 129029);
        assert_eq!(out[0].data.as_slice(), payload.as_slice());
    }

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

    /// What `run` writes as the link opens, before any frame.
    fn written_on_open(init: &[u8]) -> Vec<u8> {
        let sink = Sink::default();
        run(
            Box::new(std::io::empty()),
            Box::new(sink.clone()),
            LineFormat::YdwgRaw,
            BusProtocol::Nmea2000,
            init,
            false,
            true,
        )
        .join();
        sink.0.lock().unwrap().clone()
    }

    #[test]
    fn ydnu_is_put_in_raw_mode_on_open() {
        // A network gateway is sent nothing; a YDNU-02 gets `0` + LF.
        assert!(written_on_open(&[]).is_empty());
        assert_eq!(written_on_open(YDNU_RAW_MODE), b"0\n");
    }

    #[test]
    fn synthetic_pgns_are_dropped() {
        let enc = Encoder::new(LineFormat::YdwgRaw);
        assert!(enc.encode_frame(&frame(0x40200, &[1, 2, 3])).is_none());
    }
}