use std::net::SocketAddr;
use std::os::unix::process::CommandExt;
use std::path::{Path, PathBuf};
use std::process::{ExitStatus, Stdio};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, ChildStdout};
use tokio::sync::oneshot;
use crate::error::CliError;
pub(crate) const READY_DEADLINE: Duration = Duration::from_secs(30);
pub(crate) const TEARDOWN_DEADLINE: Duration = Duration::from_secs(15);
pub(crate) fn spawn_host(
root: &Path,
binary: &PathBuf,
arguments: &[&str],
) -> Result<Child, CliError> {
let mut command = std::process::Command::new(binary);
command.args(arguments);
command.current_dir(root);
command.process_group(0);
tokio::process::Command::from(command)
.stdout(Stdio::piped())
.spawn()
.map_err(|source| CliError::CommandSpawn {
command: binary.display().to_string(),
source,
})
}
pub(crate) async fn forward_and_detect_url(
stdout: ChildStdout,
url_tx: oneshot::Sender<SocketAddr>,
) {
let mut lines = BufReader::new(stdout).lines();
let mut url_tx = Some(url_tx);
loop {
match lines.next_line().await {
Ok(Some(line)) => {
println!("{line}");
if let Some(addr) = parse_serving_url(&line)
&& let Some(sender) = url_tx.take()
{
let _ = sender.send(addr);
}
}
Ok(None) => return,
Err(error) => {
eprintln!("frame: reading host output failed: {error}");
return;
}
}
}
}
pub(crate) fn parse_serving_url(line: &str) -> Option<SocketAddr> {
let after = line.split("http://").nth(1)?;
let token: String = after.chars().take_while(|c| !c.is_whitespace()).collect();
token.parse().ok()
}
pub(crate) enum BootOutcome {
Ready(SocketAddr),
Signaled,
}
pub(crate) async fn wait_until_ready(
child: &mut Child,
url_rx: oneshot::Receiver<SocketAddr>,
interrupt: &mut tokio::signal::unix::Signal,
terminate: &mut tokio::signal::unix::Signal,
) -> Result<BootOutcome, CliError> {
let deadline = tokio::time::sleep(READY_DEADLINE);
tokio::pin!(deadline);
let addr = tokio::select! {
announced = url_rx => match announced {
Ok(addr) => addr,
Err(_) => return Err(CliError::CommandFailed {
command: "the application host".to_owned(),
status: "exited before announcing its serving URL".to_owned(),
}),
},
status = child.wait() => {
let status = status.map_err(|source| CliError::CommandSpawn {
command: "the application host".to_owned(),
source,
})?;
return Err(CliError::CommandFailed {
command: "the application host".to_owned(),
status: status.to_string(),
});
}
_ = interrupt.recv() => return Ok(BootOutcome::Signaled),
_ = terminate.recv() => return Ok(BootOutcome::Signaled),
() = &mut deadline => return Err(CliError::NeverReady {
bind: "the page server (no URL announced)".to_owned(),
deadline: READY_DEADLINE,
}),
};
confirm_accepting(child, addr, interrupt, terminate).await
}
pub(crate) async fn confirm_accepting(
child: &mut Child,
addr: SocketAddr,
interrupt: &mut tokio::signal::unix::Signal,
terminate: &mut tokio::signal::unix::Signal,
) -> Result<BootOutcome, CliError> {
let deadline = tokio::time::Instant::now() + READY_DEADLINE;
loop {
if let Ok(Some(status)) = child.try_wait() {
return Err(CliError::CommandFailed {
command: "the application host".to_owned(),
status: status.to_string(),
});
}
if tokio::net::TcpStream::connect(addr).await.is_ok() {
return Ok(BootOutcome::Ready(addr));
}
if tokio::time::Instant::now() >= deadline {
return Err(CliError::NeverReady {
bind: addr.to_string(),
deadline: READY_DEADLINE,
});
}
tokio::select! {
() = tokio::time::sleep(Duration::from_millis(50)) => {}
_ = interrupt.recv() => return Ok(BootOutcome::Signaled),
_ = terminate.recv() => return Ok(BootOutcome::Signaled),
}
}
}
pub(crate) async fn stop_child(binary: &Path, mut child: Child) -> Result<(), CliError> {
eprintln!("stopping…");
if let Some(id) = child.id() {
send_sigterm(id)?;
}
match tokio::time::timeout(TEARDOWN_DEADLINE, child.wait()).await {
Ok(status) => {
let status = status.map_err(|source| CliError::CommandSpawn {
command: binary.display().to_string(),
source,
})?;
let verdict = exit_verdict(binary, status);
if verdict.is_ok() {
eprintln!("stopped cleanly");
}
verdict
}
Err(_elapsed) => {
child
.kill()
.await
.map_err(|source| CliError::CommandSpawn {
command: binary.display().to_string(),
source,
})?;
Err(CliError::TeardownTimeout {
deadline: TEARDOWN_DEADLINE,
})
}
}
}
fn send_sigterm(pid: u32) -> Result<(), CliError> {
let status = std::process::Command::new("kill")
.arg("-TERM")
.arg(pid.to_string())
.status()
.map_err(|source| CliError::CommandSpawn {
command: format!("kill -TERM {pid}"),
source,
})?;
if status.success() {
Ok(())
} else {
Err(CliError::CommandFailed {
command: format!("kill -TERM {pid}"),
status: status.to_string(),
})
}
}
pub(crate) fn exit_verdict(binary: &Path, status: ExitStatus) -> Result<(), CliError> {
if status.success() {
Ok(())
} else {
Err(CliError::CommandFailed {
command: binary.display().to_string(),
status: status.to_string(),
})
}
}
#[cfg(test)]
mod tests {
use super::parse_serving_url;
#[test]
fn parses_the_host_boot_line() {
assert_eq!(
parse_serving_url("demo_app serving at http://127.0.0.1:4191")
.map(|addr| addr.to_string()),
Some("127.0.0.1:4191".to_owned())
);
assert_eq!(
parse_serving_url("frame-host serving at http://127.0.0.1:6010")
.map(|addr| addr.to_string()),
Some("127.0.0.1:6010".to_owned())
);
}
#[test]
fn ignores_unrelated_lines() {
assert!(parse_serving_url("INFO booting embedded bus component").is_none());
assert!(parse_serving_url("no url here").is_none());
assert!(parse_serving_url("see http://example.com/docs").is_none());
}
}