rns_net/interface/

kiss_iface.rs

//! KISS interface with flow control and TNC configuration.
//!
//! Matches Python `KISSInterface.py` — opens a serial port,
//! sends TNC configuration commands, handles KISS framing with
//! optional flow control.

use std::collections::VecDeque;
use std::io::{self, Read, Write};
use std::os::fd::AsRawFd;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use rns_core::transport::types::InterfaceId;

use crate::event::{Event, EventSender};
use crate::interface::{
    lock_or_recover, poll_readable, wait_while_active, ListenerControl, UnavailableWriter, Writer,
};
use crate::kiss;
use crate::serial::{Parity, SerialConfig, SerialPort};

/// Configuration for a KISS interface.
#[derive(Debug, Clone)]
pub struct KissIfaceConfig {
    pub name: String,
    pub port: String,
    pub speed: u32,
    pub data_bits: u8,
    pub parity: Parity,
    pub stop_bits: u8,
    pub preamble: u16,                // ms, default 350
    pub txtail: u16,                  // ms, default 20
    pub persistence: u8,              // 0-255, default 64
    pub slottime: u16,                // ms, default 20
    pub flow_control: bool,           // default false
    pub beacon_interval: Option<u32>, // seconds
    pub beacon_data: Option<Vec<u8>>, // padded to 15 bytes
    pub interface_id: InterfaceId,
    pub ax25_source: Option<Ax25Address>,
    pub interface_type_name: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Ax25Address {
    pub callsign: String,
    pub ssid: u8,
}

impl Default for KissIfaceConfig {
    fn default() -> Self {
        KissIfaceConfig {
            name: String::new(),
            port: String::new(),
            speed: 9600,
            data_bits: 8,
            parity: Parity::None,
            stop_bits: 1,
            preamble: 350,
            txtail: 20,
            persistence: 64,
            slottime: 20,
            flow_control: false,
            beacon_interval: None,
            beacon_data: None,
            interface_id: InterfaceId(0),
            ax25_source: None,
            interface_type_name: "KISSInterface".into(),
        }
    }
}

/// Shared flow-control state between writer and reader threads.
struct FlowState {
    ready: bool,
    queue: VecDeque<Vec<u8>>,
    lock_time: Instant,
}

/// Writer that sends KISS-framed data over a serial port.
/// Handles flow control: when enabled, queues packets until CMD_READY.
struct KissWriter {
    confirmed_pending: Option<Vec<u8>>,
    file: std::fs::File,
    flow_control: bool,
    flow_state: Arc<Mutex<FlowState>>,
    ax25_source: Option<Ax25Address>,
}

impl Writer for KissWriter {
    fn send_frame_confirmed(&mut self, data: &[u8]) -> io::Result<()> {
        if !self.flow_control {
            return self.send_frame(data);
        }
        let data = self
            .ax25_source
            .as_ref()
            .map(|source| super::ax25_kiss::encode_ui_frame(source, data))
            .unwrap_or_else(|| data.to_vec());
        self.confirmed_pending = Some(kiss::frame(&data));
        self.flush()
    }

    fn flush(&mut self) -> io::Result<()> {
        if self.confirmed_pending.is_none() {
            return Ok(());
        }
        let mut state = lock_or_recover(&self.flow_state, "kiss flow state");
        if !state.ready || !state.queue.is_empty() {
            return Err(io::ErrorKind::WouldBlock.into());
        }
        state.ready = false;
        state.lock_time = Instant::now();
        drop(state);
        self.file.write_all(&self.confirmed_pending.take().unwrap())
    }
    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
        let data = self
            .ax25_source
            .as_ref()
            .map(|source| super::ax25_kiss::encode_ui_frame(source, data))
            .unwrap_or_else(|| data.to_vec());
        if self.flow_control {
            let mut state = lock_or_recover(&self.flow_state, "kiss flow state");
            if state.ready {
                state.ready = false;
                state.lock_time = Instant::now();
                drop(state);
                self.file.write_all(&kiss::frame(&data))
            } else {
                state.queue.push_back(data);
                Ok(())
            }
        } else {
            self.file.write_all(&kiss::frame(&data))
        }
    }
}

