Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/swarm-notify-router/src/test_nats.rs

77 lines
2.7 KiB
Rust

//! A throwaway `nats-server` with `JetStream`, for the tests whose subject is
//! the server's own behaviour: what a stream keeps and what a request gets.
//!
//! The same shape as `swarm-queue-client`'s: the binary comes from
//! [`NATS_SERVER_ENV`], which `nix/checks.nix` and `nix/devshell.nix` export,
//! and a test that needs it panics when it is unset rather than skipping.
use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::time::Duration;
/// The variable naming the `nats-server` binary.
const NATS_SERVER_ENV: &str = "HIVE_NATS_SERVER";
/// The subdirectory holding nothing but the server's ports file.
const PORTS_DIR: &str = "ports";
/// A running server on a loopback port it bound itself. Dropping it kills
/// the server and removes its directory.
pub struct TestNats {
child: Child,
dir: tempfile::TempDir,
}
impl TestNats {
pub fn start() -> Self {
let bin = std::env::var(NATS_SERVER_ENV).unwrap_or_else(|_| {
panic!("{NATS_SERVER_ENV} is exported by nix/checks.nix and nix/devshell.nix")
});
let dir = tempfile::Builder::new()
.prefix("test-nats-")
.tempdir()
.expect("a fresh directory for the server");
std::fs::create_dir(dir.path().join(PORTS_DIR)).expect("the ports file directory");
let child = Command::new(&bin)
.args(["-a", "127.0.0.1", "-p", "-1", "-js", "-sd"])
.arg(dir.path().join("store"))
.arg("--ports_file_dir")
.arg(dir.path().join(PORTS_DIR))
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.unwrap_or_else(|e| panic!("starting {bin}: {e}"));
Self { child, dir }
}
/// A client connected to the server, once it accepts connections.
pub async fn client(&self) -> async_nats::Client {
for _ in 0..100 {
if let Some(url) = announced_url(&self.dir.path().join(PORTS_DIR))
&& let Ok(client) = async_nats::connect(&url).await
{
return client;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
panic!(
"nats-server in {} never accepted a connection",
self.dir.path().display()
);
}
}
/// The client URL in the ports file the server writes once it listens.
fn announced_url(dir: &Path) -> Option<String> {
let ports = std::fs::read_dir(dir).ok()?.next()?.ok()?;
let ports: serde_json::Value =
serde_json::from_slice(&std::fs::read(ports.path()).ok()?).ok()?;
ports["nats"][0].as_str().map(str::to_owned)
}
impl Drop for TestNats {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}