use std::io::{self, BufRead, BufReader, Write};
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
use std::time::{Duration, Instant};
pub struct RpcClient {
child: Child,
stdin: ChildStdin,
messages: Receiver<io::Result<serde_json::Value>>,
next_id: i64,
pub timeout: Duration,
}
impl RpcClient {
pub fn spawn(program: &str, args: &[&str], cwd: &std::path::Path) -> io::Result<RpcClient> {
let mut child = Command::new(program)
.args(args)
.current_dir(cwd)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()?;
let stdin = child
.stdin
.take()
.ok_or_else(|| io::Error::other("no stdin"))?;
let stdout = child
.stdout
.take()
.ok_or_else(|| io::Error::other("no stdout"))?;
let (sender, messages) = mpsc::channel();
std::thread::Builder::new()
.name(format!("radar-rpc-reader-{program}"))
.spawn(move || read_messages(stdout, sender))?;
Ok(RpcClient {
child,
stdin,
messages,
next_id: 1,
timeout: Duration::from_secs(30),
})
}
fn send(&mut self, payload: &serde_json::Value) -> io::Result<()> {
let body = serde_json::to_vec(payload)?;
write!(self.stdin, "Content-Length: {}\r\n\r\n", body.len())?;
self.stdin.write_all(&body)?;
self.stdin.flush()
}
pub fn notify(&mut self, method: &str, params: serde_json::Value) -> io::Result<()> {
self.send(&serde_json::json!({
"jsonrpc": "2.0", "method": method, "params": params
}))
}
pub fn request(
&mut self,
method: &str,
params: serde_json::Value,
) -> io::Result<serde_json::Value> {
let id = self.next_id;
self.next_id += 1;
self.send(&serde_json::json!({
"jsonrpc": "2.0", "id": id, "method": method, "params": params
}))?;
let deadline = Instant::now() + self.timeout;
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
format!("{method} timed out"),
));
}
let msg = match self.messages.recv_timeout(remaining) {
Ok(message) => message?,
Err(RecvTimeoutError::Timeout) => {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
format!("{method} timed out"),
));
}
Err(RecvTimeoutError::Disconnected) => {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"server closed",
));
}
};
if msg.get("id").and_then(|v| v.as_i64()) == Some(id) {
if let Some(err) = msg.get("error") {
return Err(io::Error::other(format!("{method}: {err}")));
}
return Ok(msg
.get("result")
.cloned()
.unwrap_or(serde_json::Value::Null));
}
if msg.get("method").is_some() && msg.get("id").is_some() {
let sid = msg["id"].clone();
let _ = self.send(&serde_json::json!({
"jsonrpc": "2.0", "id": sid, "result": serde_json::Value::Null
}));
}
}
}
pub fn shutdown(mut self) {
if self.request("shutdown", serde_json::Value::Null).is_ok() {
let _ = self.notify("exit", serde_json::Value::Null);
let deadline = Instant::now() + Duration::from_secs(1);
while Instant::now() < deadline {
match self.child.try_wait() {
Ok(Some(_)) => return,
Ok(None) => std::thread::sleep(Duration::from_millis(10)),
Err(_) => break,
}
}
}
let _ = self.child.kill();
let _ = self.child.wait();
}
}
fn read_messages(stdout: ChildStdout, sender: mpsc::Sender<io::Result<serde_json::Value>>) {
let mut reader = BufReader::new(stdout);
loop {
let message = read_message(&mut reader);
let terminal = message.is_err();
if sender.send(message).is_err() || terminal {
break;
}
}
}
fn read_message(reader: &mut impl BufRead) -> io::Result<serde_json::Value> {
let mut content_length: Option<usize> = None;
loop {
let mut line = String::new();
if reader.read_line(&mut line)? == 0 {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"server closed",
));
}
let line = line.trim_end();
if line.is_empty() {
break;
}
if let Some(value) = line.strip_prefix("Content-Length:") {
content_length = value.trim().parse().ok();
}
}
let len = content_length
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing Content-Length"))?;
if len > 16 * 1024 * 1024 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"JSON-RPC message exceeds 16 MiB",
));
}
let mut buffer = vec![0u8; len];
reader.read_exact(&mut buffer)?;
serde_json::from_slice(&buffer)
.map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))
}
impl Drop for RpcClient {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn oversized_message_is_rejected_before_allocation() {
let mut framed = std::io::Cursor::new(b"Content-Length: 16777217\r\n\r\n");
let error = read_message(&mut framed).expect_err("oversized frame must fail");
assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
}
#[cfg(unix)]
#[test]
fn framing_round_trip_with_echo_server() {
let script = r#"
import json, sys
def send(obj):
body = json.dumps(obj).encode()
sys.stdout.buffer.write(b"Content-Length: %d\r\n\r\n" % len(body))
sys.stdout.buffer.write(body)
sys.stdout.buffer.flush()
while True:
line = sys.stdin.buffer.readline()
if not line:
break
n = int(line.split(b":")[1])
sys.stdin.buffer.readline()
msg = json.loads(sys.stdin.buffer.read(n))
if msg.get("method") == "exit":
break
if "id" in msg:
send({"jsonrpc": "2.0", "method": "noise/notification", "params": {}})
send({"jsonrpc": "2.0", "id": msg["id"], "result": {"echo": msg["method"]}})
"#;
let dir = std::env::temp_dir();
let path = dir.join(format!("radar-rpc-echo-{}.py", std::process::id()));
std::fs::write(&path, script).expect("write echo server");
let mut client =
RpcClient::spawn("python3", &[path.to_str().expect("path")], &dir).expect("spawn");
client.timeout = Duration::from_secs(10);
let res = client
.request("gean/ping", serde_json::json!({}))
.expect("request");
assert_eq!(res["echo"], "gean/ping");
let res2 = client
.request("shutdown", serde_json::Value::Null)
.expect("second");
assert_eq!(res2["echo"], "shutdown");
client.shutdown();
let _ = std::fs::remove_file(&path);
}
#[cfg(unix)]
#[test]
fn request_deadline_interrupts_a_silent_peer() {
let script = r#"
import sys, time
line = sys.stdin.buffer.readline()
if line:
n = int(line.split(b":")[1])
sys.stdin.buffer.readline()
sys.stdin.buffer.read(n)
time.sleep(5)
"#;
let dir = std::env::temp_dir();
let path = dir.join(format!("radar-rpc-silent-{}.py", std::process::id()));
std::fs::write(&path, script).expect("write silent server");
let mut client =
RpcClient::spawn("python3", &[path.to_str().expect("path")], &dir).expect("spawn");
client.timeout = Duration::from_millis(50);
let started = Instant::now();
let error = client
.request("gean/timeout", serde_json::json!({}))
.expect_err("silent peer must time out");
assert_eq!(error.kind(), io::ErrorKind::TimedOut);
assert!(
started.elapsed() < Duration::from_secs(1),
"deadline did not interrupt the blocking pipe read"
);
drop(client);
let _ = std::fs::remove_file(&path);
}
}