/// Start the KISS interface. Opens the port, configures TNC, spawns reader thread.
pub fn start(config: KissIfaceConfig, tx: EventSender) -> io::Result<Box<dyn Writer>> {
    start_with_control(config, tx).map(|(writer, _)| writer)
}

fn start_with_control(
    config: KissIfaceConfig,
    tx: EventSender,
) -> io::Result<(Box<dyn Writer>, ListenerControl)> {
    let serial_config = SerialConfig {
        path: config.port.clone(),
        baud: config.speed,
        data_bits: config.data_bits,
        parity: config.parity,
        stop_bits: config.stop_bits,
    };

    let control = ListenerControl::new();
    let flow_state = Arc::new(Mutex::new(FlowState {
        ready: true,
        queue: VecDeque::new(),
        lock_time: Instant::now(),
    }));
    let port = match SerialPort::open(&serial_config) {
        Ok(port) => port,
        Err(error) => {
            log::warn!(
                "[{}] could not open KISS port {}: {}; retrying",
                config.name,
                config.port,
                error
            );
            spawn_initial_reconnect(config, tx, flow_state, control.clone())?;
            return Ok((Box::new(UnavailableWriter), control));
        }
    };
    let reader_file = port.reader()?;
    let mut writer_file = port.writer()?;
    let flow_writer_file = port.writer()?;

    let id = config.interface_id;

    // Initial 2-second delay for TNC initialization (matches Python)
    thread::sleep(Duration::from_secs(2));

    // Send TNC configuration commands
    if let Err(error) = configure_tnc(&mut writer_file, &config) {
        log::warn!(
            "[{}] initial TNC configuration failed: {}; retrying",
            config.name,
            error
        );
        drop(reader_file);
        drop(writer_file);
        drop(flow_writer_file);
        spawn_initial_reconnect(config, tx, flow_state, control.clone())?;
        return Ok((Box::new(UnavailableWriter), control));
    }

    let _ = tx.send(Event::InterfaceUp(id, None, None));

    let reader_flow_state = flow_state.clone();

    // Spawn reader thread
    let reader_config = config.clone();
    let reader_control = control.clone();
    thread::Builder::new()
        .name(format!("kiss-reader-{}", id.0))
        .spawn(move || {
            reader_loop(
                reader_file,
                flow_writer_file,
                id,
                reader_config,
                tx,
                reader_flow_state,
                reader_control,
            );
        })?;

    Ok((
        Box::new(KissWriter {
            confirmed_pending: None,
            file: writer_file,
            flow_control: config.flow_control,
            flow_state,
            ax25_source: config.ax25_source.clone(),
        }),
        control,
    ))
}

fn spawn_initial_reconnect(
    config: KissIfaceConfig,
    tx: EventSender,
    flow_state: Arc<Mutex<FlowState>>,
    control: ListenerControl,
) -> io::Result<()> {
    thread::Builder::new()
        .name(format!("kiss-reconnect-{}", config.interface_id.0))
        .spawn(move || {
            if let Some((reader, flow_writer)) = reconnect(&config, &tx, &flow_state, &control) {
                let id = config.interface_id;
                reader_loop(reader, flow_writer, id, config, tx, flow_state, control);
            }
        })?;
    Ok(())
}

