rigless 0.2.1

// Copyright (c) 2026 Rigless Labs.
// SPDX-License-Identifier: MIT OR Apache-2.0
//! An output port of the run's: a listener that takes one client at a time, when there's something
//! to send it, and what the run sends, which goes out between stretches. Nothing here ever waits,
//! the end of the run included. With no client the bytes go, as a probe nobody plugged in hears
//! nothing, and past [`STREAM_BYTES`] waiting for a slow one, the oldest go. Nothing in the run
//! learns whether anyone's there.

use std::collections::VecDeque;
use std::io::{ErrorKind, Read, Write};
use std::net::{TcpListener, TcpStream};
#[cfg(test)]
use std::time::Duration;

use crate::emulator::Error;

/// The most bytes a stream keeps for a client that reads slower than the run sends.
const STREAM_BYTES: usize = 1 << 20;

/// An output port, and its client when it has one.
#[derive(Debug)]
pub(super) struct Stream {
    listener: TcpListener,
    client: Option<TcpStream>,
    waiting: VecDeque<u8>,
    /// Every byte the run has sent, along the run as it stands, counting the ones nobody heard.
    sent: u64,
    /// The most the run has sent, which a run gone back over what it did sends again only past.
    furthest: u64,
}

impl Stream {
    /// A stream from `listener`, which it leaves non-blocking.
    pub(super) fn listen(listener: &TcpListener) -> Result<Self, Error> {
        let listener = listener.try_clone().map_err(Error::OutputPort)?;
        listener.set_nonblocking(true).map_err(Error::OutputPort)?;
        Ok(Self {
            listener,
            client: None,
            waiting: VecDeque::new(),
            sent: 0,
            furthest: 0,
        })
    }

    /// How many bytes the run has sent here, along the run as it stands.
    pub(super) fn sent(&self) -> u64 {
        self.sent
    }

    /// The run went back to where it had sent `sent` bytes, which it sends again only past what it
    /// sent before.
    pub(super) fn went_back(&mut self, sent: u64) {
        self.sent = sent;
    }

    /// The run goes a new way from here, so whatever it sends from now on is new.
    pub(super) fn new_way(&mut self) {
        self.furthest = self.sent;
    }

    /// Sends `bytes` to the client, or drops them when there's none. What the run sent before it
    /// went back is sent once.
    pub(super) fn send(&mut self, bytes: &[u8]) {
        let before = self.sent;
        self.sent += bytes.len() as u64;
        let fresh = usize::try_from(self.sent.saturating_sub(self.furthest.max(before)))
            .expect("no more than was sent");
        self.furthest = self.furthest.max(self.sent);
        let bytes = &bytes[bytes.len() - fresh..];
        self.take_client();
        if self.client.is_none() {
            return;
        }
        self.waiting.extend(bytes);
        let over = self.waiting.len().saturating_sub(STREAM_BYTES);
        self.waiting.drain(..over);
    }

    /// Writes what the client takes now, keeping the rest for the next time. With nothing waiting
    /// it costs nothing, since it's asked at the end of every stretch.
    pub(super) fn flush(&mut self) {
        if self.waiting.is_empty() {
            return;
        }
        let Some(client) = self.client.as_mut() else {
            self.waiting.clear();
            return;
        };
        while !self.waiting.is_empty() {
            let (front, _) = self.waiting.as_slices();
            match client.write(front) {
                Ok(0) => break,
                Ok(written) => {
                    self.waiting.drain(..written);
                }
                Err(error) if error.kind() == ErrorKind::WouldBlock => return,
                Err(_) => {
                    self.client = None;
                    self.waiting.clear();
                    return;
                }
            }
        }
    }

    /// At the end of the run: the client gets what's left that its connection takes at once, which
    /// it reads after the run's gone, and the rest goes, as it would for a client that slow while
    /// the run went on. What the client wrote is read out first, since a connection closed with
    /// that unread is reset, and what it had yet to read goes with it.
    pub(super) fn finish(mut self) {
        self.flush();
        if let Some(client) = self.client.as_mut() {
            gone(client);
        }
    }

    /// Takes a client that's connecting when there's none, or when the one it has has closed its
    /// end, and closes any other that is, so it hears that the port has one. It's asked only when
    /// there's something to send.
    fn take_client(&mut self) {
        while let Ok((client, _)) = self.listener.accept() {
            if self.client.as_mut().is_some_and(gone) {
                self.client = None;
                self.waiting.clear();
            }
            if self.client.is_none() && client.set_nonblocking(true).is_ok() {
                client.set_nodelay(true).ok();
                self.client = Some(client);
            }
        }
    }
}

/// Whether `client` has closed its end, or failed. What it sent is dropped, since a port only
/// sends, a few reads' worth at a time, so a client that floods the port isn't read forever.
fn gone(client: &mut TcpStream) -> bool {
    let mut bytes = [0; 1024];
    for _ in 0..16 {
        match client.read(&mut bytes) {
            Ok(0) => return true,
            Ok(_) => {}
            Err(error) => return error.kind() != ErrorKind::WouldBlock,
        }
    }
    false
}

