rigless 0.0.5

// Copyright (c) 2026 Rigless Labs.
// SPDX-License-Identifier: MIT OR Apache-2.0
//! The far end of an asynchronous serial line: a chip, or a host.
//!
//! A line carries frames one way and frames the other, each on its own time. The controller times what
//! it sends and hands the far end each word as its stop bit ends; the far end offers what it has to send
//! and the controller times that too, so the same peer answers a controller at any baud. A peer that is a
//! chip answers on virtual time from what it was sent; a peer that is a host answers with whatever a
//! person or a program has written to it since the controller last asked, which is why the controller asks
//! at every stop of the session and not only at the end of each frame. What is on the far end is the
//! caller's business: a board hangs a chip, `rigless run` hangs a [`Host`], a test hangs a stand-in, and
//! the controller cannot tell which.

use std::any::Any;
use std::collections::VecDeque;
use std::io::{self, Read, Write};
use std::net::{TcpListener, TcpStream};

use crate::device::DeviceContext;
use crate::time::VirtualNanos;

/// What hangs on the far end of a serial line.
pub trait Peer: std::fmt::Debug + Send + Any {
    /// A word the controller finished sending: eight or nine bits as the controller frames them, at the
    /// instant its stop bit ended, which `context` gives.
    fn receive(&mut self, word: u16, context: &DeviceContext<'_>);

    /// The next word this peer has put on the line by `now`, oldest first, or `None` when it has nothing.
    ///
    /// The controller asks whenever its receiver is free: at the end of each frame it takes, and at every
    /// stop of the session, so a word a host wrote lands at the stop it was noticed at and one a chip
    /// scheduled lands at the first stop after its instant.
    fn transmit(&mut self, now: VirtualNanos) -> Option<u16>;

    /// Every word this peer has been sent, oldest first, or nothing where it keeps no record.
    fn received(&self) -> &[u16] {
        &[]
    }
}

/// How many bytes one look at the socket takes at most.
const CHUNK: usize = 4096;

/// How much virtual time passes between two looks at the socket for a connection or for bytes. A port
/// asks its far end at every frame and at every stop, tens of thousands of times a virtual second, and
/// a look is a system call; once a millisecond is a thousand a virtual second, which is nothing, and a
/// host that connects or writes waits at most that long to be noticed.
const LOOK_PERIOD: VirtualNanos = VirtualNanos(1_000_000);

/// A host on the far end, over a TCP socket the runtime listens on.
///
/// Bytes the host writes are taken when the controller asks and offered oldest first; bytes the controller
/// sends are written straight through. A word wider than a byte crosses as its low byte. Nothing here
/// waits: a host that has not connected, or has gone, is a far end with nothing to say that hears nothing,
/// and the run neither pauses for it nor notices it. A wire has one far end, so a host that connects while
/// another is there takes its place, the way a cable moved from one ground station to another does, and
/// what the first had written and the port had not yet taken goes with it.
#[derive(Debug)]
pub struct Host {
    listener: TcpListener,
    stream: Option<TcpStream>,
    incoming: VecDeque<u8>,
    outgoing: VecDeque<u8>,
    /// When the socket was last looked at, or never.
    looked: Option<VirtualNanos>,
}

impl Host {
    /// Listens on `port` of the loopback interface for a host, the latest to connect being the one on the wire.
    pub fn listen(port: u16) -> io::Result<Self> {
        let listener = TcpListener::bind(("127.0.0.1", port))?;
        listener.set_nonblocking(true)?;
        Ok(Self {
            listener,
            stream: None,
            incoming: VecDeque::new(),
            outgoing: VecDeque::new(),
            looked: None,
        })
    }

    /// The port the runtime listens on, which is the one asked for unless zero was, when it is whatever
    /// the system gave.
    pub fn port(&self) -> io::Result<u16> {
        self.listener.local_addr().map(|address| address.port())
    }

    /// Looks at the socket once a [`LOOK_PERIOD`] for a connection and for bytes, and writes what
    /// the port sent whenever there is a host to write to.
    fn serve(&mut self, now: VirtualNanos) {
        let look = self
            .looked
            .is_none_or(|looked| now.0 >= looked.0.saturating_add(LOOK_PERIOD.0));
        if look {
            self.looked = Some(now);
            if let Ok((stream, _)) = self.listener.accept()
                && stream.set_nonblocking(true).is_ok()
            {
                self.stream = Some(stream);
                self.incoming.clear();
            }
        }
        let Some(stream) = &mut self.stream else {
            self.outgoing.clear();
            return;
        };
        if look {
            let mut buffer = [0; CHUNK];
            loop {
                match stream.read(&mut buffer) {
                    Ok(0) => {
                        self.stream = None;
                        self.outgoing.clear();
                        return;
                    }
                    Ok(count) => self.incoming.extend(&buffer[..count]),
                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                    Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                    Err(_) => {
                        self.stream = None;
                        self.outgoing.clear();
                        return;
                    }
                }
            }
        }
        while let Some(&byte) = self.outgoing.front() {
            match stream.write(&[byte]) {
                Ok(_) => {
                    self.outgoing.pop_front();
                }
                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                Err(_) => {
                    self.stream = None;
                    self.outgoing.clear();
                    return;
                }
            }
        }
    }
}

impl Peer for Host {
    fn receive(&mut self, word: u16, context: &DeviceContext<'_>) {
        self.outgoing
            .push_back(u8::try_from(word & 0xFF).expect("masked to a byte"));
        self.serve(context.now());
    }

    fn transmit(&mut self, now: VirtualNanos) -> Option<u16> {
        self.serve(now);
        self.incoming.pop_front().map(u16::from)
    }
}

#[cfg(test)]
mod host_tests {
    use std::net::TcpStream;

    use super::*;

    /// A byte the host wrote is there by the next look, and not before the period has passed.
    #[test]
    fn the_socket_is_looked_at_once_a_period_and_a_byte_arrives_by_the_next_look() {
        let mut host = Host::listen(0).unwrap();
        let mut client = TcpStream::connect(("127.0.0.1", host.port().unwrap())).unwrap();
        client.write_all(b"a").unwrap();
        // The first look connects and reads; the socket may not have the byte yet, so keep looking
        // one period at a time, which is what a run does.
        let mut now = VirtualNanos(0);
        let first = loop {
            if let Some(word) = host.transmit(now) {
                break word;
            }
            now = VirtualNanos(now.0 + LOOK_PERIOD.0);
            assert!(now.0 < 1_000 * LOOK_PERIOD.0, "the byte never arrived");
        };
        assert_eq!(first, u16::from(b'a'));
        let looked_at = now;
        client.write_all(b"b").unwrap();
        for _ in 0..100 {
            assert_eq!(
                host.transmit(looked_at),
                None,
                "within the period the socket is not looked at again"
            );
        }
        let mut now = VirtualNanos(looked_at.0 + LOOK_PERIOD.0);
        let second = loop {
            if let Some(word) = host.transmit(now) {
                break word;
            }
            now = VirtualNanos(now.0 + LOOK_PERIOD.0);
            assert!(
                now.0 < looked_at.0 + 1_000 * LOOK_PERIOD.0,
                "the byte never arrived"
            );
        };
        assert_eq!(second, u16::from(b'b'));
    }
}