/// Send TNC configuration commands via KISS.
/// Matches Python `KISSInterface.configure_device()`.
fn configure_tnc(writer: &mut std::fs::File, config: &KissIfaceConfig) -> io::Result<()> {
    log::info!("[{}] configuring KISS interface parameters", config.name);

    // Preamble: value is ms/10, clamped to 0-255
    let preamble_val = (config.preamble / 10).min(255) as u8;
    writer.write_all(&kiss::command_frame(kiss::CMD_TXDELAY, &[preamble_val]))?;

    // TX tail: value is ms/10, clamped to 0-255
    let txtail_val = (config.txtail / 10).min(255) as u8;
    writer.write_all(&kiss::command_frame(kiss::CMD_TXTAIL, &[txtail_val]))?;

    // Persistence: raw value, clamped to 0-255
    writer.write_all(&kiss::command_frame(kiss::CMD_P, &[config.persistence]))?;

    // Slot time: value is ms/10, clamped to 0-255
    let slottime_val = (config.slottime / 10).min(255) as u8;
    writer.write_all(&kiss::command_frame(kiss::CMD_SLOTTIME, &[slottime_val]))?;

    // Flow control: send CMD_READY with 0x01 (matches Python setFlowControl)
    writer.write_all(&kiss::command_frame(kiss::CMD_READY, &[0x01]))?;

    log::info!("[{}] KISS interface configured", config.name);
    Ok(())
}

/// Reader loop: reads from serial, KISS-decodes, dispatches events.
/// Also handles flow control unlocking and beacon transmission.
fn reader_loop(
    mut reader: std::fs::File,
    mut flow_writer: std::fs::File,
    id: InterfaceId,
    config: KissIfaceConfig,
    tx: EventSender,
    flow_state: Arc<Mutex<FlowState>>,
    control: ListenerControl,
) {
    let mut decoder = kiss::Decoder::new();
    let mut buf = [0u8; 4096];
    let mut first_tx: Option<Instant> = None;

    loop {
        if control.should_stop() {
            return;
        }
        match poll_readable(reader.as_raw_fd(), Duration::from_millis(200)) {
            Ok(false) => continue,
            Ok(true) => {}
            Err(e) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] KISS poll error: {}", config.name, e);
                let _ = tx.send(Event::InterfaceDown(id));
                match reconnect(&config, &tx, &flow_state, &control) {
                    Some((new_reader, new_flow_writer)) => {
                        reader = new_reader;
                        flow_writer = new_flow_writer;
                        decoder = kiss::Decoder::new();
                        continue;
                    }
                    None => return,
                }
            }
        }
        match reader.read(&mut buf) {
            Ok(0) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] KISS port closed", config.name);
                let _ = tx.send(Event::InterfaceDown(id));
                match reconnect(&config, &tx, &flow_state, &control) {
                    Some((new_reader, new_flow_writer)) => {
                        reader = new_reader;
                        flow_writer = new_flow_writer;
                        decoder = kiss::Decoder::new();
                        continue;
                    }
                    None => return,
                }
            }
            Ok(n) => {
                for event in decoder.feed(&buf[..n]) {
                    match event {
                        kiss::KissEvent::DataFrame(data) => {
                            let data = if config.ax25_source.is_some() {
                                match super::ax25_kiss::decode_ui_frame(&data) {
                                    Some(data) => data.to_vec(),
                                    None => continue,
                                }
                            } else {
                                data
                            };
                            if tx
                                .send(Event::Frame {
                                    interface_id: id,
                                    data,
                                    rssi: None,
                                    snr: None,
                                })
                                .is_err()
                            {
                                return;
                            }
                        }
                        kiss::KissEvent::Ready => {
                            process_queue(&flow_state, &mut flow_writer, &mut first_tx, &config);
                        }
                    }
                }
            }
            Err(e) => {
                if control.should_stop() {
                    return;
                }
                log::warn!("[{}] KISS read error: {}", config.name, e);
                let _ = tx.send(Event::InterfaceDown(id));
                match reconnect(&config, &tx, &flow_state, &control) {
                    Some((new_reader, new_flow_writer)) => {
                        reader = new_reader;
                        flow_writer = new_flow_writer;
                        decoder = kiss::Decoder::new();
                        continue;
                    }
                    None => return,
                }
            }
        }

        // Flow control timeout check
        if config.flow_control {
            let state = lock_or_recover(&flow_state, "kiss flow state");
            if !state.ready && state.lock_time.elapsed() > Duration::from_secs(5) {
                drop(state);
                log::warn!("[{}] unlocking flow control due to timeout", config.name);
                process_queue(&flow_state, &mut flow_writer, &mut first_tx, &config);
            }
        }

        // Beacon check
        if let (Some(interval), Some(ref beacon_data)) =
            (config.beacon_interval, &config.beacon_data)
        {
            if let Some(first) = first_tx {
                if first.elapsed() > Duration::from_secs(interval as u64) {
                    log::debug!("[{}] transmitting beacon data", config.name);
                    // Pad to minimum 15 bytes
                    let mut frame = beacon_data.clone();
                    while frame.len() < 15 {
                        frame.push(0x00);
                    }
                    let _ = flow_writer.write_all(&kiss::frame(&frame));
                    first_tx = None;
                }
            }
        }
    }
}

