use super::{Live, Rail};
use crate::config::Config;
use liminal::protocol::{
CausalContext, Frame, MessageEnvelope, ProtocolError, ProtocolVersion, SchemaId, decode,
encode, encoded_len,
};
use serde_json::json;
use std::io::{self, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc;
type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
fn write(socket: &mut TcpStream, frame: &Frame) -> TestResult {
let mut bytes = vec![0; encoded_len(frame)?];
let written = encode(frame, &mut bytes)?;
socket.write_all(bytes.get(..written).ok_or("invalid encoded frame length")?)?;
socket.flush()?;
Ok(())
}
fn read(
socket: &mut TcpStream,
bytes: &mut Vec<u8>,
) -> Result<Frame, Box<dyn std::error::Error + Send + Sync>> {
loop {
match decode(bytes) {
Ok((frame, used)) => {
bytes.drain(..used);
return Ok(frame);
}
Err(
ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
) => {
let mut buffer = [0; 4096];
let n = socket.read(&mut buffer)?;
if n == 0 {
return Err("peer closed before a frame".into());
}
bytes.extend_from_slice(&buffer[..n]);
}
Err(error) => return Err(error.into()),
}
}
}
fn live(address: &str) -> Result<Live, Box<dyn std::error::Error + Send + Sync>> {
let root = tempfile::tempdir()?;
std::fs::write(root.path().join("pass"), "fixture-tap-pass")?;
let seat = root.path().join("seat.toml");
std::fs::write(
&seat,
format!(
"server = '{address}'\nchannel = 'orders'\ntoken_file = 'pass'\n[limits]\nframe_bytes = 65536\nrecord_bytes = 1048576\nreply_timeout_ms = 10000\nflow_steps = 8\nrequest_bytes = 65536\n"
),
)?;
Ok(Live::new(&Config::load(&seat)?)?)
}
fn handshake(socket: &mut TcpStream, bytes: &mut Vec<u8>) -> TestResult {
let Frame::Connect { auth_token, .. } = read(socket, bytes)? else {
return Err("expected Connect".into());
};
if auth_token != liminal_sdk::lys_pass_credential("fixture-tap-pass") {
return Err("the connection did not present the seat credential".into());
}
write(
socket,
&Frame::ConnectAck {
flags: 0,
selected_version: ProtocolVersion::new(1, 0),
capabilities: 0,
},
)?;
let Frame::Subscribe { channel, .. } = read(socket, bytes)? else {
return Err("expected Subscribe".into());
};
assert_eq!(channel, "console.tap.orders.one");
Ok(())
}
fn acknowledge(socket: &mut TcpStream) -> TestResult {
write(
socket,
&Frame::SubscribeAck {
flags: 0,
stream_id: 1,
subscription_id: 41,
selected_schema: SchemaId::new([7; SchemaId::WIRE_LEN]),
},
)
}
fn join(thread: std::thread::ScopedJoinHandle<'_, TestResult>) -> TestResult {
thread
.join()
.map_err(|_| io::Error::other("tap wire fixture panicked"))?
}
#[test]
fn net027_live_tap_uses_the_seat_pass_and_receives_pushed_records() -> TestResult {
let listener = TcpListener::bind("127.0.0.1:0")?;
let live = live(&listener.local_addr()?.to_string())?;
let batch = json!({"v":1,"type":"tap.batch","target":"orders.one","seq":1,"sampled":1,"dropped":0,"messages":[]});
let expected = batch.clone();
std::thread::scope(|scope| -> TestResult {
let server = scope.spawn(move || -> TestResult {
let (mut socket, _) = listener.accept()?;
handshake(&mut socket, &mut Vec::new())?;
acknowledge(&mut socket)?;
let envelope = MessageEnvelope::new(
SchemaId::new([7; SchemaId::WIRE_LEN]),
CausalContext::independent(),
serde_json::to_vec(&batch)?,
);
write(&mut socket, &Frame::new_deliver(1, 1, envelope)?)?;
let mut bytes = [0; 4096];
while socket.read(&mut bytes)? != 0 {}
Ok(())
});
let tap = live.tap("console.tap.orders.one")?;
assert_eq!(tap.reader.receive()?, expected);
tap.closer.close()?;
drop(tap);
join(server)
})
}
#[test]
fn net027_live_tap_refusal_preserves_the_server_words() -> TestResult {
let listener = TcpListener::bind("127.0.0.1:0")?;
let live = live(&listener.local_addr()?.to_string())?;
std::thread::scope(|scope| -> TestResult {
let server = scope.spawn(move || -> TestResult {
let (mut socket, _) = listener.accept()?;
handshake(&mut socket, &mut Vec::new())?;
write(
&mut socket,
&Frame::SubscribeError {
flags: 0,
stream_id: 1,
reason_code: 1,
message: Some("missing_grant: orders.one".to_owned()),
},
)
});
let refused = live.tap("console.tap.orders.one");
assert!(refused.is_err());
let words = refused.err().ok_or("missing tap refusal")?.to_string();
assert!(words.contains("missing_grant: orders.one"), "{words}");
join(server)
})
}
#[test]
fn net027_live_close_signals_a_receive_with_no_records() -> TestResult {
let listener = TcpListener::bind("127.0.0.1:0")?;
let live = live(&listener.local_addr()?.to_string())?;
std::thread::scope(|scope| -> TestResult {
let server = scope.spawn(move || -> TestResult {
let (mut socket, _) = listener.accept()?;
handshake(&mut socket, &mut Vec::new())?;
acknowledge(&mut socket)?;
let mut bytes = [0; 4096];
while socket.read(&mut bytes)? != 0 {}
Ok(())
});
let tap = live.tap("console.tap.orders.one")?;
let closer = std::sync::Arc::clone(&tap.closer);
let (ready_tx, ready_rx) = mpsc::sync_channel(0);
let receiver = scope.spawn(move || -> TestResult {
ready_tx.send(())?;
let end = tap.reader.receive();
assert!(end.is_err());
assert!(
end.err()
.ok_or("missing local end")?
.to_string()
.contains("closed")
);
Ok(())
});
ready_rx.recv()?;
closer.close()?;
join(receiver)?;
join(server)
})
}
#[test]
fn net027_live_server_end_preserves_the_terminal_reason() -> TestResult {
let listener = TcpListener::bind("127.0.0.1:0")?;
let live = live(&listener.local_addr()?.to_string())?;
std::thread::scope(|scope| -> TestResult {
let server = scope.spawn(move || -> TestResult {
let (mut socket, _) = listener.accept()?;
handshake(&mut socket, &mut Vec::new())?;
acknowledge(&mut socket)?;
write(
&mut socket,
&Frame::SubscribeError {
flags: 0,
stream_id: 1,
reason_code: 1,
message: Some("service retired".to_owned()),
},
)
});
let tap = live.tap("console.tap.orders.one")?;
let end = tap.reader.receive();
assert!(end.is_err());
assert!(
end.err()
.ok_or("missing server end")?
.to_string()
.contains("service retired")
);
drop(tap);
join(server)
})
}