use std::io::{self, BufRead, Cursor, Read, Write};
use std::sync::mpsc::{self, Receiver, Sender};
use std::time::Duration;
use serde_json::{Value, json};
use super::notification_tests::fixture;
use super::tools::Server;
use crate::network::RailError;
use crate::network::fake_taps::Harness;
type TestResult = Result<(), Box<dyn std::error::Error>>;
struct Input {
lines: Receiver<Vec<u8>>,
current: Cursor<Vec<u8>>,
}
impl Read for Input {
fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
let buffer = self.fill_buf()?;
let n = buffer.len().min(bytes.len());
bytes[..n].copy_from_slice(&buffer[..n]);
self.consume(n);
Ok(n)
}
}
impl BufRead for Input {
fn fill_buf(&mut self) -> io::Result<&[u8]> {
if self.current.position() == self.current.get_ref().len() as u64 {
self.current = Cursor::new(self.lines.recv().unwrap_or_default());
}
self.current.fill_buf()
}
fn consume(&mut self, n: usize) {
self.current.consume(n);
}
}
struct Output(Sender<Vec<u8>>);
impl Write for Output {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.0.send(bytes.to_vec()).map_err(io::Error::other)?;
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
struct Host {
input: Sender<Vec<u8>>,
output: Receiver<Vec<u8>>,
taps: Harness,
next_id: u64,
}
impl Host {
fn request(&mut self, method: &str, params: Value) -> TestResult {
self.next_id += 1;
let mut request = json!({"jsonrpc":"2.0", "id":self.next_id, "method":method});
request["params"] = params;
let mut bytes = serde_json::to_vec(&request)?;
bytes.push(b'\n');
self.input.send(bytes)?;
Ok(())
}
fn line(&self) -> Result<Value, Box<dyn std::error::Error>> {
let bytes = self.output.recv_timeout(Duration::from_secs(10))?;
assert_eq!(bytes.last(), Some(&b'\n'));
assert!(!bytes[..bytes.len() - 1].contains(&b'\n'));
Ok(serde_json::from_slice(&bytes)?)
}
fn subscribe(&mut self) -> TestResult {
self.request(
"tools/call",
json!({"name":"subscribe", "arguments":{"service":"one"}}),
)?;
let answer = self.line()?;
assert_eq!(answer["result"]["isError"], false, "{answer}");
Ok(())
}
}
fn run(
subjects: &[&str],
frame_bytes: usize,
body: impl FnOnce(&mut Host) -> TestResult,
) -> TestResult {
let rail = fixture(subjects);
let taps = rail.taps.clone();
let (input_tx, input_rx) = mpsc::channel();
let (output_tx, output_rx) = mpsc::channel();
std::thread::scope(|scope| -> TestResult {
let server = scope.spawn(move || {
let mut server = Server::new(rail, frame_bytes);
server.serve(
&mut Input {
lines: input_rx,
current: Cursor::new(Vec::new()),
},
&mut Output(output_tx),
)
});
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()?;
let outcome = body(&mut host);
drop(host);
server
.join()
.map_err(|_| io::Error::other("host server panicked"))??;
assert_eq!(taps.active()?, 0, "host close left a subscription running");
outcome
})
}
fn message(id: &str, subject: &str) -> Value {
json!({"id":id,"subject":subject,"size":12,"outcome":"delivered","t":1_791_272_000_123_u64,"from":"7","from_station":"orders.worker","hops":[{"at_ms":2,"node":"local"}]})
}
fn batch(subject: &str, dropped: u64, messages: Vec<Value>) -> Value {
let mut batch =
json!({"v":1,"type":"tap.batch","target":subject,"seq":1,"sampled":1,"dropped":dropped});
batch["messages"] = Value::Array(messages);
batch
}
#[test]
fn net027_batches_push_ordered_message_lines_and_name_dropped_records() -> TestResult {
run(&["orders.one"], 65536, |host| {
host.subscribe()?;
let mut second = message("local:42", "orders.one");
second
.as_object_mut()
.ok_or("message object")?
.remove("from_station");
host.taps.feed(
"console.tap.orders.one",
Ok(batch(
"orders.one",
2,
vec![message("local:41", "orders.one"), second],
)),
)?;
let first = host.line()?;
assert_eq!(
first,
json!({"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"liminal.one","data":{"device":"fixture:7000","service":"one","subject":"orders.one","from":"orders.worker","outcome":"delivered","at_ms":1_791_272_000_123_u64,"words":"orders.one: delivered, 12 B from orders.worker","tap":"local:41"}}})
);
let second = host.line()?;
assert_eq!(second["params"]["data"]["tap"], "local:42");
assert_eq!(second["params"]["data"]["from"], "connection 7");
let gap = host.line()?;
assert_eq!(gap["params"]["level"], "warning");
assert_eq!(
gap["params"]["data"]["words"],
"2 messages on orders.one were not kept"
);
assert_eq!(gap["params"]["data"]["tap"], Value::Null);
Ok(())
})
}
#[test]
fn net027_low_logging_levels_keep_info() -> TestResult {
for level in ["debug", "info", "notice"] {
run(&["orders.one"], 65536, |host| {
host.subscribe()?;
host.request("logging/setLevel", json!({"level":level}))?;
assert!(host.line()?.get("result").is_some());
host.taps.feed(
"console.tap.orders.one",
Ok(batch(
"orders.one",
0,
vec![message("local:41", "orders.one")],
)),
)?;
assert_eq!(host.line()?["params"]["level"], "info");
Ok(())
})?;
}
Ok(())
}
#[test]
fn net027_high_logging_levels_keep_only_gap_warnings() -> TestResult {
for level in ["warning", "error", "critical", "alert", "emergency"] {
run(&["orders.one"], 65536, |host| {
host.subscribe()?;
host.request("logging/setLevel", json!({"level":level}))?;
assert!(host.line()?.get("result").is_some());
host.taps.feed(
"console.tap.orders.one",
Ok(batch(
"orders.one",
1,
vec![message("local:41", "orders.one")],
)),
)?;
assert_eq!(host.line()?["params"]["data"]["outcome"], "dropped");
host.request("ping", json!({}))?;
assert_eq!(host.line()?["id"], host.next_id);
Ok(())
})?;
}
Ok(())
}
#[test]
fn net027_a_server_end_warns_once_closes_siblings_and_allows_resubscribe() -> TestResult {
run(&["orders.one", "events.one"], 65536, |host| {
host.subscribe()?;
host.taps.feed(
"console.tap.orders.one",
Err(RailError::Refused("service retired".to_owned())),
)?;
let ended = host.line()?;
assert_eq!(ended["params"]["level"], "warning");
assert_eq!(
ended["params"]["data"]["words"],
"subscription to one ended: service retired"
);
host.subscribe()?;
assert_eq!(host.taps.opened()?.len(), 4);
host.taps.feed(
"console.tap.events.one",
Ok(batch(
"events.one",
0,
vec![message("local:43", "events.one")],
)),
)?;
assert_eq!(host.line()?["params"]["data"]["tap"], "local:43");
Ok(())
})
}
#[test]
fn net027_host_close_stops_blocked_subscription_threads() -> TestResult {
run(&["orders.one", "events.one"], 65536, |host| {
host.subscribe()?;
assert_eq!(host.taps.active()?, 2);
Ok(())
})
}
#[test]
fn net027_malformed_tap_records_end_in_words() -> TestResult {
for record in [
json!({"type":"topology"}),
batch("wrong", 0, vec![]),
batch("orders.one", 0, vec![json!({"subject":"orders.one"})]),
] {
run(&["orders.one"], 65536, |host| {
host.subscribe()?;
host.taps.feed("console.tap.orders.one", Ok(record))?;
let ended = host.line()?;
assert_eq!(ended["params"]["level"], "warning");
assert!(
ended["params"]["data"]["words"]
.as_str()
.is_some_and(|words| words.starts_with("subscription to one ended: "))
);
Ok(())
})?;
}
Ok(())
}
#[test]
fn net027_frame_bound_cuts_only_words_at_unicode_boundaries() -> TestResult {
let subject = "é".repeat(60);
run(&[&subject], 512, |host| {
host.subscribe()?;
host.taps.feed(
&format!("console.tap.{subject}"),
Ok(batch(&subject, 0, vec![message("local:41", &subject)])),
)?;
let line = host.line()?;
assert!(serde_json::to_vec(&line)?.len() < 512);
assert_eq!(line["params"]["data"]["subject"], subject);
assert_eq!(line["params"]["data"]["tap"], "local:41");
assert!(
line["params"]["data"]["words"]
.as_str()
.is_some_and(|words| words.ends_with('…'))
);
Ok(())
})
}
#[test]
fn net027_answers_and_concurrent_subjects_are_whole_lines() -> TestResult {
run(&["orders.one", "events.one"], 65536, |host| {
host.subscribe()?;
std::thread::scope(|scope| -> TestResult {
let mut feeds = Vec::new();
for subject in ["orders.one", "events.one"] {
let taps = host.taps.clone();
feeds.push(scope.spawn(move || {
taps.feed(
&format!("console.tap.{subject}"),
Ok(batch(
subject,
0,
(0..8)
.map(|n| message(&format!("{subject}:{n}"), subject))
.collect(),
)),
)
}));
}
host.request("ping", json!({}))?;
let mut ids = std::collections::BTreeMap::<String, Vec<String>>::new();
let mut answers = 0;
for _ in 0..17 {
let line = host.line()?;
if line.get("id").is_some() {
answers += 1;
} else {
let data = &line["params"]["data"];
ids.entry(data["subject"].as_str().ok_or("subject")?.to_owned())
.or_default()
.push(data["tap"].as_str().ok_or("tap")?.to_owned());
}
}
assert_eq!(answers, 1);
for subject in ["orders.one", "events.one"] {
assert_eq!(
ids.get(subject),
Some(&(0..8).map(|n| format!("{subject}:{n}")).collect())
);
}
for feed in feeds {
feed.join()
.map_err(|_| io::Error::other("feeder panicked"))??;
}
Ok(())
})
})
}
#[test]
fn net027_words_are_one_readable_line_without_changing_fields() -> TestResult {
run(&["orders.one"], 65536, |host| {
host.subscribe()?;
let mut record = message("local:41", "orders.one");
record["from_station"] = json!("orders\r\nworker");
host.taps.feed(
"console.tap.orders.one",
Ok(batch("orders.one", 0, vec![record])),
)?;
let line = host.line()?;
assert_eq!(line["params"]["data"]["from"], "orders\r\nworker");
assert_eq!(
line["params"]["data"]["words"],
"orders.one: delivered, 12 B from orders worker"
);
Ok(())
})
}
#[path = "notification_batch_tests.rs"]
mod batches;