/// Process the flow control queue: send next queued packet, mark ready.
fn process_queue(
    flow_state: &Arc<Mutex<FlowState>>,
    writer: &mut std::fs::File,
    first_tx: &mut Option<Instant>,
    _config: &KissIfaceConfig,
) {
    let mut state = lock_or_recover(flow_state, "kiss flow state");
    if let Some(data) = state.queue.pop_front() {
        state.ready = false;
        state.lock_time = Instant::now();
        drop(state);
        let _ = writer.write_all(&kiss::frame(&data));
        if first_tx.is_none() {
            *first_tx = Some(Instant::now());
        }
    } else {
        state.ready = true;
    }
}

/// Attempt to reconnect the serial port.
fn reconnect(
    config: &KissIfaceConfig,
    tx: &EventSender,
    flow_state: &Arc<Mutex<FlowState>>,
    control: &ListenerControl,
) -> Option<(std::fs::File, std::fs::File)> {
    loop {
        if !wait_while_active(control, Duration::from_secs(5)) {
            return None;
        }
        log::info!(
            "[{}] attempting to reconnect KISS port {}...",
            config.name,
            config.port
        );

        let serial_config = SerialConfig {
            path: config.port.clone(),
            baud: config.speed,
            data_bits: config.data_bits,
            parity: config.parity,
            stop_bits: config.stop_bits,
        };

        match SerialPort::open(&serial_config) {
            Ok(port) => {
                match (port.reader(), port.writer(), port.writer()) {
                    (Ok(reader), Ok(mut cfg_writer), Ok(flow_writer)) => {
                        // 2-second init delay
                        if !wait_while_active(control, Duration::from_secs(2)) {
                            return None;
                        }
                        if let Err(e) = configure_tnc(&mut cfg_writer, config) {
                            log::warn!("[{}] TNC config failed: {}", config.name, e);
                            continue;
                        }
                        // Reset flow state
                        let mut state = lock_or_recover(flow_state, "kiss flow state");
                        state.ready = true;
                        state.queue.clear();
                        drop(state);

                        let new_writer: Box<dyn Writer> = Box::new(KissWriter {
                            confirmed_pending: None,
                            file: cfg_writer,
                            flow_control: config.flow_control,
                            flow_state: flow_state.clone(),
                            ax25_source: config.ax25_source.clone(),
                        });
                        let _ = tx.send(Event::InterfaceUp(
                            config.interface_id,
                            Some(new_writer),
                            None,
                        ));
                        log::info!("[{}] KISS port reconnected", config.name);
                        return Some((reader, flow_writer));
                    }
                    _ => {
                        log::warn!("[{}] failed to get handles from serial port", config.name);
                    }
                }
            }
            Err(e) => {
                log::warn!("[{}] KISS reconnect failed: {}", config.name, e);
            }
        }
    }
}

// --- Factory implementation ---

use super::{InterfaceConfigData, InterfaceFactory, StartContext, StartResult};
use rns_core::transport::types::InterfaceInfo;
use std::collections::HashMap;

/// Factory for `KISSInterface`.
pub struct KissFactory;

impl InterfaceFactory for KissFactory {
    fn type_name(&self) -> &str {
        "KISSInterface"
    }

    fn default_ifac_size(&self) -> usize {
        8
    }

