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 enum Endpoint {
Tcp { host: String, port: u16 },
Unix { path: std::path::PathBuf },
}
pub fn connect_host(bind: &str) -> &str {
match bind {
"0.0.0.0" => "127.0.0.1",
"::" | "[::]" => "::1",
other => other,
}
}
impl Endpoint {
fn url(&self, password: Option<&str>) -> String {
let auth = match password {
Some(pw) => format!(":{}@", encode(pw)),
None => String::new(),
};
match self {
Endpoint::Tcp { host, port } => {
let host = if host.contains(':') && !host.starts_with('[') {
format!("[{host}]")
} else {
host.clone()
};
format!("redis://{auth}{host}:{port}")
}
Endpoint::Unix { path } => format!("unix://{auth}{}", path.display()),
}
}
}
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, endpoint: &Endpoint, password: Option<&str>) -> std::io::Result<Child> {
let url = endpoint.url(password);
let mut cmd = Command::new(&spec.command);
cmd.args(&spec.args).env("REDIS_URL", &url);
match endpoint {
Endpoint::Tcp { host, port } => {
cmd.env("REDIS_HOST", host)
.env("REDIS_PORT", port.to_string());
}
Endpoint::Unix { path } => {
cmd.env("REDIS_SOCKET", path)
.env_remove("REDIS_HOST")
.env_remove("REDIS_PORT");
}
}
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::*;
fn tcp(host: &str, port: u16) -> Endpoint {
Endpoint::Tcp {
host: host.into(),
port,
}
}
fn unix(path: &str) -> Endpoint {
Endpoint::Unix { path: path.into() }
}
#[test]
fn url_without_password() {
assert_eq!(tcp("127.0.0.1", 6400).url(None), "redis://127.0.0.1:6400");
}
#[test]
fn url_brackets_ipv6() {
assert_eq!(tcp("::1", 6379).url(None), "redis://[::1]:6379");
assert_eq!(tcp("[::1]", 6379).url(None), "redis://[::1]:6379");
}
#[test]
fn url_encodes_the_password() {
assert_eq!(
tcp("127.0.0.1", 6379).url(Some("p@ss:w/rd")),
"redis://:p%40ss%3Aw%2Frd@127.0.0.1:6379"
);
assert_eq!(
tcp("127.0.0.1", 6379).url(Some("aZ0-._~")),
"redis://:aZ0-._~@127.0.0.1:6379"
);
}
#[test]
fn url_for_a_socket_is_the_path() {
assert_eq!(
unix("/tmp/w/redis.sock").url(None),
"unix:///tmp/w/redis.sock"
);
assert_eq!(
unix("/tmp/w/redis.sock").url(Some("hunter2")),
"unix://:hunter2@/tmp/w/redis.sock"
);
}
#[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\"");
}
}