use std::process::ExitStatus;
use tokio::process::{Child, Command};
pub struct Spec {
pub command: String,
pub args: Vec<String>,
pub extra_env: Vec<String>,
}
impl Spec {
pub fn display(&self) -> String {
std::iter::once(&self.command)
.chain(self.args.iter())
.map(|part| {
if part.is_empty() || part.contains(char::is_whitespace) {
format!("{part:?}")
} else {
part.clone()
}
})
.collect::<Vec<_>>()
.join(" ")
}
}
pub fn connect_host(bind: &str) -> &str {
match bind {
"0.0.0.0" => "127.0.0.1",
"::" | "[::]" => "::1",
other => other,
}
}
fn url(host: &str, port: u16, password: Option<&str>) -> String {
let host = if host.contains(':') && !host.starts_with('[') {
format!("[{host}]")
} else {
host.to_string()
};
match password {
Some(pw) => format!("redis://:{}@{host}:{port}", encode(pw)),
None => format!("redis://{host}:{port}"),
}
}
fn encode(s: &str) -> String {
let mut out = String::with_capacity(s.len());
for b in s.bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
out.push(b as char)
}
_ => out.push_str(&format!("%{b:02X}")),
}
}
out
}
pub fn spawn(spec: &Spec, host: &str, port: u16, password: Option<&str>) -> std::io::Result<Child> {
let url = url(host, port, password);
let mut cmd = Command::new(&spec.command);
cmd.args(&spec.args)
.env("REDIS_URL", &url)
.env("REDIS_HOST", host)
.env("REDIS_PORT", port.to_string());
for name in &spec.extra_env {
cmd.env(name, &url);
}
cmd.spawn()
}
pub async fn supervise(mut child: Child) -> i32 {
let pid = child.id();
let mut escalate = false;
#[cfg(unix)]
{
use tokio::signal::unix::{signal, SignalKind};
let mut sigterm = match signal(SignalKind::terminate()) {
Ok(s) => Some(s),
Err(e) => {
eprintln!("meebis: run: could not listen for SIGTERM: {e}");
None
}
};
loop {
tokio::select! {
status = child.wait() => return code(status),
_ = tokio::signal::ctrl_c() => {
stop(pid, libc::SIGINT, &mut escalate, &mut child)
}
_ = async {
match sigterm.as_mut() {
Some(s) => { s.recv().await; }
None => std::future::pending::<()>().await,
}
} => stop(pid, libc::SIGTERM, &mut escalate, &mut child),
}
}
}
#[cfg(not(unix))]
{
let _ = pid;
loop {
tokio::select! {
status = child.wait() => return code(status),
_ = tokio::signal::ctrl_c() => {
if escalate {
let _ = child.start_kill();
}
escalate = true;
}
}
}
}
}
#[cfg(unix)]
fn stop(pid: Option<u32>, sig: i32, escalate: &mut bool, child: &mut Child) {
if *escalate {
let _ = child.start_kill();
return;
}
*escalate = true;
if let Some(pid) = pid {
unsafe { libc::kill(pid as i32, sig) };
}
}
fn code(status: std::io::Result<ExitStatus>) -> i32 {
match status {
Ok(status) => match status.code() {
Some(code) => code,
None => {
#[cfg(unix)]
{
use std::os::unix::process::ExitStatusExt;
status.signal().map(|s| 128 + s).unwrap_or(1)
}
#[cfg(not(unix))]
{
1
}
}
},
Err(e) => {
eprintln!("meebis: run: could not wait for the command: {e}");
1
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn url_without_password() {
assert_eq!(url("127.0.0.1", 6400, None), "redis://127.0.0.1:6400");
}
#[test]
fn url_brackets_ipv6() {
assert_eq!(url("::1", 6379, None), "redis://[::1]:6379");
assert_eq!(url("[::1]", 6379, None), "redis://[::1]:6379");
}
#[test]
fn url_encodes_the_password() {
assert_eq!(
url("127.0.0.1", 6379, Some("p@ss:w/rd")),
"redis://:p%40ss%3Aw%2Frd@127.0.0.1:6379"
);
assert_eq!(
url("127.0.0.1", 6379, Some("aZ0-._~")),
"redis://:aZ0-._~@127.0.0.1:6379"
);
}
#[test]
fn wildcard_binds_become_loopback() {
assert_eq!(connect_host("0.0.0.0"), "127.0.0.1");
assert_eq!(connect_host("::"), "::1");
assert_eq!(connect_host("127.0.0.1"), "127.0.0.1");
assert_eq!(connect_host("192.168.1.5"), "192.168.1.5");
}
#[test]
fn display_quotes_only_what_needs_it() {
let spec = Spec {
command: "npm".into(),
args: vec!["test".into(), "a b".into()],
extra_env: vec![],
};
assert_eq!(spec.display(), "npm test \"a b\"");
}
}