kyleandClaude Opus 5.5
d7009dc488
The test client set its read timeout before every frame. On macOS that
fails with EINVAL once the handler has sent its final frame and closed,
so every admin test that reads a second frame failed there (40 runs of
40 at 2408e2c). The frame is already buffered, so the client now takes
that one refusal as the peer having closed and reads it. The plan's copy
changes with it.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
80 lines
2.5 KiB
Rust
80 lines
2.5 KiB
Rust
//! A client for `broker::handle` and `admin::handle` over a socket pair, and a `Broker` built on
|
|
//! a `Rig`. Do not edit.
|
|
//!
|
|
//! Included with `#[path = "support/client.rs"] mod client;`, beside `rig`, `runtime`, `sink`
|
|
//! and `tmp`.
|
|
|
|
#![allow(dead_code)] // each test file uses a different part of this module
|
|
|
|
use std::os::unix::net::UnixStream;
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use brokerd::broker::{self, Broker};
|
|
use proto::{Envelope, Message, PROTOCOL_VERSION, ToolRequest};
|
|
|
|
use crate::rig::Rig;
|
|
use crate::runtime::{Recording, Shared};
|
|
|
|
pub trait Serve {
|
|
/// A `Broker` on this rig's home, with `runtime` and the rig's flaky sink and log.
|
|
fn broker(&self, runtime: &Arc<Recording>) -> Arc<Broker>;
|
|
}
|
|
|
|
impl Serve for Rig {
|
|
fn broker(&self, runtime: &Arc<Recording>) -> Arc<Broker> {
|
|
let runtime = Box::new(Shared(Arc::clone(runtime)));
|
|
Arc::new(Broker::new(
|
|
self.cfg.clone(),
|
|
self.ledger(),
|
|
runtime,
|
|
self.lines.sink(),
|
|
))
|
|
}
|
|
}
|
|
|
|
/// A connection to `handler` running on its own thread, with `msg` already sent under `id`.
|
|
pub fn open(
|
|
broker: &Arc<Broker>,
|
|
handler: fn(UnixStream, &Broker),
|
|
id: u64,
|
|
msg: Message,
|
|
) -> UnixStream {
|
|
let (mut client, server) = UnixStream::pair().unwrap();
|
|
let broker = Arc::clone(broker);
|
|
std::thread::spawn(move || handler(server, &broker));
|
|
let env = Envelope {
|
|
v: PROTOCOL_VERSION,
|
|
id,
|
|
r#final: true,
|
|
msg,
|
|
};
|
|
proto::write_frame(&mut client, &env).unwrap();
|
|
client
|
|
}
|
|
|
|
/// The next frame, waiting at most ten seconds. Once the handler has closed its end, macOS
|
|
/// refuses the timeout with EINVAL (22); the frame is buffered by then and the read cannot block.
|
|
pub fn next(stream: &mut UnixStream) -> Envelope {
|
|
if let Err(e) = stream.set_read_timeout(Some(Duration::from_secs(10))) {
|
|
let peer_closed = cfg!(target_vendor = "apple") && e.raw_os_error() == Some(22);
|
|
assert!(peer_closed, "set_read_timeout: {e}");
|
|
}
|
|
proto::read_frame(stream).unwrap()
|
|
}
|
|
|
|
/// Sends a tool request to `broker::handle` and reads every frame up to the final one.
|
|
pub fn call(broker: &Arc<Broker>, req: ToolRequest) -> Vec<Envelope> {
|
|
let id = req.call.0;
|
|
let mut stream = open(broker, broker::handle, id, Message::ToolRequest(req));
|
|
let mut frames = Vec::new();
|
|
loop {
|
|
let env = next(&mut stream);
|
|
let last = env.r#final;
|
|
frames.push(env);
|
|
if last {
|
|
return frames;
|
|
}
|
|
}
|
|
}
|