liminal-cli 0.2.0

//! A sibling ending cannot discard the rest of an admitted batch.

use super::*;

struct PausedOutput {
    lines: Sender<Vec<u8>>,
    release: Receiver<()>,
    paused: bool,
}

impl Write for PausedOutput {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.lines.send(bytes.to_vec()).map_err(io::Error::other)?;
        let value: Value = serde_json::from_slice(bytes)?;
        if !self.paused && value["params"]["level"] == "info" {
            self.paused = true;
            self.release.recv().map_err(io::Error::other)?;
        }
        Ok(bytes.len())
    }

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

#[test]
fn net027_a_sibling_end_keeps_the_whole_batch_already_being_written() -> TestResult {
    let rail = fixture(&["orders.one", "events.one"]);
    let taps = rail.taps.clone();
    let (input_tx, input_rx) = mpsc::channel();
    let (output_tx, output_rx) = mpsc::channel();
    let (release_tx, release_rx) = mpsc::channel();
    std::thread::scope(|scope| -> TestResult {
        let server = scope.spawn(move || {
            Server::new(rail, 65536).serve(
                &mut Input {
                    lines: input_rx,
                    current: Cursor::new(Vec::new()),
                },
                &mut PausedOutput {
                    lines: output_tx,
                    release: release_rx,
                    paused: false,
                },
            )
        });
        let mut host = Host {
            input: input_tx,
            output: output_rx,
            taps: taps.clone(),
            next_id: 0,
        };
        host.request("initialize", json!({"protocolVersion":"2025-11-25"}))?;
        let _answer = host.line()?;
        host.subscribe()?;
        host.taps.feed(
            "console.tap.orders.one",
            Ok(batch(
                "orders.one",
                0,
                (0..8)
                    .map(|n| message(&format!("local:{n}"), "orders.one"))
                    .collect(),
            )),
        )?;
        let first = host.line()?;
        assert_eq!(first["params"]["data"]["tap"], "local:0");
        host.taps.feed(
            "console.tap.events.one",
            Err(RailError::Refused("service retired".to_owned())),
        )?;
        host.taps.wait_closed("console.tap.orders.one")?;
        release_tx.send(())?;
        host.request("ping", json!({}))?;
        let mut messages = vec!["local:0".to_owned()];
        let mut ended = 0;
        let mut answers = 0;
        while ended == 0 || answers == 0 {
            let line = host.line()?;
            if line.get("id").is_some() {
                answers += 1;
            } else if line["params"]["level"] == "info" {
                messages.push(
                    line["params"]["data"]["tap"]
                        .as_str()
                        .ok_or("tap id")?
                        .to_owned(),
                );
            } else {
                ended += 1;
            }
        }
        drop(host);
        server
            .join()
            .map_err(|_| io::Error::other("server panicked"))??;
        assert_eq!(
            messages,
            (0..8).map(|n| format!("local:{n}")).collect::<Vec<_>>()
        );
        assert_eq!(ended, 1);
        assert_eq!(taps.active()?, 0);
        Ok(())
    })
}

#[test]
fn net027_terminal_words_are_one_line() -> TestResult {
    run(&["orders.one"], 65536, |host| {
        host.subscribe()?;
        host.taps.feed(
            "console.tap.orders.one",
            Err(RailError::Refused("service\nretired".to_owned())),
        )?;
        assert_eq!(
            host.line()?["params"]["data"]["words"],
            "subscription to one ended: service retired"
        );
        Ok(())
    })
}