use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::mpsc::{Sender, channel};
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail};
use crate::progress::{self, Step, human_duration};
use crate::registry::{self, Entry, Work};
const TAIL_LINES: usize = 40;
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Mode {
Quiet,
UntilReady,
Stream,
Log,
}
pub const MODE_ENV: &str = "TURNOUT_CONSOLE";
impl Mode {
pub fn resolve(wanted: Mode, verbose: bool) -> Mode {
if let Some(forced) = Self::forced() {
return forced;
}
if verbose || !is_terminal() { Mode::Stream } else { wanted }
}
fn forced() -> Option<Mode> {
match std::env::var(MODE_ENV).ok()?.trim().to_ascii_lowercase().as_str() {
"quiet" => Some(Mode::Quiet),
"ready" => Some(Mode::UntilReady),
"stream" => Some(Mode::Stream),
_ => None,
}
}
}
fn is_terminal() -> bool {
use std::io::IsTerminal;
std::io::stdout().is_terminal()
}
struct Line {
text: String,
is_err: bool,
}
pub struct Job {
key: String,
log_path: PathBuf,
file: Option<std::fs::File>,
}
impl Job {
pub fn claim(app: &str, command: &str, detached: bool) -> Result<Self> {
let key = registry::command_key(app, command);
ensure_free(&key)?;
let (log_path, file) = match open_log(&key) {
Ok((path, file)) => (path, Some(file)),
Err(err) => {
progress::warn(&format!("cannot write the job log: {err:#}"));
(PathBuf::new(), None)
}
};
let work = Work::Command {
app: app.to_string(),
command: command.to_string(),
};
registry::save(&Entry::own(work, detached, file.as_ref().map(|_| log_path.clone())))?;
Ok(Self { key, log_path, file })
}
pub fn log_path(&self) -> Option<&Path> {
self.file.as_ref().map(|_| self.log_path.as_path())
}
pub fn note(&mut self, line: &str) {
self.write_line(&format!("turnout: {line}"));
}
pub fn keep(&mut self) -> Option<PathBuf> {
self.file.as_ref()?;
let kept = registry::failed_log(&self.log_path);
let copied = kept
.parent()
.map_or(Ok(()), std::fs::create_dir_all)
.and_then(|()| std::fs::copy(&self.log_path, &kept));
match copied {
Ok(_) => Some(kept),
Err(err) => {
self.note(&format!("cannot keep a copy of this log in {}: {err}", kept.display()));
None
}
}
}
fn write_line(&mut self, line: &str) {
if let Some(file) = &mut self.file {
if writeln!(file, "{line}").is_err() {
self.file = None;
}
}
}
fn mark_ready(&self, url: Option<String>) {
let _ = registry::update_own(&self.key, |entry| entry.ready = Some(registry::Ready { at: registry::now(), url }));
}
fn finish(&self, code: Option<i32>) {
let _ = registry::update_own(&self.key, |entry| entry.ended = Some(registry::Ended { at: registry::now(), code }));
}
}
pub fn ensure_free(key: &str) -> Result<()> {
if let Some(entry) = registry::load(key)?
&& entry.is_running()
{
bail!(
"{} is already running (pid {}) - see `turnout ps`, stop it with `turnout stop {}`",
entry.title(),
entry.pid,
entry.title()
);
}
Ok(())
}
fn open_log(key: &str) -> Result<(PathBuf, std::fs::File)> {
let path = registry::log_path(key)?;
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir).with_context(|| format!("cannot create {}", dir.display()))?;
}
let file = std::fs::File::create(&path).with_context(|| format!("cannot write {}", path.display()))?;
Ok((path, file))
}
pub enum Program<'a> {
Shell(&'a str),
Turnout(&'a [String]),
}
impl Program<'_> {
fn command(&self) -> Result<Command> {
Ok(match self {
#[cfg(windows)]
Program::Shell(line) => {
let mut command = Command::new("cmd");
command.args(["/C", line]);
command
}
#[cfg(not(windows))]
Program::Shell(line) => {
let mut command = Command::new("sh");
command.args(["-c", line]);
command
}
Program::Turnout(args) => {
let mut command = Command::new(std::env::current_exe().context("cannot locate the turnout binary")?);
command.args(*args);
command
}
})
}
fn describe(&self) -> String {
match self {
Program::Shell(line) => (*line).to_string(),
Program::Turnout(args) => format!("turnout {}", args.join(" ")),
}
}
}
pub struct Outcome {
pub status: std::process::ExitStatus,
pub elapsed: Duration,
pub tail: Vec<String>,
pub kept: Option<PathBuf>,
}
impl Outcome {
pub fn failed(&self) -> bool {
!self.status.success() && !crate::term::interrupted()
}
}
pub fn run(program: Program<'_>, dir: &Path, env: &[(&str, String)], mode: Mode, label: &str, job: &mut Job, mut ready: Option<Ready<'_>>) -> Result<Outcome> {
let command_line = program.describe();
let mut command = program.command()?;
command.current_dir(dir);
for (name, value) in env {
command.env(name, value);
}
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = match command.spawn() {
Ok(child) => child,
Err(err) => {
job.finish(None);
return Err(err).with_context(|| format!("cannot run '{command_line}'"));
}
};
crate::term::confine(&child);
crate::term::child_begin();
let (sender, receiver) = channel::<Line>();
let stdout = child.stdout.take().expect("stdout was piped");
let stderr = child.stderr.take().expect("stderr was piped");
let out_thread = spawn_reader(stdout, sender.clone(), false);
let err_thread = spawn_reader(stderr, sender, true);
let started = Instant::now();
let mut tail: Vec<String> = Vec::new();
let mut shown_any = false;
let mut step = match mode {
Mode::Stream | Mode::Log => None,
Mode::Quiet | Mode::UntilReady => Some(Step::start(format!("{label} ..."))),
};
let mut watching = ready.is_some();
let mut streaming = mode == Mode::Stream;
loop {
match receiver.recv_timeout(Duration::from_millis(200)) {
Ok(line) => {
job.write_line(&line.text);
remember(&mut tail, &line.text);
let mut settled = false;
if watching
&& let Some(detector) = ready.as_mut()
&& let Some(address) = detector.sees(&line.text)
{
watching = false;
settled = true;
let reached = detector.address(address.clone());
job.mark_ready(reached.clone());
if let Some(step) = step.take() {
step.done(detector.settled(address, started.elapsed()));
}
if let Some(on_ready) = detector.on_ready.take() {
on_ready(reached, started.elapsed());
}
}
if streaming {
emit(&line);
shown_any = true;
continue;
}
if !settled && !watching && mode == Mode::UntilReady && is_noteworthy(&line) {
if let Some(step) = step.take() {
step.clear();
}
emit(&line);
shown_any = true;
}
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
if watching
&& let Some(detector) = ready.as_ref()
&& started.elapsed() >= detector.patience
{
watching = false;
if !streaming && mode != Mode::Log {
streaming = true;
if let Some(step) = step.take() {
step.clear();
}
replay(&tail);
shown_any = !tail.is_empty();
}
}
if let Some(step) = &step {
step.update(format!("{label} ... {}", human_duration(started.elapsed())));
}
}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
}
}
let status = child.wait();
crate::term::child_end();
let _ = out_thread.join();
let _ = err_thread.join();
let status = status.with_context(|| format!("cannot run '{command_line}'"))?;
let elapsed = started.elapsed();
job.finish(status.code());
let mut outcome = Outcome {
status,
elapsed,
tail,
kept: None,
};
if outcome.failed() {
outcome.kept = job.keep();
}
match step {
Some(step) if status.success() => step.done(format!("{label} finished in {}", human_duration(elapsed))),
Some(step) => {
step.clear();
progress::outro_error(format!("{label} failed after {}", human_duration(elapsed)));
}
None => {}
}
if !status.success() && !shown_any && mode != Mode::Log {
replay(&outcome.tail);
if let Some(path) = outcome.kept.as_deref().or(job.log_path()) {
eprintln!("full output: {}", path.display());
}
}
Ok(outcome)
}
fn emit(line: &Line) {
if line.is_err {
eprintln!("{}", line.text);
} else {
println!("{}", line.text);
}
}
fn replay(tail: &[String]) {
if tail.is_empty() {
return;
}
let mut stderr = std::io::stderr().lock();
for line in tail {
let _ = writeln!(stderr, "{line}");
}
let _ = stderr.flush();
}
fn remember(tail: &mut Vec<String>, line: &str) {
if tail.len() == TAIL_LINES {
tail.remove(0);
}
tail.push(line.to_string());
}
fn is_noteworthy(line: &Line) -> bool {
if line.text.trim().is_empty() {
return false;
}
if line.is_err {
return true;
}
let lower = line.text.to_ascii_lowercase();
["error", "err!", "warn", "failed", "fail:", "cannot", "unable to", "exception"]
.iter()
.any(|needle| lower.contains(needle))
}
fn spawn_reader(pipe: impl Read + Send + 'static, sender: Sender<Line>, is_err: bool) -> std::thread::JoinHandle<()> {
std::thread::spawn(move || {
let mut reader = BufReader::new(pipe);
let mut buffer = Vec::new();
loop {
buffer.clear();
match reader.read_until(b'\n', &mut buffer) {
Ok(0) | Err(_) => break,
Ok(_) => {
while buffer.last().is_some_and(|b| *b == b'\n' || *b == b'\r') {
buffer.pop();
}
let text = String::from_utf8_lossy(&buffer).into_owned();
if sender.send(Line { text, is_err }).is_err() {
break;
}
}
}
}
})
}
pub type OnReady = Box<dyn FnOnce(Option<String>, Duration) + Send>;
pub struct Ready<'a> {
pub app: &'a str,
pub door: Option<String>,
pub own: Option<String>,
pub patience: Duration,
pub on_ready: Option<OnReady>,
}
impl Ready<'_> {
fn sees(&self, line: &str) -> Option<Option<String>> {
ready(line)
}
fn address(&self, seen: Option<String>) -> Option<String> {
self.door.clone().or(seen).or_else(|| self.own.clone())
}
fn settled(&self, address: Option<String>, elapsed: Duration) -> String {
match self.address(address) {
Some(url) => format!("{} ready in {} - {url}", self.app, human_duration(elapsed)),
None => format!("{} ready in {}", self.app, human_duration(elapsed)),
}
}
}
pub fn ready(line: &str) -> Option<Option<String>> {
let plain = strip_ansi(line);
let text = plain.trim();
if text.is_empty() {
return None;
}
let lower = text.to_ascii_lowercase();
let announced = lower.contains("ready in") || lower.contains("compiled ") || lower.ends_with("compiled");
let address = url_in(text);
if announced || address.is_some() {
return Some(address);
}
None
}
fn url_in(text: &str) -> Option<String> {
let at = text.find("http://").or_else(|| text.find("https://"))?;
let rest = &text[at..];
let end = rest
.find(|c: char| c.is_whitespace() || c == '"' || c == '\'' || c == '<' || c == ')')
.unwrap_or(rest.len());
let url = rest[..end].trim_end_matches(['.', ',', ';']);
(!url.is_empty()).then(|| url.to_string())
}
pub(crate) fn strip_ansi(line: &str) -> String {
let mut out = String::with_capacity(line.len());
let mut chars = line.chars();
while let Some(c) = chars.next() {
if c != '\x1b' {
out.push(c);
continue;
}
if chars.next() != Some('[') {
continue;
}
for c in chars.by_ref() {
if ('\x40'..='\x7e').contains(&c) {
break;
}
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ready_recognises_the_servers_it_claims_to() {
assert_eq!(ready(" \x1b[32mVITE v5.4.0\x1b[0m \x1b[2mready in\x1b[0m 431 ms"), Some(None));
assert_eq!(ready(" \x1b[2mready\x1b[0m \x1b[2min\x1b[0m 431 ms"), Some(None));
assert_eq!(ready(" ➜ Local: http://localhost:5100/"), Some(Some("http://localhost:5100/".to_string())));
assert_eq!(ready("✓ Compiled in 1.2s (412 modules)"), Some(None));
assert_eq!(ready("- Ready in 2.3s"), Some(None));
assert_eq!(
ready("Server listening on http://127.0.0.1:3000"),
Some(Some("http://127.0.0.1:3000".to_string()))
);
assert_eq!(ready(""), None);
assert_eq!(ready(" "), None);
assert_eq!(ready("$ vite --port 5100"), None);
assert_eq!(ready("watching for file changes"), None);
assert_eq!(ready("error: cannot resolve ./missing"), None);
}
#[test]
fn the_address_comes_out_clean() {
assert_eq!(url_in("see http://localhost:5100, then"), Some("http://localhost:5100".to_string()));
assert_eq!(url_in("open (http://a.localhost) now"), Some("http://a.localhost".to_string()));
assert_eq!(url_in("\"https://example.test/path\""), Some("https://example.test/path".to_string()));
assert_eq!(url_in("no address here"), None);
}
#[test]
fn the_quiet_filter_keeps_problems_and_drops_chatter() {
let out = |text: &str| Line {
text: text.to_string(),
is_err: false,
};
let err = |text: &str| Line {
text: text.to_string(),
is_err: true,
};
assert!(is_noteworthy(&err("anything on stderr")));
assert!(is_noteworthy(&out("Error: Failed to resolve import")));
assert!(is_noteworthy(&out("[plugin:vite] warning: unused")));
assert!(is_noteworthy(&out("build failed with 1 error")));
assert!(!is_noteworthy(&out("hmr update /src/App.vue")));
assert!(!is_noteworthy(&out("page reload src/main.ts")));
assert!(!is_noteworthy(&err(" ")));
}
#[test]
fn the_tail_keeps_the_last_lines_and_no_more() {
let mut tail = Vec::new();
for i in 0..(TAIL_LINES + 10) {
remember(&mut tail, &format!("line {i}"));
}
assert_eq!(tail.len(), TAIL_LINES);
assert_eq!(tail.first().unwrap(), &format!("line {}", 10));
assert_eq!(tail.last().unwrap(), &format!("line {}", TAIL_LINES + 9));
}
#[test]
fn the_ready_line_names_an_address_whenever_one_exists() {
let detector = |door: Option<&str>, own: Option<&str>| Ready {
app: "myapp",
door: door.map(str::to_string),
own: own.map(str::to_string),
patience: Duration::from_secs(1),
on_ready: None,
};
let at = Duration::from_millis(800);
assert_eq!(
detector(Some("http://myapp.localhost"), Some("http://localhost:5100")).settled(Some("http://localhost:5173".into()), at),
"myapp ready in 0.8s - http://myapp.localhost"
);
assert_eq!(
detector(None, Some("http://localhost:5100")).settled(Some("http://localhost:5173".into()), at),
"myapp ready in 0.8s - http://localhost:5173"
);
assert_eq!(
detector(None, Some("http://localhost:5100")).settled(None, at),
"myapp ready in 0.8s - http://localhost:5100"
);
assert_eq!(detector(None, None).settled(None, at), "myapp ready in 0.8s");
}
#[test]
fn verbosity_and_pipes_both_force_streaming() {
assert!(
std::env::var(MODE_ENV).is_err(),
"{MODE_ENV} is set in this process; the resolve tests describe the unforced path"
);
assert_eq!(Mode::resolve(Mode::Quiet, true), Mode::Stream);
assert_eq!(Mode::resolve(Mode::UntilReady, true), Mode::Stream);
assert_eq!(Mode::resolve(Mode::Quiet, false), Mode::Stream);
}
}