    fn parse_config(
        &self,
        name: &str,
        id: InterfaceId,
        params: &HashMap<String, String>,
    ) -> Result<Box<dyn InterfaceConfigData>, String> {
        let port = params
            .get("port")
            .cloned()
            .ok_or_else(|| "KISSInterface requires 'port'".to_string())?;

        let speed = params
            .get("speed")
            .and_then(|v| v.parse().ok())
            .unwrap_or(9600u32);

        let data_bits = params
            .get("databits")
            .and_then(|v| v.parse().ok())
            .unwrap_or(8u8);

        let parity = params
            .get("parity")
            .map(|v| match v.to_lowercase().as_str() {
                "e" | "even" => crate::serial::Parity::Even,
                "o" | "odd" => crate::serial::Parity::Odd,
                _ => crate::serial::Parity::None,
            })
            .unwrap_or(crate::serial::Parity::None);

        let stop_bits = params
            .get("stopbits")
            .and_then(|v| v.parse().ok())
            .unwrap_or(1u8);

        let preamble = params
            .get("preamble")
            .and_then(|v| v.parse().ok())
            .unwrap_or(350u16);

        let txtail = params
            .get("txtail")
            .and_then(|v| v.parse().ok())
            .unwrap_or(20u16);

        let persistence = params
            .get("persistence")
            .and_then(|v| v.parse().ok())
            .unwrap_or(64u8);

        let slottime = params
            .get("slottime")
            .and_then(|v| v.parse().ok())
            .unwrap_or(20u16);

        let flow_control = params
            .get("flow_control")
            .and_then(|v| crate::config::parse_bool_pub(v))
            .unwrap_or(false);

        let beacon_interval = params
            .get("id_interval")
            .or_else(|| params.get("beacon_interval"))
            .and_then(|v| v.parse().ok());

        let beacon_data = params
            .get("id_callsign")
            .or_else(|| params.get("beacon_data"))
            .map(|v| v.as_bytes().to_vec());

        Ok(Box::new(KissIfaceConfig {
            name: name.to_string(),
            port,
            speed,
            data_bits,
            parity,
            stop_bits,
            preamble,
            txtail,
            persistence,
            slottime,
            flow_control,
            beacon_interval,
            beacon_data,
            interface_id: id,
            ax25_source: None,
            interface_type_name: "KISSInterface".into(),
        }))
    }

    fn start(
        &self,
        config: Box<dyn InterfaceConfigData>,
        ctx: StartContext,
    ) -> std::io::Result<StartResult> {
        let kiss_config = *config
            .into_any()
            .downcast::<KissIfaceConfig>()
            .map_err(|_| {
                std::io::Error::new(std::io::ErrorKind::InvalidData, "wrong config type")
            })?;

        let id = kiss_config.interface_id;
        let name = kiss_config.name.clone();
        let interface_type_name = kiss_config.interface_type_name.clone();

        let info = InterfaceInfo {
            id,
            name,
            mode: ctx.mode,
            gravity: ctx.gravity,
            recursive_prs: ctx.recursive_prs,
            announces_from_internal: ctx.announces_from_internal,
            announces_to_internal: ctx.announces_to_internal,
            out_capable: true,
            in_capable: true,
            bitrate: Some(1200),
            airtime_profile: None,
            announce_rate_target: None,
            announce_rate_grace: 0,
            announce_rate_penalty: 0.0,
            announce_cap: rns_core::constants::ANNOUNCE_CAP,
            is_local_client: false,
            wants_tunnel: false,
            tunnel_id: None,
            mtu: rns_core::constants::MTU as u32,
            ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
            ia_freq: 0.0,
            ip_freq: 0.0,
            op_freq: 0.0,
            op_samples: 0,
            started: crate::time::now(),
        };

        let (writer, control) = start_with_control(kiss_config, ctx.tx)?;

        Ok(StartResult::Simple {
            control: Some(control),
            id,
            info,
            writer,
            interface_type_name,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::serial::open_pty_pair;
    use std::os::unix::io::{AsRawFd, FromRawFd};
    use std::sync::mpsc;

    /// Helper: poll an fd for reading with timeout (ms).
    fn poll_read(fd: i32, timeout_ms: i32) -> bool {
        let mut pfd = libc::pollfd {
            fd,
            events: libc::POLLIN,
            revents: 0,
        };
        let ret = unsafe { libc::poll(&mut pfd, 1, timeout_ms) };
        ret > 0
    }

    #[test]
    fn kiss_data_roundtrip() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let mut master_file = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let mut slave_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        // Write a KISS data frame to the master side
        let payload = vec![0x01, 0x02, 0x03, 0x04, 0x05];
        let framed = kiss::frame(&payload);
        master_file.write_all(&framed).unwrap();
        master_file.flush().unwrap();

        // Read from slave using KISS decoder
        assert!(poll_read(slave_file.as_raw_fd(), 2000));

        let mut decoder = kiss::Decoder::new();
        let mut buf = [0u8; 4096];
        let n = slave_file.read(&mut buf).unwrap();
        let events = decoder.feed(&buf[..n]);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0], kiss::KissEvent::DataFrame(payload));
    }

