128 lines
4.2 KiB
Rust
128 lines
4.2 KiB
Rust
//! A fake `brokerd` state socket for tests: it records each question and answers with a fixed
|
|
//! state, an error frame, or nothing at all (it holds the connection open without a byte). It
|
|
//! stops, and releases every held connection, when dropped; every wait has a deadline. Do not
|
|
//! edit.
|
|
|
|
#![allow(dead_code)] // each test file uses a different part of this module
|
|
|
|
use std::os::unix::net::{UnixListener, UnixStream};
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::thread::JoinHandle;
|
|
use std::time::{Duration, Instant};
|
|
|
|
use bxcore::{
|
|
Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionStateReply, WireError, read_frame,
|
|
write_frame,
|
|
};
|
|
|
|
/// How the fake answers.
|
|
#[derive(Debug, Clone)]
|
|
pub enum Answer {
|
|
State(SessionStateReply),
|
|
Error(String),
|
|
/// Read the question, then send nothing until the fake is dropped (at most 30 s).
|
|
Hang,
|
|
}
|
|
|
|
pub struct FakeBrokerd {
|
|
path: PathBuf,
|
|
stop: Arc<AtomicBool>,
|
|
seen: Arc<Mutex<Vec<Envelope>>>,
|
|
accept: Option<JoinHandle<()>>,
|
|
}
|
|
|
|
impl FakeBrokerd {
|
|
/// Listen on `path` (its directory is made) and answer every connection with `answer`.
|
|
pub fn start(path: &Path, answer: Answer) -> FakeBrokerd {
|
|
if let Some(dir) = path.parent() {
|
|
std::fs::create_dir_all(dir).unwrap();
|
|
}
|
|
let _ = std::fs::remove_file(path);
|
|
let listener = UnixListener::bind(path).unwrap();
|
|
let stop = Arc::new(AtomicBool::new(false));
|
|
let seen = Arc::new(Mutex::new(Vec::new()));
|
|
let (stop_accept, seen_accept) = (Arc::clone(&stop), Arc::clone(&seen));
|
|
let accept = std::thread::spawn(move || {
|
|
for stream in listener.incoming() {
|
|
if stop_accept.load(Ordering::SeqCst) {
|
|
return;
|
|
}
|
|
let Ok(stream) = stream else { return };
|
|
let (stop, seen, answer) = (
|
|
Arc::clone(&stop_accept),
|
|
Arc::clone(&seen_accept),
|
|
answer.clone(),
|
|
);
|
|
std::thread::spawn(move || serve(stream, &answer, &seen, &stop));
|
|
}
|
|
});
|
|
FakeBrokerd {
|
|
path: path.to_path_buf(),
|
|
stop,
|
|
seen,
|
|
accept: Some(accept),
|
|
}
|
|
}
|
|
|
|
/// Every question read so far, in order.
|
|
pub fn seen(&self) -> Vec<Envelope> {
|
|
self.seen.lock().unwrap().clone()
|
|
}
|
|
|
|
/// Waits up to 10 s until at least `n` questions were read; true if they were.
|
|
pub fn wait_for(&self, n: usize) -> bool {
|
|
let deadline = Instant::now() + Duration::from_secs(10);
|
|
while Instant::now() < deadline {
|
|
if self.seen().len() >= n {
|
|
return true;
|
|
}
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
false
|
|
}
|
|
}
|
|
|
|
fn serve(mut stream: UnixStream, answer: &Answer, seen: &Mutex<Vec<Envelope>>, stop: &AtomicBool) {
|
|
let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
|
|
let Ok(question) = read_frame(&mut stream) else {
|
|
return;
|
|
};
|
|
let id = question.id;
|
|
seen.lock().unwrap().push(question);
|
|
let msg = match answer {
|
|
Answer::State(state) => Message::SessionStateReply(state.clone()),
|
|
Answer::Error(detail) => Message::Error(WireError {
|
|
code: ErrorCode::Internal,
|
|
detail: detail.clone(),
|
|
}),
|
|
Answer::Hang => {
|
|
let deadline = Instant::now() + Duration::from_secs(30);
|
|
while !stop.load(Ordering::SeqCst) && Instant::now() < deadline {
|
|
std::thread::sleep(Duration::from_millis(10));
|
|
}
|
|
return;
|
|
}
|
|
};
|
|
let env = Envelope {
|
|
v: PROTOCOL_VERSION,
|
|
id,
|
|
r#final: true,
|
|
msg,
|
|
};
|
|
let _ = write_frame(&mut stream, &env);
|
|
}
|
|
|
|
impl Drop for FakeBrokerd {
|
|
fn drop(&mut self) {
|
|
self.stop.store(true, Ordering::SeqCst);
|
|
// Wakes the accept loop, which then sees `stop`.
|
|
let _ = UnixStream::connect(&self.path);
|
|
if let Some(accept) = self.accept.take() {
|
|
let _ = accept.join();
|
|
}
|
|
let _ = std::fs::remove_file(&self.path);
|
|
}
|
|
}
|