diff --git a/src/cli.rs b/src/cli.rs index 255d01c..db1e4b7 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -50,6 +50,17 @@ pub enum Commands { List, /// Start all enabled services Daemon, + /// Send one line to a service's stdin + Send { + name: String, + message: Vec, + }, + /// Attach to a service's stdin interactively + Attach { + name: String, + #[arg(long, help = "Send SIGINT to service on Ctrl+C before disconnecting")] + full: bool, + }, /// Internal: run supervisor for a service #[command(hide = true)] Supervise { diff --git a/src/daemon.rs b/src/daemon.rs index 76d7992..d52bf38 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -7,6 +7,7 @@ use std::os::unix::process::CommandExt; use std::process::{Child, Command, Stdio}; use std::time::{Duration, SystemTime}; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; use crate::config::{Config, RestartPolicy, ServiceConfig}; use crate::state; @@ -22,6 +23,7 @@ struct Proc { pid: u32, cfg: ServiceConfig, restart_count: u32, + stdin: Option>>, } pub fn run(config_path: String) { @@ -174,7 +176,7 @@ fn spawn(name: &str, cfg: &ServiceConfig) -> Option { .envs(&cfg.env) .stdout(Stdio::piped()) .stderr(Stdio::piped()) - .stdin(Stdio::null()); + .stdin(Stdio::piped()); // SAFETY: pre_exec runs in the child after fork; minimal ops only unsafe { sh.pre_exec(|| { @@ -196,6 +198,7 @@ fn spawn(name: &str, cfg: &ServiceConfig) -> Option { }; let pid = child.id(); + let child_stdin = child.stdin.take().map(|s| Arc::new(Mutex::new(s))); let log_out = log_path.clone(); let log_err = log_path.clone(); @@ -225,6 +228,7 @@ fn spawn(name: &str, cfg: &ServiceConfig) -> Option { pid, cfg: cfg.clone(), restart_count: 0, + stdin: child_stdin, }) } @@ -360,9 +364,36 @@ fn handle_client( if reader.read_line(&mut line).is_err() { return; } - let line = line.trim(); + let line = line.trim().to_string(); - let response = match line { + if line.starts_with("stdin ") { + drop(reader); + handle_stdin(&line, stream, procs); + return; + } + if line.starts_with("signal ") { + let rest = &line[7..]; + let response = if let Some(space) = rest.find(' ') { + let name = &rest[..space]; + let sig_name = &rest[space + 1..]; + if let Some(p) = procs.get(name) { + let sig = crate::supervisor::parse_signal(sig_name); + match nix::sys::signal::kill(nix::unistd::Pid::from_raw(p.pid as i32), sig) { + Ok(_) => "ok".to_string(), + Err(e) => format!("error: signal failed: {}", e), + } + } else { + format!("error: '{}' not running", name) + } + } else { + "error: usage: signal ".to_string() + }; + let _ = writeln!(stream, "{}", response); + let _ = stream.flush(); + return; + } + + let response = match line.as_str() { l if l.starts_with("start ") => { let names: Vec<&str> = l[6..].split_whitespace().collect(); let mut out = Vec::new(); @@ -499,6 +530,79 @@ fn handle_client( let _ = stream.flush(); } +fn handle_stdin(line: &str, mut stream: &UnixStream, procs: &mut HashMap) { + let rest = &line[6..]; + if let Some(space) = rest.find(' ') { + let name = &rest[..space]; + let msg = &rest[space + 1..]; + match procs.get(name) { + Some(p) => match &p.stdin { + Some(stdin) => { + let mut stdin = stdin.lock().unwrap(); + let _ = writeln!(stdin, "{}", msg); + let _ = stdin.flush(); + let _ = writeln!(stream, "ok"); + } + None => { + let _ = writeln!(stream, "error: '{}' has no stdin", name); + } + }, + None => { + let _ = writeln!(stream, "error: '{}' not running", name); + } + } + let _ = stream.flush(); + } else { + let name = rest; + let has_stdin = procs.get(name).and_then(|p| p.stdin.as_ref()).is_some(); + if !procs.contains_key(name) { + let _ = writeln!(stream, "error: '{}' not running", name); + let _ = stream.flush(); + return; + } + if !has_stdin { + let _ = writeln!(stream, "error: '{}' has no stdin", name); + let _ = stream.flush(); + return; + } + let stdin_arc = procs.get(name).and_then(|p| p.stdin.clone()).unwrap(); + match stream.try_clone() { + Ok(clone) => { + let _ = writeln!(stream, "ok"); + let _ = stream.flush(); + std::thread::spawn(move || { + let mut reader = BufReader::new(clone); + let mut line = String::new(); + loop { + line.clear(); + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => { + let t = line.trim(); + if !t.is_empty() { + let mut s = match stdin_arc.lock() { + Ok(s) => s, + Err(_) => break, + }; + if writeln!(s, "{}", t).is_err() { + break; + } + let _ = s.flush(); + } + } + Err(_) => break, + } + } + }); + } + Err(e) => { + let _ = writeln!(stream, "error: {}", e); + let _ = stream.flush(); + } + } + } +} + fn append_log(path: &str, msg: &str) { let ts = chrono::Local::now().format("%Y-%m-%d %H:%M:%S"); if let Ok(mut f) = std::fs::OpenOptions::new() diff --git a/src/ipc.rs b/src/ipc.rs index 30ad592..3cfa83e 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -21,3 +21,27 @@ pub fn send_cmd(cmd: &str) -> Result, String> { } Ok(lines) } + +pub fn send_cmd_get_stream(cmd: &str) -> Result { + let path = crate::state::socket_path(); + let mut stream = + UnixStream::connect(&path).map_err(|e| format!("daemon not running: {}", e))?; + + writeln!(stream, "{}", cmd).map_err(|e| format!("write: {}", e))?; + stream.flush().map_err(|e| format!("flush: {}", e))?; + + let mut reader = BufReader::new(&stream); + let mut line = String::new(); + reader + .read_line(&mut line) + .map_err(|e| format!("read: {}", e))?; + let trimmed = line.trim(); + if let Some(msg) = trimmed.strip_prefix("error:") { + return Err(msg.trim().to_string()); + } + if trimmed != "ok" { + return Err(format!("unexpected response: {}", trimmed)); + } + drop(reader); + Ok(stream) +} diff --git a/src/main.rs b/src/main.rs index d67581e..cc56a34 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,6 +8,8 @@ mod supervisor; use clap::Parser; use cli::{Cli, Commands}; +use std::io::Write; +use std::sync::atomic::{AtomicBool, Ordering}; fn main() { @@ -26,6 +28,8 @@ fn main() { Commands::Logs { ref name, lines, follow } => cmd_logs(name, lines, follow), Commands::Enable { ref names, now } => cmd_enable(names, now), Commands::Disable { ref names } => cmd_disable(names), + Commands::Send { ref name, ref message } => cmd_send(name, message), + Commands::Attach { ref name, full } => cmd_attach(name, full), Commands::Supervise { .. } => { eprintln!("error: supervise is internal, use daemon"); std::process::exit(1); @@ -180,3 +184,80 @@ fn cmd_disable(names: &[String]) { } } } + +fn cmd_send(name: &str, message: &[String]) { + let msg = message.join(" "); + let cmd = format!("stdin {} {}", name, msg); + match ipc::send_cmd(&cmd) { + Ok(_) => println!("ok"), + Err(e) => { + eprintln!("error: {}", e); + std::process::exit(1); + } + } +} + +static ATTACH_INTERRUPT: AtomicBool = AtomicBool::new(false); + +extern "C" fn handle_attach_int(_: i32) { + ATTACH_INTERRUPT.store(true, Ordering::Relaxed); +} + +fn cmd_attach(name: &str, full: bool) { + let sig_action = nix::sys::signal::SigAction::new( + nix::sys::signal::SigHandler::Handler(handle_attach_int), + nix::sys::signal::SaFlags::empty(), + nix::sys::signal::SigSet::empty(), + ); + unsafe { + let _ = nix::sys::signal::sigaction(nix::sys::signal::Signal::SIGINT, &sig_action); + } + + let stream = match ipc::send_cmd_get_stream(&format!("stdin {}", name)) { + Ok(s) => s, + Err(e) => { + eprintln!("error: {}", e); + std::process::exit(1); + } + }; + + // Tail the log file in a background thread + let log_path = crate::state::log_path(name); + if log_path.exists() { + let log_path_str = log_path.to_string_lossy().to_string(); + std::thread::spawn(move || { + let _ = crate::log::show_logs(&log_path_str, Some(0), true); + }); + } + + // Main loop: read stdin line by line, send to daemon + let stdin = std::io::stdin(); + loop { + if ATTACH_INTERRUPT.load(Ordering::Relaxed) { + if full { + let _ = ipc::send_cmd(&format!("signal {} INT", name)); + } + break; + } + let mut line = String::new(); + match stdin.read_line(&mut line) { + Ok(0) => break, + Ok(_) => { + let trimmed = line.trim_end_matches('\n').trim_end_matches('\r'); + let bytes = trimmed.as_bytes(); + if bytes.len() >= 2 && bytes[0] == 0x10 && (bytes[1] == b'd' || bytes[1] == b'D') + { + break; + } + if !trimmed.is_empty() { + let _ = writeln!(&stream, "{}", trimmed); + let _ = (&stream).flush(); + } + } + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => { + continue; + } + Err(_) => break, + } + } +}