#[cfg(test)]
mod tests {
    use std::io::{Read, Write};
    use std::net::Ipv4Addr;

    use super::*;

    fn listener() -> TcpListener {
        TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap()
    }

    /// With no client, what's sent goes at once and nothing waits, so a client that connects hears
    /// only what comes after, and the end of the run hands it the rest.
    #[test]
    fn a_stream_with_no_client_drops_and_never_waits() {
        let listening = listener();
        let mut stream = Stream::listen(&listening).unwrap();
        stream.send(b"nobody hears this");
        assert!(stream.waiting.is_empty());
        let mut client = TcpStream::connect(listening.local_addr().unwrap()).unwrap();
        stream.send(b"hello");
        stream.flush();
        stream.send(&vec![b'x'; STREAM_BYTES + 3]);
        assert_eq!(
            stream.waiting.len(),
            STREAM_BYTES,
            "the oldest go past the limit"
        );
        // The client reads only once the run's over, more than a socket holds.
        stream.waiting.truncate(3);
        stream.finish();
        let mut heard = Vec::new();
        client.read_to_end(&mut heard).unwrap();
        assert!(heard.starts_with(b"hello"), "{:?}", &heard[..8]);
        assert!(!heard.windows(6).any(|seen| seen == b"nobody"));
    }

    /// A client that closed its end makes way for the next one, even when nothing was sent since it
    /// went.
    #[test]
    fn a_client_that_went_makes_way_for_the_next() {
        let listening = listener();
        let mut stream = Stream::listen(&listening).unwrap();
        let address = listening.local_addr().unwrap();
        let first = TcpStream::connect(address).unwrap();
        stream.send(b"first");
        stream.flush();
        drop(first);
        let mut second = TcpStream::connect(address).unwrap();
        stream.send(b"second");
        stream.flush();
        second
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let mut heard = [0; 6];
        second.read_exact(&mut heard).unwrap();
        assert_eq!(&heard, b"second");
    }

    /// The end of the run doesn't wait on a client that reads slowly: it hands over what the
    /// connection takes at once, far less than a slow client would take given the time, and goes.
    #[test]
    #[expect(
        clippy::disallowed_methods,
        reason = "a client reading slowly on its own thread, while the stream finishes"
    )]
    fn the_end_of_the_run_never_waits_on_a_slow_client() {
        const LEFT: usize = 32 << 20;
        let listening = listener();
        let mut stream = Stream::listen(&listening).unwrap();
        let mut client = TcpStream::connect(listening.local_addr().unwrap()).unwrap();
        stream.send(b"x");
        stream.waiting.extend(std::iter::repeat_n(b'x', LEFT));
        let reading = std::thread::spawn(move || {
            let mut bytes = vec![0; 64 << 10];
            let mut heard = 0;
            while let Ok(read @ 1..) = client.read(&mut bytes) {
                heard += read;
                std::thread::sleep(Duration::from_millis(1));
            }
            heard
        });
        stream.finish();
        let heard = reading.join().unwrap();
        assert!(heard < LEFT, "{heard} bytes");
    }

    /// What the run sends again after going back is sent once, and what it sends a new way is
    /// sent.
    #[test]
    fn a_run_that_went_back_sends_each_byte_once() {
        let listening = listener();
        let mut stream = Stream::listen(&listening).unwrap();
        let mut client = TcpStream::connect(listening.local_addr().unwrap()).unwrap();
        stream.send(b"abc");
        stream.went_back(1);
        stream.send(b"bcde");
        stream.went_back(2);
        stream.new_way();
        stream.send(b"X");
        stream.finish();
        let mut heard = Vec::new();
        client.read_to_end(&mut heard).unwrap();
        assert_eq!(heard, b"abcdeX");
    }

    /// A client that wrote to the port still hears everything the end of the run handed over, with
    /// no reset, since the port reads what it wrote out before it closes.
    #[test]
    fn a_client_that_wrote_hears_everything_before_the_close() {
        let listening = listener();
        let mut stream = Stream::listen(&listening).unwrap();
        let mut client = TcpStream::connect(listening.local_addr().unwrap()).unwrap();
        client.write_all(b"?").unwrap();
        let sent = vec![b'y'; 64 << 10];
        stream.send(&sent);
        stream.finish();
        let mut heard = Vec::new();
        client.read_to_end(&mut heard).unwrap();
        assert!(heard == sent, "{} of {} bytes", heard.len(), sent.len());
    }

    /// A second client while the first is connected is closed when the run next sends.
    #[test]
    fn a_stream_takes_one_client_at_a_time() {
        let listening = listener();
        let mut stream = Stream::listen(&listening).unwrap();
        let address = listening.local_addr().unwrap();
        let _first = TcpStream::connect(address).unwrap();
        stream.send(b"first");
        let mut second = TcpStream::connect(address).unwrap();
        stream.send(b"second");
        second
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let closed = match second.read(&mut [0]) {
            Ok(read) => read == 0,
            Err(error) => error.kind() == ErrorKind::ConnectionReset,
        };
        assert!(closed);
    }
}