    #[test]
    fn detach_stops_idle_kiss_reader_without_reconnect() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let master = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let flow_writer = unsafe { std::fs::File::from_raw_fd(libc::dup(slave_fd)) };
        let slave = unsafe { std::fs::File::from_raw_fd(slave_fd) };
        let (tx, rx) = crate::event::channel();
        let control = ListenerControl::new();
        let reader_control = control.clone();
        let flow_state = Arc::new(Mutex::new(FlowState {
            ready: true,
            queue: VecDeque::new(),
            lock_time: Instant::now(),
        }));
        let config = KissIfaceConfig {
            name: "detach-test".into(),
            interface_id: InterfaceId(72),
            ..Default::default()
        };

        let reader = thread::spawn(move || {
            reader_loop(
                slave,
                flow_writer,
                InterfaceId(72),
                config,
                tx,
                flow_state,
                reader_control,
            )
        });
        thread::sleep(Duration::from_millis(50));
        control.request_stop();
        reader.join().unwrap();
        drop(master);
        assert!(rx.try_recv().is_err());
    }

    #[test]
    fn missing_kiss_port_recovers_after_it_appears() {
        let tempdir = tempfile::tempdir().unwrap();
        let port_path = tempdir.path().join("kiss-port");
        let (tx, rx) = crate::event::channel();
        let config = KissIfaceConfig {
            name: "late-kiss".into(),
            port: port_path.display().to_string(),
            interface_id: InterfaceId(74),
            ..Default::default()
        };

        let (_writer, control) = start_with_control(config, tx).unwrap();
        assert!(rx.try_recv().is_err());

        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let master = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let slave_path = std::fs::read_link(format!("/proc/self/fd/{slave_fd}")).unwrap();
        std::os::unix::fs::symlink(slave_path, &port_path).unwrap();
        let slave = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        let event = rx.recv_timeout(Duration::from_secs(10)).unwrap();
        assert!(matches!(
            event,
            Event::InterfaceUp(InterfaceId(74), Some(_), None)
        ));
        control.request_stop();
        drop(slave);
        drop(master);
    }

    #[test]
    fn kiss_writer_frames() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();

        let writer_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };
        let flow_state = Arc::new(Mutex::new(FlowState {
            ready: true,
            queue: VecDeque::new(),
            lock_time: Instant::now(),
        }));

        let mut writer = KissWriter {
            confirmed_pending: None,
            file: writer_file,
            flow_control: false,
            flow_state,
            ax25_source: None,
        };

        let payload = vec![0xC0, 0xDB, 0x01]; // includes bytes that need KISS escaping
        writer.send_frame(&payload).unwrap();

        // Read from master
        let mut master_file = unsafe { std::fs::File::from_raw_fd(master_fd) };
        assert!(poll_read(master_file.as_raw_fd(), 2000));

        let expected = kiss::frame(&payload);
        let mut buf = [0u8; 256];
        let n = master_file.read(&mut buf).unwrap();
        assert_eq!(&buf[..n], &expected[..]);
    }

    #[test]
    fn kiss_config_commands() {
        use std::time::Instant;

        let (master_fd, slave_fd) = open_pty_pair().unwrap();

        let mut writer_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };
        let config = KissIfaceConfig {
            preamble: 350,
            txtail: 20,
            persistence: 64,
            slottime: 20,
            ..Default::default()
        };

        configure_tnc(&mut writer_file, &config).unwrap();

        // Read all commands from master
        let mut master_file = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let deadline = Instant::now() + Duration::from_secs(2);
        let mut data = Vec::new();
        let mut buf = [0u8; 1024];
        while Instant::now() < deadline {
            let remaining_ms = deadline
                .saturating_duration_since(Instant::now())
                .as_millis()
                .min(i32::MAX as u128) as i32;
            if remaining_ms <= 0 || !poll_read(master_file.as_raw_fd(), remaining_ms) {
                break;
            }

            let n = master_file.read(&mut buf).unwrap();
            if n == 0 {
                break;
            }
            data.extend_from_slice(&buf[..n]);

            let have_all = data.windows(4).any(|w| {
                w[0] == kiss::FEND && w[1] == kiss::CMD_TXDELAY && w[2] == 35 && w[3] == kiss::FEND
            }) && data.windows(4).any(|w| {
                w[0] == kiss::FEND && w[1] == kiss::CMD_TXTAIL && w[2] == 2 && w[3] == kiss::FEND
            }) && data.windows(4).any(|w| {
                w[0] == kiss::FEND && w[1] == kiss::CMD_P && w[2] == 64 && w[3] == kiss::FEND
            }) && data.windows(4).any(|w| {
                w[0] == kiss::FEND && w[1] == kiss::CMD_SLOTTIME && w[2] == 2 && w[3] == kiss::FEND
            });
            if have_all {
                break;
            }
        }

        // Should contain TXDELAY command: FEND, CMD_TXDELAY, value, FEND
        // preamble: 350/10 = 35
        assert!(
            data.windows(4).any(|w| w[0] == kiss::FEND
                && w[1] == kiss::CMD_TXDELAY
                && w[2] == 35
                && w[3] == kiss::FEND),
            "should contain TXDELAY command"
        );

        // TXTAIL: 20/10 = 2
        assert!(
            data.windows(4).any(|w| w[0] == kiss::FEND
                && w[1] == kiss::CMD_TXTAIL
                && w[2] == 2
                && w[3] == kiss::FEND),
            "should contain TXTAIL command"
        );

        // Persistence: 64
        assert!(
            data.windows(4).any(|w| w[0] == kiss::FEND
                && w[1] == kiss::CMD_P
                && w[2] == 64
                && w[3] == kiss::FEND),
            "should contain P command"
        );

        // Slottime: 20/10 = 2
        assert!(
            data.windows(4).any(|w| w[0] == kiss::FEND
                && w[1] == kiss::CMD_SLOTTIME
                && w[2] == 2
                && w[3] == kiss::FEND),
            "should contain SLOTTIME command"
        );
    }

    #[test]
    fn kiss_flow_control_lock() {
        let flow_state = Arc::new(Mutex::new(FlowState {
            ready: true,
            queue: VecDeque::new(),
            lock_time: Instant::now(),
        }));

        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let writer_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        let mut writer = KissWriter {
            confirmed_pending: None,
            file: writer_file,
            flow_control: true,
            flow_state: flow_state.clone(),
            ax25_source: None,
        };

        // First send should go through (ready=true) and lock
        writer.send_frame(b"hello").unwrap();
        assert!(!flow_state.lock().unwrap().ready);

        // Second send should be queued (ready=false)
        writer.send_frame(b"world").unwrap();
        assert_eq!(flow_state.lock().unwrap().queue.len(), 1);

        // Simulate CMD_READY: process_queue
        let mut flow_writer = unsafe { std::fs::File::from_raw_fd(libc::dup(master_fd)) };
        let mut first_tx = None;
        let config = KissIfaceConfig::default();
        process_queue(&flow_state, &mut flow_writer, &mut first_tx, &config);

        // Queue should be empty now (dequeued "world"), but ready=false because it sent
        assert_eq!(flow_state.lock().unwrap().queue.len(), 0);
        assert!(!flow_state.lock().unwrap().ready);

        // Process again with empty queue: should set ready=true
        process_queue(&flow_state, &mut flow_writer, &mut first_tx, &config);
        assert!(flow_state.lock().unwrap().ready);

        // Clean up
        unsafe { libc::close(master_fd) };
    }

    #[test]
    fn kiss_flow_control_timeout() {
        let flow_state = Arc::new(Mutex::new(FlowState {
            ready: false,
            queue: VecDeque::new(),
            lock_time: Instant::now() - Duration::from_secs(6), // already timed out
        }));

        // Check that the timeout condition triggers
        let state = flow_state.lock().unwrap();
        assert!(!state.ready);
        assert!(state.lock_time.elapsed() > Duration::from_secs(5));
    }

    #[test]
    fn confirmed_kiss_frame_waits_for_ready_and_flushes_once() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let mut master = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let state = Arc::new(Mutex::new(FlowState {
            ready: false,
            queue: VecDeque::new(),
            lock_time: Instant::now(),
        }));
        let mut writer = KissWriter {
            confirmed_pending: None,
            file: unsafe { std::fs::File::from_raw_fd(slave_fd) },
            flow_control: true,
            flow_state: state.clone(),
            ax25_source: None,
        };
        assert_eq!(
            writer
                .send_frame_confirmed(b"confirmed")
                .unwrap_err()
                .kind(),
            io::ErrorKind::WouldBlock
        );
        assert!(state.lock().unwrap().queue.is_empty());
        assert!(!poll_read(master.as_raw_fd(), 20));
        state.lock().unwrap().ready = true;
        Writer::flush(&mut writer).unwrap();
        let mut actual = vec![0; kiss::frame(b"confirmed").len()];
        master.read_exact(&mut actual).unwrap();
        assert_eq!(actual, kiss::frame(b"confirmed"));
        Writer::flush(&mut writer).unwrap();
        assert!(!poll_read(master.as_raw_fd(), 20));
    }

    #[test]
    fn kiss_fragmented() {
        let (master_fd, slave_fd) = open_pty_pair().unwrap();
        let mut master_file = unsafe { std::fs::File::from_raw_fd(master_fd) };
        let slave_file = unsafe { std::fs::File::from_raw_fd(slave_fd) };

        let payload = vec![0x01, 0x02, 0x03, 0x04, 0x05];
        let framed = kiss::frame(&payload);
        let mid = framed.len() / 2;

        // Spawn reader thread first (it will block waiting for data)
        let (tx, rx) = mpsc::channel::<kiss::KissEvent>();
        let reader_thread = thread::spawn(move || {
            let mut reader = slave_file;
            let mut decoder = kiss::Decoder::new();
            let mut buf = [0u8; 4096];

            loop {
                match reader.read(&mut buf) {
                    Ok(n) if n > 0 => {
                        for event in decoder.feed(&buf[..n]) {
                            let _ = tx.send(event.clone());
                            if matches!(event, kiss::KissEvent::DataFrame(_)) {
                                return;
                            }
                        }
                    }
                    _ => return,
                }
            }
        });

        // Write first half
        master_file.write_all(&framed[..mid]).unwrap();
        master_file.flush().unwrap();

        thread::sleep(Duration::from_millis(50));

        // Write second half
        master_file.write_all(&framed[mid..]).unwrap();
        master_file.flush().unwrap();

        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
        assert_eq!(event, kiss::KissEvent::DataFrame(payload));

        let _ = reader_thread.join();
    }
}