Index: .efiles ================================================================== --- .efiles +++ .efiles @@ -1,8 +1,10 @@ Cargo.toml README.md www/index.md +www/opinionated.md +www/qsurt.md www/design-notes.md www/changelog.md src/err.rs src/lib.rs src/lumberjack.rs @@ -24,11 +26,14 @@ src/argp.rs tests/err/mod.rs tests/apps/mod.rs tests/apperr.rs tests/initrunshutdown.rs +examples/simplesync.rs +examples/simpletokio.rs +examples/simplerocket.rs examples/hellosvc.rs examples/hellosvc-tokio.rs examples/hellosvc-rocket.rs examples/err/mod.rs examples/procres/mod.rs examples/argp/mod.rs Index: Cargo.toml ================================================================== --- Cargo.toml +++ Cargo.toml @@ -1,21 +1,22 @@ [package] name = "qsu" -version = "0.5.0" +version = "0.6.0" edition = "2021" license = "0BSD" # https://crates.io/category_slugs categories = [ "os" ] keywords = [ "service", "systemd", "winsvc" ] repository = "https://repos.qrnch.tech/pub/qsu" -description = "Service subsystem wrapper." -rust-version = "1.56" +description = "Service subsystem utilities and runtime wrapper." +rust-version = "1.70" exclude = [ ".fossil-settings", ".efiles", ".fslckout", "www", + "bacon.toml", "build_docs.sh", "Rocket.toml", "rustfmt.toml" ] @@ -33,54 +34,53 @@ rt = [] tokio = ["rt", "tokio/macros", "tokio/rt-multi-thread", "tokio/signal"] wait-for-debugger = ["dep:dbgtools-win"] [dependencies] -apperr = { version = "0.2.0" } -async-trait = { version = "0.1.80" } +async-trait = { version = "0.1.82" } chrono = { version = "0.4.38" } -clap = { version = "4.5.4", optional = true, features = [ +clap = { version = "4.5.17", optional = true, features = [ "derive", "env", "string", "wrap_help" ] } -env_logger = { version = "0.11.3" } +env_logger = { version = "0.11.5" } futures = { version = "0.3.30" } itertools = { version = "0.13.0", optional = true } killswitch = { version = "0.4.2" } -log = { version = "0.4.20" } -parking_lot = { version = "0.12.2" } -rocket = { version = "0.5.0", optional = true } +log = { version = "0.4.22" } +parking_lot = { version = "0.12.3" } +rocket = { version = "0.5.1", optional = true } sidoc = { version = "0.1.0", optional = true } -tokio = { version = "1.37.0", features = ["sync"] } +tokio = { version = "1.40.0", features = ["sync"] } time = { version = "0.3.36", features = ["macros"] } tracing = { version = "0.1.40" } [dependencies.tracing-subscriber] version = "0.3.18" default-features = false features = ["env-filter", "time", "fmt", "ansi"] [target.'cfg(target_os = "linux")'.dependencies] -sd-notify = { version = "0.4.1", optional = true } +sd-notify = { version = "0.4.2", optional = true } [target.'cfg(unix)'.dependencies] -libc = { version = "0.2.155" } -nix = { version = "0.28.0", features = ["pthread", "signal"] } +libc = { version = "0.2.158" } +nix = { version = "0.29.0", features = ["pthread", "signal", "time"] } [target.'cfg(windows)'.dependencies] dbgtools-win = { version = "0.2.1", optional = true } eventlog = { version = "0.2.2" } registry = { version = "1.2.3" } scopeguard = { version = "1.2.0" } windows-service = { version = "0.7.0" } -windows-sys = { version = "0.52.0", features = [ +windows-sys = { version = "0.59.0", features = [ "Win32_Foundation", "Win32_System_Console" ] } winreg = { version = "0.52.0" } [dev-dependencies] clap = { version = "4.5.4", features = ["derive", "env", "wrap_help"] } -tokio = { version = "1.37.0", features = ["time"] } +tokio = { version = "1.40.0", features = ["time"] } [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs", "--generate-link-to-definition"] @@ -94,5 +94,13 @@ [[example]] name = "hellosvc-rocket" required-features = ["clap", "installer", "rt", "rocket"] +[lints.clippy] +all = { level = "deny", priority = -1 } +pedantic = { level = "warn", priority = -1 } +nursery = { level = "warn", priority = -1 } +cargo = { level = "warn", priority = -1 } + +multiple_crate_versions = "allow" + Index: README.md ================================================================== --- README.md +++ README.md @@ -1,9 +1,11 @@ # qsu -The _qsu_ ("kazoo") crate is a: +The _qsu_ ("kazoo") crate offers portable service utilities with an +opinionated service wrapper runtime. -- service runtime that acts as a layer betwen the server application code - and an operating system service subsystem (launchd, systemd, Windows - Services) -- set of utility functions for working with services. +_qsu_'s primary objective is to allow a service developer to focus on the +actual service application code, without having to bother with service +subsystem-specific integrations -- while at the same time allowing the +service application to run as a regular foreground process, without the code +needing to diverge between the two. ADDED bacon.toml Index: bacon.toml ================================================================== --- /dev/null +++ bacon.toml @@ -0,0 +1,107 @@ +# This is a configuration file for the bacon tool +# +# Bacon repository: https://github.com/Canop/bacon +# Complete help on configuration: https://dystroy.org/bacon/config/ +# You can also check bacon's own bacon.toml file +# as an example: https://github.com/Canop/bacon/blob/main/bacon.toml + +# For information about clippy lints, see: +# https://github.com/rust-lang/rust-clippy/blob/master/README.md + +#default_job = "check" +default_job = "clippy-all" + +[jobs.check] +command = ["cargo", "check", "--color", "always"] +need_stdout = false + +[jobs.check-all] +command = ["cargo", "check", "--all-targets", "--color", "always"] +need_stdout = false + +# Run clippy on the default target +[jobs.clippy] +command = [ + "cargo", "clippy", + "--all-features", + "--color", "always", +] +need_stdout = false + +# Run clippy on all targets +# To disable some lints, you may change the job this way: +# [jobs.clippy-all] +# command = [ +# "cargo", "clippy", +# "--all-targets", +# "--color", "always", +# "--", +# "-A", "clippy::bool_to_int_with_if", +# "-A", "clippy::collapsible_if", +# "-A", "clippy::derive_partial_eq_without_eq", +# ] +# need_stdout = false +[jobs.clippy-all] +command = [ + "cargo", "clippy", + "--all-features", + "--all-targets", + "--color", "always", +] +need_stdout = false + +# This job lets you run +# - all tests: bacon test +# - a specific test: bacon test -- config::test_default_files +# - the tests of a package: bacon test -- -- -p config +[jobs.test] +command = [ + "cargo", "test", "--color", "always", + "--", "--color", "always", # see https://github.com/Canop/bacon/issues/124 +] +need_stdout = true + +[jobs.doc] +command = ["cargo", "doc", "--color", "always", "--no-deps"] +need_stdout = false + +# If the doc compiles, then it opens in your browser and bacon switches +# to the previous job +[jobs.doc-open] +command = ["cargo", "doc", "--color", "always", "--no-deps", "--open"] +need_stdout = false +on_success = "back" # so that we don't open the browser at each change + +# You can run your application and have the result displayed in bacon, +# *if* it makes sense for this crate. +# Don't forget the `--color always` part or the errors won't be +# properly parsed. +# If your program never stops (eg a server), you may set `background` +# to false to have the cargo run output immediately displayed instead +# of waiting for program's end. +[jobs.run] +command = [ + "cargo", "run", + "--color", "always", + # put launch parameters for your program behind a `--` separator +] +need_stdout = true +allow_warnings = true +background = true + +# This parameterized job runs the example of your choice, as soon +# as the code compiles. +# Call it as +# bacon ex -- my-example +[jobs.ex] +command = ["cargo", "run", "--color", "always", "--example"] +need_stdout = true +allow_warnings = true + +# You may define here keybindings that would be specific to +# a project, for example a shortcut to launch a specific job. +# Shortcuts to internal functions (scrolling, toggling, etc.) +# should go in your personal global prefs.toml file instead. +[keybindings] +# alt-m = "job:my-job" +c = "job:clippy-all" # comment this to have 'c' run clippy on only the default target Index: examples/argp/mod.rs ================================================================== --- examples/argp/mod.rs +++ examples/argp/mod.rs @@ -1,47 +1,59 @@ use clap::ArgMatches; -use qsu::{installer::RegSvc, rt::SrvAppRt, AppErr}; +use qsu::{installer::RegSvc, rt::SrvAppRt}; use crate::err::Error; -pub(crate) struct AppArgsProc { - pub(crate) bldr: Box SrvAppRt> +pub struct AppArgsProc { + pub(crate) bldr: Box SrvAppRt> } impl qsu::argp::ArgsProc for AppArgsProc { + type AppErr = Error; + /// Process an `register-service` subcommand. fn proc_inst( &mut self, _sub_m: &ArgMatches, regsvc: RegSvc - ) -> Result { + ) -> Result { // This is split out into its own function because the orphan rule wouldn't // allow the application to implement a std::io::Error -> qsu::AppErr // conversion in one go, so we do it in two steps instead. // proc_inst_inner()'s '?' converts "all" errors into 'Error`. // The proc_inst() method's `?` converts from `Error` to `qsu::AppError` - Ok(proc_inst_inner(regsvc)?) + proc_inst_inner(regsvc) } - fn build_apprt(&mut self) -> Result { + fn build_apprt(&mut self) -> Result, Self::AppErr> { Ok((self.bldr)()) } } fn proc_inst_inner(regsvc: RegSvc) -> Result { // Use current working directory as the service's workdir let cwd = std::env::current_dir()?.to_str().unwrap().to_string(); - let regsvc = regsvc + let mut regsvc = regsvc .workdir(cwd) .env("HOLY", "COW") .env("Private", "Public") .env("General", "Specific"); + + // Set display name, but only if it wasn't already specified on the command + // line + if regsvc.display_name.is_none() { + regsvc.display_name_ref("Hello Service"); + } + + // Hard-code servie as supporting conf-reload service events. + let regsvc = regsvc.conf_reload(); // Add a callback that will increase log and trace levels by deafault. #[cfg(windows)] let regsvc = regsvc.regconf(|_svcname, params| { + // Leave a key/value pair in the registry as a hello params.set_value("AppArgParser", &"SaysHello")?; Ok(()) }); Index: examples/err/mod.rs ================================================================== --- examples/err/mod.rs +++ examples/err/mod.rs @@ -5,34 +5,29 @@ IO(String), Qsu(String) } impl std::error::Error for Error {} -impl apperr::Blessed for Error {} impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match self { - Error::IO(s) => { - write!(f, "I/O error; {}", s) - } - Error::Qsu(s) => { - write!(f, "qsu error; {}", s) - } + Self::IO(s) => write!(f, "I/O error; {s}"), + Self::Qsu(s) => write!(f, "qsu error; {s}") } } } impl From for Error { fn from(err: io::Error) -> Self { - Error::IO(err.to_string()) + Self::IO(err.to_string()) } } -impl From for Error { - fn from(err: qsu::Error) -> Self { - Error::Qsu(err.to_string()) +impl From> for Error { + fn from(err: qsu::CbErr) -> Self { + Self::Qsu(err.to_string()) } } /* /// Convenience converter used to pass application-defined errors from the @@ -43,13 +38,13 @@ } } */ /// Convenience converter for mapping application-specific errors to -/// `qsu::AppErr`. -impl From for qsu::AppErr { - fn from(err: Error) -> qsu::AppErr { - qsu::AppErr::new(err) +/// `qsu::CbErr::App`. +impl From for qsu::CbErr { + fn from(err: Error) -> Self { + Self::App(err) } } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: examples/hellosvc-rocket.rs ================================================================== --- examples/hellosvc-rocket.rs +++ examples/hellosvc-rocket.rs @@ -5,30 +5,33 @@ mod err; mod procres; use qsu::{ argp::ArgParser, - rt::{ - InitCtx, RocketServiceHandler, RunEnv, SrvAppRt, SvcEvt, SvcEvtReader, - TermCtx - } + rt::{InitCtx, RocketServiceHandler, RunEnv, SrvAppRt, SvcEvt, TermCtx} }; + +use tokio::sync::mpsc; use rocket::{Build, Ignite, Rocket}; use err::Error; use procres::ProcRes; -struct MyService {} +struct MyService { + rx: mpsc::UnboundedReceiver +} #[qsu::async_trait] impl RocketServiceHandler for MyService { + type AppErr = Error; + async fn init( &mut self, ictx: InitCtx - ) -> Result>, qsu::AppErr> { + ) -> Result>, Self::AppErr> { tracing::trace!("Running init()"); let mut rockets = vec![]; ictx.report(Some("Building a rocket!".into())); @@ -38,36 +41,29 @@ rockets.push(rocket); Ok(rockets) } + #[allow(clippy::redundant_pub_crate)] async fn run( &mut self, rockets: Vec>, - _re: &RunEnv, - mut set: SvcEvtReader - ) -> Result<(), qsu::AppErr> { + _re: &RunEnv + ) -> Result<(), Self::AppErr> { for rocket in rockets { tokio::task::spawn(async { rocket.launch().await.unwrap(); }); } loop { tokio::select! { - evt = set.arecv() => { + evt = self.rx.recv() => { match evt { - Some(SvcEvt::Shutdown) => { + Some(SvcEvt::Shutdown(_)) => { tracing::info!("The service subsystem requested that the application shut down"); - break; - } - Some(SvcEvt::Terminate) => { - tracing::info!( - "The service subsystem requested that the application - terminate" - ); break; } Some(SvcEvt::ReloadConf) => { tracing::info!("The service subsystem requested that application reload configuration"); @@ -79,11 +75,11 @@ } Ok(()) } - async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), qsu::AppErr> { + async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr> { tctx.report(Some(format!("Entered {}", "shutdown").into())); tracing::trace!("Running shutdown()"); Ok(()) } } @@ -98,20 +94,38 @@ fn main2() -> Result<(), Error> { // Derive default service name from executable name. let svcname = qsu::default_service_name() .expect("Unable to determine default service name"); - let creator = || { - let handler = Box::new(MyService {}); - SrvAppRt::Rocket(handler) - }; + let bldr = Box::new(|| { + let (tx, rx) = mpsc::unbounded_channel(); + let svcevt_handler = Box::new(move |msg| { + // Just foward messages to runtime handler + tx.send(msg).unwrap(); + }); + let rt_handler = Box::new(MyService { rx }); + SrvAppRt::Rocket { + svcevt_handler, + rt_handler + } + }); // Parse, and process, command line arguments. - let mut argsproc = argp::AppArgsProc { - bldr: Box::new(creator) - }; - let ap = ArgParser::new(&svcname, &mut argsproc); + let mut argsproc = argp::AppArgsProc { bldr }; + let ap = + ArgParser::new(&svcname, &mut argsproc).regsvc_proc(|mut regsvc| { + // Hard-code this as a network service application. + regsvc.netservice_ref(); + + // Set description, but only if it hasn't been set already (presumably + // via the command line). + if regsvc.description.is_none() { + regsvc.description_ref("A simple Rocket server that logs requests"); + } + + regsvc + }); ap.proc()?; Ok(()) } Index: examples/hellosvc-tokio.rs ================================================================== --- examples/hellosvc-tokio.rs +++ examples/hellosvc-tokio.rs @@ -4,41 +4,45 @@ mod err; mod procres; use std::time::{Duration, Instant}; +use tokio::sync::mpsc; + use qsu::{ argp::ArgParser, - rt::{ - InitCtx, RunEnv, SrvAppRt, SvcEvt, SvcEvtReader, TermCtx, - TokioServiceHandler - } + rt::{InitCtx, RunEnv, SrvAppRt, SvcEvt, TermCtx, TokioServiceHandler} }; use err::Error; use procres::ProcRes; -struct MyService {} +struct MyService { + rx: mpsc::UnboundedReceiver +} #[qsu::async_trait] impl TokioServiceHandler for MyService { - async fn init(&mut self, ictx: InitCtx) -> Result<(), qsu::AppErr> { + type AppErr = Error; + + async fn init(&mut self, ictx: InitCtx) -> Result<(), Self::AppErr> { ictx.report(Some("Entered init".into())); tracing::trace!("Running init()"); Ok(()) } - async fn run( - &mut self, - _re: &RunEnv, - mut set: SvcEvtReader - ) -> Result<(), qsu::AppErr> { + #[allow(clippy::redundant_pub_crate)] + async fn run(&mut self, _re: &RunEnv) -> Result<(), Self::AppErr> { const SECS: u64 = 30; - let mut last_dump = Instant::now() - Duration::from_secs(SECS); + + // unwrap() is okay due to time scales involed + let mut last_dump = Instant::now() + .checked_sub(Duration::from_secs(SECS)) + .unwrap(); loop { - if Instant::now() - last_dump > Duration::from_secs(SECS) { + if last_dump.elapsed() > Duration::from_secs(SECS) { log::error!("error"); log::warn!("warn"); log::info!("info"); log::debug!("debug"); log::trace!("trace"); @@ -50,23 +54,17 @@ tracing::trace!("trace"); last_dump = Instant::now(); } tokio::select! { - _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => { + () = tokio::time::sleep(std::time::Duration::from_secs(1)) => { continue; } - evt = set.arecv() => { + evt = self.rx.recv() => { match evt { - Some(SvcEvt::Shutdown) => { + Some(SvcEvt::Shutdown(_)) => { tracing::info!("The service subsystem requested that the application shut down"); - break; - } - Some(SvcEvt::Terminate) => { - tracing::info!( - "The service subsystem requested that the application terminate" - ); break; } Some(SvcEvt::ReloadConf) => { tracing::info!("The service subsystem requested that application reload configuration"); } @@ -77,11 +75,11 @@ } Ok(()) } - async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), qsu::AppErr> { + async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr> { tctx.report(Some(("Entered shutdown".to_string()).into())); tracing::trace!("Running shutdown()"); Ok(()) } } @@ -96,21 +94,28 @@ fn main2() -> Result<(), Error> { // Derive default service name from executable name. let svcname = qsu::default_service_name() .expect("Unable to determine default service name"); - let creator = || { - let handler = Box::new(MyService {}); - SrvAppRt::Tokio(None, handler) - }; + let bldr = Box::new(|| { + let (tx, rx) = mpsc::unbounded_channel(); + let svcevt_handler = Box::new(move |msg| { + // Just foward messages to runtime handler + tx.send(msg).unwrap(); + }); + let rt_handler = Box::new(MyService { rx }); + SrvAppRt::Tokio { + rtbldr: None, + svcevt_handler, + rt_handler + } + }); // Parse, and process, command line arguments. - let mut argsproc = argp::AppArgsProc { - bldr: Box::new(creator) - }; + let mut argsproc = argp::AppArgsProc { bldr }; let ap = ArgParser::new(&svcname, &mut argsproc); ap.proc()?; Ok(()) } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: examples/hellosvc.rs ================================================================== --- examples/hellosvc.rs +++ examples/hellosvc.rs @@ -7,39 +7,42 @@ use std::{ thread, time::{Duration, Instant} }; +use tokio::sync::mpsc; + use qsu::{ argp::ArgParser, - rt::{ - InitCtx, RunEnv, ServiceHandler, SrvAppRt, SvcEvt, SvcEvtReader, TermCtx - } + rt::{InitCtx, RunEnv, ServiceHandler, SrvAppRt, SvcEvt, TermCtx} }; use err::Error; use procres::ProcRes; -struct MyService {} +struct MyService { + rx: mpsc::UnboundedReceiver +} impl ServiceHandler for MyService { - fn init(&mut self, ictx: InitCtx) -> Result<(), qsu::AppErr> { + type AppErr = Error; + + fn init(&mut self, ictx: InitCtx) -> Result<(), Self::AppErr> { ictx.report(Some("Entered init".into())); tracing::trace!("Running init()"); Ok(()) } - fn run( - &mut self, - _re: &RunEnv, - mut set: SvcEvtReader - ) -> Result<(), qsu::AppErr> { + fn run(&mut self, _re: &RunEnv) -> Result<(), Self::AppErr> { const SECS: u64 = 30; - let mut last_dump = Instant::now() - Duration::from_secs(SECS); + // unwrap() is okay due to time scale + let mut last_dump = Instant::now() + .checked_sub(Duration::from_secs(SECS)) + .unwrap(); loop { - if Instant::now() - last_dump > Duration::from_secs(SECS) { + if last_dump.elapsed() > Duration::from_secs(SECS) { log::error!("error"); log::warn!("warn"); log::info!("info"); log::debug!("debug"); log::trace!("trace"); @@ -51,24 +54,16 @@ tracing::trace!("trace"); last_dump = Instant::now(); } - match set.try_recv() { - Some(SvcEvt::Shutdown) => { - tracing::info!( - "The service subsystem requested that the application shut down" - ); + match self.rx.try_recv() { + Ok(SvcEvt::Shutdown(_)) => { + tracing::info!("Service application shutdown"); break; } - Some(SvcEvt::Terminate) => { - tracing::info!( - "The service subsystem requested that the application terminate" - ); - break; - } - Some(SvcEvt::ReloadConf) => { + Ok(SvcEvt::ReloadConf) => { tracing::info!( "The service subsystem requested that the application reload its \ configuration" ); } @@ -79,11 +74,11 @@ } Ok(()) } - fn shutdown(&mut self, tctx: TermCtx) -> Result<(), qsu::AppErr> { + fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr> { tctx.report(Some(("Entered shutdown".to_string()).into())); tracing::trace!("Running shutdown()"); Ok(()) } } @@ -98,21 +93,38 @@ fn main2() -> Result<(), Error> { // Derive default service name from executable name. let svcname = qsu::default_service_name() .expect("Unable to determine default service name"); - let creator = || { - let handler = Box::new(MyService {}); - SrvAppRt::Sync(handler) - }; + // Create a closure that will generate a sync service runtime with a + // service event handler that will forward any messages it receives to a + // channel receiver end-point held by the main service handler. + let bldr = Box::new(|| { + let (tx, rx) = mpsc::unbounded_channel(); + SrvAppRt::Sync { + svcevt_handler: Box::new(move |msg| { + // Just foward messages to runtime handler + tx.send(msg).unwrap(); + }), + rt_handler: Box::new(MyService { rx }) + } + }); // Parse, and process, command line arguments. - let mut argsproc = argp::AppArgsProc { - bldr: Box::new(creator) - }; - let ap = ArgParser::new(&svcname, &mut argsproc); + // Add a service registration callback that will set service's description. + let mut argsproc = argp::AppArgsProc { bldr }; + let ap = + ArgParser::new(&svcname, &mut argsproc).regsvc_proc(|mut regsvc| { + // Set description, but only if it hasn't been set already (presumably + // via the command line). + if regsvc.description.is_none() { + regsvc + .description_ref("A sync server that says hello every 30 seconds"); + } + regsvc + }); ap.proc()?; Ok(()) } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: examples/procres/mod.rs ================================================================== --- examples/procres/mod.rs +++ examples/procres/mod.rs @@ -9,27 +9,27 @@ } impl Termination for ProcRes { fn report(self) -> ExitCode { match self { - ProcRes::Success => { + Self::Success => { //eprintln!("Process terminated successfully"); ExitCode::from(0) } - ProcRes::Error(e) => { - eprintln!("Abnormal termination: {}", e); + Self::Error(e) => { + eprintln!("Abnormal termination: {e}"); ExitCode::from(1) } } } } impl From> for ProcRes { - fn from(res: Result) -> ProcRes { + fn from(res: Result) -> Self { match res { - Ok(_) => ProcRes::Success, - Err(e) => ProcRes::Error(e) + Ok(_) => Self::Success, + Err(e) => Self::Error(e) } } } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : ADDED examples/simplerocket.rs Index: examples/simplerocket.rs ================================================================== --- /dev/null +++ examples/simplerocket.rs @@ -0,0 +1,138 @@ +#[cfg(all(feature = "rt", feature = "rocket"))] +#[macro_use] +extern crate rocket; + +#[cfg(all(feature = "rt", feature = "rocket"))] +mod gate { + use tokio::{sync::mpsc, task}; + + use qsu::rocket::Rocket; + + use qsu::{ + async_trait, + rt::{ + InitCtx, RocketServiceHandler, RunCtx, RunEnv, SrvAppRt, SvcEvt, TermCtx + }, + tracing + }; + + #[derive(Debug)] + enum MyError {} + + struct MyService { + rx: Option> + } + + #[async_trait] + impl RocketServiceHandler for MyService { + type AppErr = MyError; + + async fn init( + &mut self, + ictx: InitCtx + ) -> Result>, Self::AppErr> { + tracing::info!("Running RocketServiceHandler::init()"); + + ictx.report(Some("Building a rocket".into())); + let rockets = vec![rocket::build().mount("/", routes![index])]; + + ictx.report(Some("Finalized service handler initialization".into())); + + Ok(rockets) + } + + async fn run( + &mut self, + mut rockets: Vec>, + _re: &RunEnv + ) -> Result<(), Self::AppErr> { + tracing::info!("Running RocketServiceHandler::run()"); + + // Spawn a task to handle service events + let mut rx = self.rx.take().unwrap(); + task::spawn(async move { + loop { + if let Some(msg) = rx.recv().await { + tracing::info!("Application received qsu event {:?}", msg); + if let SvcEvt::Shutdown(_) = msg { + break; + } + } + } + }); + + if let Some(rocket) = rockets.pop() { + rocket.launch().await.unwrap(); + } + + Ok(()) + } + + async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr> { + tracing::info!("Running RocketServiceHandler::shutdown()"); + tctx.report(Some("Terminating service handler".into())); + + // .. do termination here .. + + tctx.report(Some("Finalized service handler termination".into())); + + Ok(()) + } + } + + + pub fn main() { + let svcname = qsu::default_service_name().unwrap(); + + // Channel used to signal termination from service event handler to main + // application + let (tx, rx) = mpsc::unbounded_channel(); + + // Service event handler + let svcevt_handler = move |msg| { + let _ = tx.send(msg); + }; + + // Set up main service runtime handler + let svcrt = MyService { rx: Some(rx) }; + + // Define service application runtime components + let apprt = SrvAppRt::Rocket { + svcevt_handler: Box::new(svcevt_handler), + rt_handler: Box::new(svcrt) + }; + + // Launch service + let rctx = RunCtx::new(&svcname); + rctx.run(apprt).unwrap(); + } + + + #[get("/")] + fn index() -> &'static str { + log::error!("error"); + log::warn!("warn"); + log::info!("info"); + log::debug!("debug"); + log::trace!("trace"); + + tracing::error!("error"); + tracing::warn!("warn"); + tracing::info!("info"); + tracing::debug!("debug"); + tracing::trace!("trace"); + + "Hello, world!" + } +} + + +fn main() { + #[cfg(all(feature = "rt", feature = "rocket"))] + gate::main(); + + #[cfg(not(all(feature = "rt", feature = "rocket")))] + println!("simplerocket example requires the rt and rocket features"); +} + +// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : ADDED examples/simplesync.rs Index: examples/simplesync.rs ================================================================== --- /dev/null +++ examples/simplesync.rs @@ -0,0 +1,106 @@ +#[cfg(feature = "rt")] +mod gate { + use std::{sync::mpsc, time::Duration}; + + #[derive(Debug)] + enum MyError {} + + use qsu::{ + rt::{ + Demise, InitCtx, RunCtx, RunEnv, ServiceHandler, SrvAppRt, SvcEvt, + TermCtx, UserSig + }, + tracing + }; + + struct MyService { + rx_term: mpsc::Receiver<()> + } + + impl ServiceHandler for MyService { + type AppErr = MyError; + + fn init(&mut self, _ictx: InitCtx) -> Result<(), Self::AppErr> { + tracing::info!("Running ServiceHandler::init()"); + Ok(()) + } + + fn run(&mut self, _re: &RunEnv) -> Result<(), Self::AppErr> { + tracing::info!("Running ServiceHandler::run()"); + + tracing::info!("Wait for termination event for up to 4 seconds .."); + let _ = self.rx_term.recv_timeout(Duration::from_secs(4)); + + Ok(()) + } + + fn shutdown(&mut self, _tctx: TermCtx) -> Result<(), Self::AppErr> { + tracing::info!("Running ServiceHandler::shutdown()"); + Ok(()) + } + } + + + pub fn main() { + let svcname = qsu::default_service_name().unwrap(); + + // Channel used to signal termination from service event handler to main + // application + let (tx, rx) = mpsc::channel::<()>(); + + // Service event handler + let svcevt_handler = move |msg| match msg { + SvcEvt::Shutdown(demise) => { + tracing::debug!("Shutdown event received"); + match demise { + Demise::Interrupted => { + tracing::info!("Service application was interrupted"); + } + Demise::Terminated => { + tracing::info!("Service application was terminated"); + } + Demise::ReachedEnd => { + tracing::info!("Service application reached its end"); + } + } + + // Once a Shutdown event has been received, signal to the main service + // callback to terminate. + tx.send(()).unwrap(); + } + SvcEvt::User(us) => match us { + UserSig::Sig1 => { + log::info!("User signal 1"); + } + UserSig::Sig2 => { + log::info!("User signal 2"); + } + }, + _ => {} + }; + + // Set up main service runtime handler + let svcrt = MyService { rx_term: rx }; + + // Define service application runtime components + let apprt = SrvAppRt::Sync { + svcevt_handler: Box::new(svcevt_handler), + rt_handler: Box::new(svcrt) + }; + + // Launch service + let rctx = RunCtx::new(&svcname); + rctx.run(apprt).unwrap(); + } +} + + +fn main() { + #[cfg(feature = "rt")] + gate::main(); + + #[cfg(not(feature = "rt"))] + println!("simplesync example requires the rt feature"); +} + +// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : ADDED examples/simpletokio.rs Index: examples/simpletokio.rs ================================================================== --- /dev/null +++ examples/simpletokio.rs @@ -0,0 +1,118 @@ +#[cfg(all(feature = "rt", feature = "tokio"))] +mod gate { + use std::time::Duration; + + use tokio::{sync::mpsc, time}; + + use qsu::{ + async_trait, + rt::{ + InitCtx, RunCtx, RunEnv, SrvAppRt, SvcEvt, TermCtx, TokioServiceHandler, + UserSig + }, + tracing + }; + + #[derive(Debug)] + enum MyError {} + + struct MyService { + rx_term: mpsc::UnboundedReceiver<()> + } + + #[async_trait] + impl TokioServiceHandler for MyService { + type AppErr = MyError; + + async fn init(&mut self, ictx: InitCtx) -> Result<(), Self::AppErr> { + tracing::info!("Running TokioServiceHandler::init()"); + ictx.report(Some("Initializing service handler".into())); + + // .. do initialization here .. + + ictx.report(Some("Finalized service handler initialization".into())); + + Ok(()) + } + + async fn run(&mut self, _re: &RunEnv) -> Result<(), Self::AppErr> { + tracing::info!("Running TokioServiceHandler::run()"); + + tracing::info!("Wait for termination event for up to 4 seconds .."); + tokio::select! { + _ = self.rx_term.recv() => { + tracing::info!("Got kill signal from svc"); + } + () = time::sleep(Duration::from_secs(4)) => { + tracing::info!("Got timeout"); + } + } + + Ok(()) + } + + async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr> { + tracing::info!("Running TokioServiceHandler::shutdown()"); + tctx.report(Some("Terminating service handler".into())); + + // .. do termination here .. + + tctx.report(Some("Finalized service handler termination".into())); + + Ok(()) + } + } + + + pub fn main() { + let svcname = qsu::default_service_name().unwrap(); + + // Channel used to signal termination from service event handler to main + // application + let (tx, rx) = mpsc::unbounded_channel::<()>(); + + // Service event handler + let svcevt_handler = move |msg| match msg { + SvcEvt::Shutdown(_) => { + tracing::debug!("Shutdown event from service subsystem"); + + // Once a Shutdown or Termination event has been received, signal to + // the main service callback to terminate. + let _ = tx.send(()); + } + SvcEvt::User(us) => match us { + UserSig::Sig1 => { + log::info!("User signal 1"); + } + UserSig::Sig2 => { + log::info!("User signal 2"); + } + }, + _ => {} + }; + + // Set up main service runtime handler + let svcrt = MyService { rx_term: rx }; + + // Define service application runtime components + let apprt = SrvAppRt::Tokio { + rtbldr: None, + svcevt_handler: Box::new(svcevt_handler), + rt_handler: Box::new(svcrt) + }; + + // Launch service + let rctx = RunCtx::new(&svcname); + rctx.run(apprt).unwrap(); + } +} + +fn main() { + #[cfg(all(feature = "rt", feature = "tokio"))] + gate::main(); + + #[cfg(not(all(feature = "rt", feature = "tokio")))] + println!("simpletokio example requires the rt and tokio features"); +} + +// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/argp.rs ================================================================== --- src/argp.rs +++ src/argp.rs @@ -30,11 +30,11 @@ //! callbacks. use clap::{builder::Str, Arg, ArgAction, ArgMatches, Args, Command}; use crate::{ - err::{AppErr, Error}, + err::CbErr, installer::{self, RegSvc}, lumberjack::LogLevel, rt::{RunCtx, SrvAppRt} }; @@ -41,17 +41,18 @@ /// Modify a `clap` [`Command`] instance to accept common service management /// subcommands. /// /// If `inst_subcmd` is `Some()`, it should be the name of the subcommand used -/// to register a service. If `rm_subcmd_ is `Some()` it should be the name of +/// to register a service. If `rm_subcmd` is `Some()` it should be the name of /// the subcommand used to deregister a service. Similarly `run_subcmd` is /// used to add a subcommand for running the service under a service subsystem /// [where applicable]. /// /// It is recommended that applications use the higher-level `ArgParser` /// instead. +#[must_use] pub fn add_subcommands( cli: Command, svcname: &str, inst_subcmd: Option<&str>, rm_subcmd: Option<&str>, @@ -109,23 +110,33 @@ workdir: Option, #[arg(long, value_enum, value_name = "LEVEL")] log_level: Option, - #[arg(long, value_enum, hide(true), value_name = "LEVEL")] - trace_level: Option, + #[arg(long, hide(true), value_name = "FILTER")] + trace_filter: Option, #[arg(long, value_enum, hide(true), value_name = "FNAME")] - trace_file: Option + trace_file: Option, + + /// Forcibly install service. + /// + /// On Windows, attempt to stop and uninstall service if it already exists. + /// + /// On macos/launchd and linux/systemd, overwrite the service specification + /// file if it already exists. + #[arg(long, short)] + force: bool } /// Create a `clap` [`Command`] object that accepts service registration /// arguments. /// /// It is recommended that applications use the higher-level `ArgParser` /// instead, but this call exists in case applications need finer grained /// control. +#[must_use] pub fn mk_inst_cmd(cmd: &str, svcname: &str) -> Command { let namearg = Arg::new("svcname") .short('n') .long("name") .action(ArgAction::Set) @@ -148,10 +159,11 @@ /// arguments. /// /// It is recommended that applications use the higher-level `ArgParser` /// instead, but this call exists in case applications need finer grained /// control. +#[must_use] pub fn mk_rm_cmd(cmd: &str, svcname: &str) -> Command { let namearg = Arg::new("svcname") .short('n') .long("name") .action(ArgAction::Set) @@ -169,11 +181,17 @@ pub struct DeregSvc { pub svcname: String } impl DeregSvc { + #[allow( + clippy::missing_panics_doc, + reason = "svcname should always have been set" + )] + #[must_use] pub fn from_cmd_match(matches: &ArgMatches) -> Self { + // unwrap() should be okay here, because svcname should always be set. let svcname = matches.get_one::("svcname").unwrap().to_owned(); Self { svcname } } } @@ -186,10 +204,11 @@ /// Create a `clap` [`Command`] object that accepts service running arguments. /// /// It is recommended that applications use the higher-level `ArgParser` /// instead, but this call exists in case applications need finer grained /// control. +#[must_use] pub fn mk_run_cmd(cmd: &str, svcname: &str) -> Command { let namearg = Arg::new("svcname") .short('n') .long("name") .action(ArgAction::Set) @@ -201,13 +220,13 @@ RunSvcArgs::augment_args(cli) } -pub(crate) enum ArgpRes<'cb> { +pub(crate) enum ArgpRes<'cb, ApEr> { /// Run server application. - RunApp(RunCtx, &'cb mut dyn ArgsProc), + RunApp(RunCtx, &'cb mut dyn ArgsProc), /// Nothing to do (service was probably registered/deregistred). Quit } @@ -216,10 +235,15 @@ pub struct RunSvc { pub svcname: String } impl RunSvc { + #[allow( + clippy::missing_panics_doc, + reason = "svcname should always have been set" + )] + #[must_use] pub fn from_cmd_match(matches: &ArgMatches) -> Self { let svcname = matches.get_one::("svcname").unwrap().to_owned(); Self { svcname } } } @@ -234,21 +258,27 @@ Run } /// Allow application to customise behavior of an [`ArgParser`] instance. pub trait ArgsProc { + type AppErr; + /// Give the application an opportunity to modify the root and subcommand /// `Command`s. /// /// `cmdtype` indicates whether `cmd` is the root `Command` or one of the /// subcommand `Command`s. + /// + /// # Errors + /// Application-defined error will be returned as `CbErr::Aop` to the + /// original caller. #[allow(unused_variables)] fn conf_cmd( &mut self, cmdtype: Cmd, cmd: Command - ) -> Result { + ) -> Result { Ok(cmd) } /// Callback allowing application to configure the service registration /// context just before the service is registered. @@ -261,27 +291,43 @@ /// The `sub_m` argument represents `clap`'s parsed subcommand context for /// the service registration subcommand. Applications that want to add /// custom arguments to the parser should implement the /// [`ArgsProc::conf_cmd()`] trait method and perform the subcommand /// augmentation there. + /// + /// This callback is called after the system-defined command line arguments + /// have been parsed and populated into `regsvc`. Implementation should be + /// careful not to overwrite settings that should be configurable via the + /// command line, and can inspect the current value of fields before setting + /// them if conditional reconfiguations are required. /// /// The default implementation does nothing but return `regsvc` unmodified. + /// + /// # Errors + /// Application-defined error will be returned as `CbErr::Aop` to the + /// original caller. #[allow(unused_variables)] fn proc_inst( &mut self, sub_m: &ArgMatches, regsvc: RegSvc - ) -> Result { + ) -> Result { Ok(regsvc) } + /// Callback allowing application to configure the service deregistration + /// context just before the service is deregistered. + /// + /// # Errors + /// Application-defined error will be returned as `CbErr::Aop` to the + /// original caller. #[allow(unused_variables)] fn proc_rm( &mut self, sub_m: &ArgMatches, deregsvc: DeregSvc - ) -> Result { + ) -> Result { Ok(deregsvc) } /// Callback allowing application to configure the run context before /// launching the server application. @@ -289,32 +335,44 @@ /// qsu will have performed all its own initialization of the [`RunCtx`] /// before calling this function. /// /// The application can differentiate between running in a service mode and /// running as a foreground by calling [`RunCtx::is_service()`]. + /// + /// # Errors + /// Application-defined error will be returned as `CbErr::Aop` to the + /// original caller. #[allow(unused_variables)] fn proc_run( &mut self, matches: &ArgMatches, runctx: RunCtx - ) -> Result { + ) -> Result { Ok(runctx) } /// Called when a subcommand is encountered that is not one of the three /// subcommands regognized by qsu. + /// + /// # Errors + /// Application-defined error will be returned as `CbErr::Aop` to the + /// original caller. #[allow(unused_variables)] fn proc_other( &mut self, subcmd: &str, sub_m: &ArgMatches - ) -> Result<(), AppErr> { + ) -> Result<(), Self::AppErr> { Ok(()) } /// Construct an server application runtime. - fn build_apprt(&mut self) -> Result; + /// + /// # Errors + /// Application-defined error will be returned as `CbErr::Aop` to the + /// original caller. + fn build_apprt(&mut self) -> Result, Self::AppErr>; } /// High-level argument parser. /// @@ -323,33 +381,38 @@ /// - Registering a service /// - Deregistering a service /// - Running as a service /// - Running without any subcommands should run the server application as a /// foreground process. -pub struct ArgParser<'cb> { +pub struct ArgParser<'cb, ApEr> { svcname: String, reg_subcmd: String, dereg_subcmd: String, run_subcmd: String, cli: Command, - cb: &'cb mut dyn ArgsProc + cb: &'cb mut dyn ArgsProc, + regcb: Option RegSvc>> } -impl<'cb> ArgParser<'cb> { +impl<'cb, ApEr> ArgParser<'cb, ApEr> +where + ApEr: Send + 'static +{ /// Create a new argument parser. /// /// `svcname` is the _default_ service name. It may be overridden using /// command line arguments. - pub fn new(svcname: &str, cb: &'cb mut dyn ArgsProc) -> Self { + pub fn new(svcname: &str, cb: &'cb mut dyn ArgsProc) -> Self { let cli = Command::new(""); Self { svcname: svcname.to_string(), reg_subcmd: "register-service".into(), dereg_subcmd: "deregister-service".into(), run_subcmd: "run-service".into(), cli, - cb + cb, + regcb: None } } /// Create a new argument parser, basing the root command parser on an @@ -358,51 +421,51 @@ /// `svcname` is the _default_ service name. It may be overridden using /// command line arguments. pub fn with_cmd( svcname: &str, cli: Command, - cb: &'cb mut dyn ArgsProc + cb: &'cb mut dyn ArgsProc ) -> Self { Self { svcname: svcname.to_string(), reg_subcmd: "register-service".into(), dereg_subcmd: "deregister-service".into(), run_subcmd: "run-service".into(), cli, - cb + cb, + regcb: None } } /// Rename the service registration subcommand. + #[must_use] pub fn reg_subcmd(mut self, nm: &str) -> Self { self.reg_subcmd = nm.to_string(); self } /// Rename the service deregistration subcommand. + #[must_use] pub fn dereg_subcmd(mut self, nm: &str) -> Self { self.dereg_subcmd = nm.to_string(); self } /// Rename the subcommand for running the service. + #[must_use] pub fn run_subcmd(mut self, nm: &str) -> Self { self.run_subcmd = nm.to_string(); self } - fn inner_proc(self) -> Result, Error> { + fn inner_proc(self) -> Result, CbErr> { let matches = match self.cli.try_get_matches() { Ok(m) => m, Err(e) => match e.kind() { - clap::error::ErrorKind::DisplayHelp => { + clap::error::ErrorKind::DisplayHelp + | clap::error::ErrorKind::DisplayVersion => { e.exit(); - //return Ok(ArgpRes::Quit); - } - clap::error::ErrorKind::DisplayVersion => { - e.exit(); - //return Ok(ArgpRes::Quit); } _ => { // ToDo: Convert error to Error::ArgP, pass along the error type so // that the Termination handler can output the specific error. //Err(e)?; @@ -412,14 +475,21 @@ }; match matches.subcommand() { Some((subcmd, sub_m)) if subcmd == self.reg_subcmd => { //println!("{:#?}", sub_m); + // Convert known register-service command line arguments to a RegSvc + // context. let mut regsvc = RegSvc::from_cmd_match(sub_m); // To trigger the server to run in service mode, run with the - // subcommand "run-service". + // subcommand "run-service" (or whatever it has been changed to). + // + // We're making a pretty safe assumption that the service will + // be run though qsu's argument processor because it is being + // registered using it. + // // If the service name is different that the name drived from the // executable's name, then add "--name " arguments. let mut args = vec![String::from(&self.run_subcmd)]; if regsvc.svcname() != self.svcname { args.push(String::from("--name")); @@ -427,13 +497,28 @@ } regsvc.args_ref(args); // Call application call-back, to allow application-specific // service configuration. + // // This is a good place to stick custom environment, arguments, - // registry changes. - let regsvc = self.cb.proc_inst(sub_m, regsvc)?; + // registry changes that may depend on command line arguments. + let regsvc = self + .cb + .proc_inst(sub_m, regsvc) + .map_err(|ae| CbErr::App(ae))?; + + // Give the application a final chance to modify the service + // registration parameters. + // + // This can be used to set hardcoded parameters like service display + // name, description, etc. + let regsvc = if let Some(cb) = self.regcb { + cb(regsvc) + } else { + regsvc + }; // Register the service with the operating system's service subsystem. regsvc.register()?; Ok(ArgpRes::Quit) @@ -440,11 +525,12 @@ } Some((subcmd, sub_m)) if subcmd == self.dereg_subcmd => { // Get arguments relating to service deregistration. let args = DeregSvc::from_cmd_match(sub_m); - let args = self.cb.proc_rm(sub_m, args)?; + let args = + self.cb.proc_rm(sub_m, args).map_err(|ae| CbErr::App(ae))?; installer::uninstall(&args.svcname)?; Ok(ArgpRes::Quit) } @@ -453,33 +539,68 @@ let args = RunSvc::from_cmd_match(sub_m); // Create a RunCtx, mark it as a service, and allow application the // opportunity to modify it based on the parsed command line. let rctx = RunCtx::new(&args.svcname).service(); - let rctx = self.cb.proc_run(sub_m, rctx)?; + let rctx = + self.cb.proc_run(sub_m, rctx).map_err(|ae| CbErr::App(ae))?; // Return a run context for a background service process. Ok(ArgpRes::RunApp(rctx, self.cb)) } Some((subcmd, sub_m)) => { // Call application callback for processing "other" subcmd - self.cb.proc_other(subcmd, sub_m)?; + self + .cb + .proc_other(subcmd, sub_m) + .map_err(|ae| CbErr::App(ae))?; // Return a run context for a background service process. Ok(ArgpRes::Quit) } _ => { // Create a RunCtx, mark it as a service, and allow application the // opportunity to modify it based on the parsed command line. let rctx = RunCtx::new(&self.svcname); - let rctx = self.cb.proc_run(&matches, rctx)?; + let rctx = self + .cb + .proc_run(&matches, rctx) + .map_err(|ae| CbErr::App(ae))?; // Return a run context for a foreground process. Ok(ArgpRes::RunApp(rctx, self.cb)) } } } + + /// Register a closure that will be called before service registration. + /// + /// This callback serves a different purpose than implementing the + /// [`ArgsProc::proc_inst()`] method, which is primarily meant to tranlate + /// command line arguments to service registration parameters. + /// + /// The `regsvc_proc()` callback on the other hand is called just before + /// actual registration and is intended to overwrite service registration + /// parametmer. It is suitable to use this callback to set hard-coded + /// application-specific service parameters, like service display name, + /// description, dependencies, and other parameters that the user should + /// not need to specify manually. + /// + /// Just as with `ArgsProc::proc_inst()` this callback is called after the + /// system-defined command line arguments have been parsed and populated + /// into the [`RegSvc`] context. Closures should be careful not to overwrite + /// settings that should be configurable via the command line, and can + /// inspect the current value of fields before setting them if conditional + /// reconfiguations are required. + #[must_use] + pub fn regsvc_proc( + mut self, + f: impl FnOnce(RegSvc) -> RegSvc + 'static + ) -> Self { + self.regcb = Some(Box::new(f)); + self + } /// Process command line arguments. /// /// Calling this method will initialize the command line parser, parse the /// command line, using the associated [`ArgsProc`] as appropriate to modify @@ -507,41 +628,57 @@ /// arguments. /// - If the specified service name is different than the default service /// name (determined by /// [`default_service_name()`](crate::default_service_name), then the /// aguments `--name ` will be added. - pub fn proc(mut self) -> Result<(), Error> { + /// + /// # Errors + /// Application-defined error will be returned as `CbErr::Aop` to the + /// original caller. + pub fn proc(mut self) -> Result<(), CbErr> { // Give application the opportunity to modify root Command - self.cli = self.cb.conf_cmd(Cmd::Root, self.cli)?; + self.cli = self + .cb + .conf_cmd(Cmd::Root, self.cli) + .map_err(|ae| CbErr::App(ae))?; // Create registration subcommand and give application the opportunity to // modify the subcommand's Command let sub = mk_inst_cmd(&self.reg_subcmd, &self.svcname); - let sub = self.cb.conf_cmd(Cmd::Inst, sub)?; + let sub = self + .cb + .conf_cmd(Cmd::Inst, sub) + .map_err(|ae| CbErr::App(ae))?; self.cli = self.cli.subcommand(sub); // Create deregistration subcommand let sub = mk_rm_cmd(&self.dereg_subcmd, &self.svcname); - let sub = self.cb.conf_cmd(Cmd::Rm, sub)?; + let sub = self + .cb + .conf_cmd(Cmd::Rm, sub) + .map_err(|ae| CbErr::App(ae))?; self.cli = self.cli.subcommand(sub); // Create run subcommand let sub = mk_run_cmd(&self.run_subcmd, &self.svcname); - let sub = self.cb.conf_cmd(Cmd::Run, sub)?; + let sub = self + .cb + .conf_cmd(Cmd::Run, sub) + .map_err(|ae| CbErr::App(ae))?; self.cli = self.cli.subcommand(sub); // Parse command line arguments. Run the service application if requiested // to do so. if let ArgpRes::RunApp(runctx, cb) = self.inner_proc()? { // Argument parser asked us to run, so call the application to ask it to // create the service handler, and then kick off the service runtime. //let st = bldr(ctx)?; - let st = cb.build_apprt()?; + let st = cb.build_apprt().map_err(|ae| CbErr::App(ae))?; runctx.run(st)?; } Ok(()) } } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/err.rs ================================================================== --- src/err.rs +++ src/err.rs @@ -1,49 +1,121 @@ use std::{fmt, io}; -pub use apperr::AppErr; - #[derive(Debug)] pub enum ArgsError { #[cfg(feature = "clap")] Clap(clap::Error), Msg(String) } - /// Indicate failure for server applicatino callbacks. #[derive(Debug, Default)] -pub struct AppErrors { - pub init: Option, - pub run: Option, - pub shutdown: Option +pub struct AppErrors { + pub init: Option, + pub run: Option, + pub shutdown: Option } -impl AppErrors { - pub fn init_failed(&self) -> bool { +impl AppErrors { + pub const fn init_failed(&self) -> bool { self.init.is_some() } - pub fn run_failed(&self) -> bool { + pub const fn run_failed(&self) -> bool { self.run.is_some() } - pub fn shutdown_failed(&self) -> bool { + pub const fn shutdown_failed(&self) -> bool { self.shutdown.is_some() } } + +/// Errors that can be returned from functions that call application callbacks. +#[derive(Debug)] +#[allow(clippy::module_name_repetitions)] +pub enum CbErr { + /// An qsu library error was generated. + Lib(Error), + + /// Application-defined error. + /// + /// Applications can use this variant to pass application-specific errors + /// through the runtime back to itself. + App(ApEr), + + /// Returned by [`RunCtx::run()`](crate::rt::RunCtx) to indicate which + /// server application callbacks that failed. + #[cfg(feature = "rt")] + #[cfg_attr(docsrs, doc(cfg(feature = "rt")))] + SrvApp(AppErrors) +} + +impl std::error::Error for CbErr {} + +impl fmt::Display for CbErr { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match self { + Self::Lib(e) => e.fmt(f), + Self::App(_ae) => { + // ToDo: Add ApErr: Error bound and forward the call to it + write!(f, "Application-defined error") + } + #[cfg(feature = "rt")] + Self::SrvApp(e) => { + let mut v = vec![]; + if e.init.is_some() { + v.push("init"); + } + if e.run.is_some() { + v.push("run"); + } + if e.shutdown.is_some() { + v.push("shutdown"); + } + write!(f, "Server application failed [{}]", v.join(",")) + } + } + } +} + +impl CbErr { + /// Extract application-specific error. + /// + /// # Panics + /// The error must be [`CbErr::App`] of this method will panic. + pub fn unwrap_apperr(self) -> ApEr { + let Self::App(ae) = self else { + panic!("Not CbErr::App()"); + }; + ae + } +} + +impl From for CbErr { + fn from(err: Error) -> Self { + Self::Lib(err) + } +} + +#[cfg(feature = "rocket")] +impl From for CbErr { + fn from(err: rocket::Error) -> Self { + Self::Lib(Error::Rocket(err.to_string())) + } +} + +impl From for CbErr { + fn from(err: std::io::Error) -> Self { + Self::Lib(Error::IO(err.to_string())) + } +} + /// Errors that qsu will return to application. #[derive(Debug)] pub enum Error { - /// Application-defined error. - /// - /// Applications can use this variant to pass application-specific errors - /// through the runtime back to itself. - App(AppErr), - ArgP(ArgsError), BadFormat(String), Internal(String), IO(String), @@ -62,125 +134,56 @@ #[cfg(feature = "rocket")] #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))] Rocket(String), SubSystem(String), - /// Returned by [`RunCtx::run()`](crate::rt::RunCtx) to indicate which - /// server application callbacks that failed. - #[cfg(feature = "rt")] - #[cfg_attr(docsrs, doc(cfg(feature = "rt")))] - SrvApp(AppErrors), - Unsupported } +#[allow(clippy::needless_pass_by_value)] impl Error { - pub fn is_apperr(&self) -> bool { - matches!(self, Error::App(_)) - } - - /// Attempt to convert [`Error`] into application-specific error. - /// - /// If it's not an `Error::App()` nor can be downcast to type `E`, the error - /// will be returned back as an `Err()`. - pub fn try_into_apperr(self) -> Result - where - E: std::error::Error + apperr::Blessed + 'static - { - match self { - Error::App(e) => match e.try_into_inner::() { - Ok(e) => Ok(e), - Err(e) => Err(Error::App(e)) - }, - e => Err(e) - } - } - - /// Unwrap application-specific error from an [`Error`]. - /// - /// # Panic - /// Panics if `Error` variant is not `Error::App()`. - pub fn unwrap_apperr(self) -> E - where - E: std::error::Error + apperr::Blessed + 'static - { - let Ok(e) = self.try_into_apperr::() else { - panic!("Unable to unwrap error E"); - }; - e - } - - pub fn bad_format(s: S) -> Self { - Error::BadFormat(s.to_string()) - } - - pub fn internal(s: S) -> Self { - Error::Internal(s.to_string()) - } - - pub fn io(s: S) -> Self { - Error::IO(s.to_string()) - } - - pub fn lumberjack(s: S) -> Self { - Error::LumberJack(s.to_string()) - } - - pub fn missing(s: S) -> Self { - Error::Missing(s.to_string()) - } -} + pub fn bad_format(s: impl ToString) -> Self { + Self::BadFormat(s.to_string()) + } + + pub fn internal(s: impl ToString) -> Self { + Self::Internal(s.to_string()) + } + + pub fn io(s: impl ToString) -> Self { + Self::IO(s.to_string()) + } + + pub fn lumberjack(s: impl ToString) -> Self { + Self::LumberJack(s.to_string()) + } + + pub fn missing(s: impl ToString) -> Self { + Self::Missing(s.to_string()) + } +} + impl std::error::Error for Error {} impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match self { - Error::App(_e) => { - write!(f, "Application-defined error") - } - Error::ArgP(s) => { - // ToDo: Handle the ArgsError::Clap and ArgsError::Msg differently - write!(f, "Argument parser; {:?}", s) - } - Error::BadFormat(s) => { - write!(f, "Bad format error; {}", s) - } - Error::Internal(s) => { - write!(f, "Internal error; {}", s) - } - Error::IO(s) => { - write!(f, "I/O error; {}", s) - } - Error::LumberJack(s) => { - write!(f, "LumberJack error; {}", s) - } - Error::Missing(s) => { - write!(f, "Missing data; {}", s) - } + Self::ArgP(s) => { + // ToDo: Handle the ArgsError::Clap and ArgsError::Msg differently + write!(f, "Argument parser; {s:?}") + } + Self::BadFormat(s) => write!(f, "Bad format error; {s}"), + Self::Internal(s) => write!(f, "Internal error; {s}"), + Self::IO(s) => write!(f, "I/O error; {s}"), + Self::LumberJack(s) => write!(f, "LumberJack error; {s}"), + Self::Missing(s) => write!(f, "Missing data; {s}"), #[cfg(feature = "rocket")] - Error::Rocket(s) => { - write!(f, "Rocket error; {}", s) - } - Error::SubSystem(s) => { - write!(f, "Service subsystem error; {}", s) - } - Error::SrvApp(e) => { - let mut v = vec![]; - if e.init.is_some() { - v.push("init"); - } - if e.run.is_some() { - v.push("run"); - } - if e.shutdown.is_some() { - v.push("shutdown"); - } - write!(f, "Server application failed [{}]", v.join(",")) - } - Error::Unsupported => { + Self::Rocket(s) => write!(f, "Rocket error; {s}"), + Self::SubSystem(s) => write!(f, "Service subsystem error; {s}"), + Self::Unsupported => { write!(f, "Operation is unsupported [on this platform]") } } } } @@ -197,59 +200,62 @@ #[cfg(windows)] impl From for Error { /// Map eventlog initialization errors to `Error::LumberJack`. fn from(err: eventlog::InitError) -> Self { - Error::LumberJack(err.to_string()) + Self::LumberJack(err.to_string()) } } #[cfg(windows)] impl From for Error { /// Map eventlog errors to `Error::LumberJack`. fn from(err: eventlog::Error) -> Self { - Error::LumberJack(err.to_string()) + Self::LumberJack(err.to_string()) } } impl From for Error { fn from(err: io::Error) -> Self { - Error::IO(err.to_string()) + Self::IO(err.to_string()) } } #[cfg(windows)] impl From for Error { fn from(err: registry::key::Error) -> Self { - Error::SubSystem(err.to_string()) + Self::SubSystem(err.to_string()) } } #[cfg(feature = "rocket")] impl From for Error { fn from(err: rocket::Error) -> Self { - Error::Rocket(err.to_string()) + Self::Rocket(err.to_string()) } } #[cfg(feature = "installer")] impl From for Error { fn from(err: sidoc::Error) -> Self { - Error::Internal(err.to_string()) + Self::Internal(err.to_string()) } } #[cfg(windows)] impl From for Error { fn from(err: windows_service::Error) -> Self { - Error::SubSystem(err.to_string()) + Self::SubSystem(err.to_string()) } } -impl From for Error { + +/* +impl From for Error { /// Wrap an [`AppErr`] in an [`Error`]. - fn from(err: AppErr) -> Self { + fn from(err: ApEr) -> Self { Error::App(err) } } +*/ // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/installer.rs ================================================================== --- src/installer.rs +++ src/installer.rs @@ -19,10 +19,11 @@ use clap::ArgMatches; use itertools::Itertools; use crate::{err::Error, lumberjack::LogLevel}; + /* #[cfg(any( target_os = "macos", all(target_os = "linux", feature = "systemd") @@ -145,22 +146,49 @@ all(target_os = "linux", feature = "systemd") ))] umask: Option } + +#[cfg(windows)] +pub type BoxRegCb = + Box Result<(), Error>>; + + +#[allow(clippy::struct_excessive_bools)] pub struct RegSvc { + /// If `true`, then attempt to forcibly install service. + pub force: bool, + + /// Set to `true` if this service uses the qsu service argument parser. + /// + /// This will ensure that `run-service` is the first argument passed to the + /// service executable. + pub qsu_argp: bool, + pub svcname: String, - #[cfg(windows)] + /// Service's display name. + /// + /// Only used on Windows. pub display_name: Option, - #[cfg(any(windows, all(target_os = "linux", feature = "systemd")))] + /// Service's description. + /// + /// Only used on Windows and on linux/systemd. pub description: Option, + /// Set to `true` if service supports configuration reloading. + pub conf_reload: bool, + + /// Set to `true` if this is a network service. + /// + /// Note that this does not magically solve startup dependencies. + pub netservice: bool, + #[cfg(windows)] - pub regconf: - Option Result<(), Error>>>, + pub regconf: Option, /// Command line arguments. pub args: Vec, /// Environment variables. @@ -177,11 +205,11 @@ /// List of service dependencies. deps: Vec, log_level: Option, - trace_level: Option, + trace_filter: Option, trace_file: Option, runas: RunAs } @@ -190,19 +218,26 @@ Network, Custom(Vec) } impl RegSvc { + #[must_use] pub fn new(svcname: &str) -> Self { Self { + force: false, + + qsu_argp: false, + svcname: svcname.to_string(), - #[cfg(windows)] display_name: None, - #[cfg(any(windows, all(target_os = "linux", feature = "systemd")))] description: None, + + conf_reload: false, + + netservice: false, #[cfg(windows)] regconf: None, args: Vec::new(), @@ -215,236 +250,334 @@ deps: Vec::new(), log_level: None, - trace_level: None, + trace_filter: None, trace_file: None, runas: RunAs::default() } } #[cfg(feature = "clap")] + #[allow(clippy::missing_panics_doc)] pub fn from_cmd_match(matches: &ArgMatches) -> Self { + let force = matches.get_flag("force"); + + // unwrap should be okay, because svcname is mandatory let svcname = matches.get_one::("svcname").unwrap().to_owned(); let autostart = matches.get_flag("auto_start"); - #[cfg(windows)] + let dispname = matches.get_one::("display_name"); - #[cfg(any(windows, all(target_os = "linux", feature = "systemd")))] let descr = matches.get_one::("description"); - let args: Vec = if let Some(vr) = matches.get_many::("arg") + let args: Vec = matches + .get_many::("arg") + .map_or_else(Vec::new, |vr| vr.map(String::from).collect()); + + let envs: Vec = matches + .get_many::("env") + .map_or_else(Vec::new, |vr| vr.map(String::from).collect()); + + /* + if let Some(vr) = matches.get_many::("env") { vr.map(String::from).collect() } else { Vec::new() }; - let envs: Vec = if let Some(vr) = matches.get_many::("env") - { - vr.map(String::from).collect() - } else { - Vec::new() - }; + */ let workdir = matches.get_one::("workdir"); let mut environ = Vec::new(); let mut it = envs.into_iter(); while let Some((key, value)) = it.next_tuple() { environ.push((key, value)); } let log_level = matches.get_one::("log_level").copied(); - let trace_level = matches.get_one::("trace_level").copied(); + let trace_filter = matches.get_one::("trace_filter").cloned(); let trace_file = matches.get_one::("trace_file").cloned(); let runas = RunAs::default(); Self { + force, + qsu_argp: true, svcname, - #[cfg(windows)] display_name: dispname.cloned(), - #[cfg(any(windows, all(target_os = "linux", feature = "systemd")))] description: descr.cloned(), + conf_reload: false, + netservice: false, #[cfg(windows)] regconf: None, - args: args.to_vec(), + args, envs: environ, autostart, workdir: workdir.cloned(), deps: Vec::new(), log_level, - trace_level, + trace_filter, trace_file, runas } } + #[must_use] pub fn svcname(&self) -> &str { &self.svcname } - #[cfg(windows)] + /// Set the service's display name. + /// + /// This only has an effect on Windows. + #[must_use] pub fn display_name(mut self, name: impl ToString) -> Self { + self.display_name_ref(name); + self + } + + /// Set the service's _display name_. + /// + /// This only has an effect on Windows. + #[allow(clippy::needless_pass_by_value)] + pub fn display_name_ref(&mut self, name: impl ToString) -> &mut Self { self.display_name = Some(name.to_string()); self } - #[cfg(any(windows, all(target_os = "linux", feature = "systemd")))] + /// Set the service's description. + /// + /// This only has an effect on Windows and linux/systemd. + #[must_use] pub fn description(mut self, text: impl ToString) -> Self { - self.description = Some(text.to_string()); + self.description_ref(text); self } - #[cfg(any(windows, all(target_os = "linux", feature = "systemd")))] + /// Set the service's description. + /// + /// This only has an effect on Windows and linux/systemd. + #[allow(clippy::needless_pass_by_value)] pub fn description_ref(&mut self, text: impl ToString) -> &mut Self { self.description = Some(text.to_string()); self } + /// Mark service as able to live reload its configuration. + #[must_use] + pub fn conf_reload(mut self) -> Self { + self.conf_reload_ref(); + self + } + + /// Mark service as able to live reload its configuration. + pub fn conf_reload_ref(&mut self) -> &mut Self { + self.conf_reload = true; + self + } + + /// Mark service as a network application. + /// + /// # Windows + /// Calling this will implicitly add a `Tcpip` service dependency. + #[must_use] + pub fn netservice(mut self) -> Self { + self.netservice_ref(); + self + } + + /// Mark service as a network application. + /// + /// # Windows + /// Calling this will implicitly add a `Tcpip` service dependency. + pub fn netservice_ref(&mut self) -> &mut Self { + self.netservice = true; + + #[cfg(windows)] + self.deps.push(Depend::Network); + + self + } + + /// Register a callback that will be used to set service registry keys. #[cfg(windows)] + #[cfg_attr(docsrs, doc(cfg(windows)))] + #[must_use] pub fn regconf(mut self, f: F) -> Self where F: FnOnce(&str, &mut winreg::RegKey) -> Result<(), Error> + 'static { self.regconf = Some(Box::new(f)); self } + /// Register a callback that will be used to set service registry keys. #[cfg(windows)] + #[cfg_attr(docsrs, doc(cfg(windows)))] pub fn regconf_ref(&mut self, f: F) -> &mut Self where F: FnOnce(&str, &mut winreg::RegKey) -> Result<(), Error> + 'static { self.regconf = Some(Box::new(f)); self } - pub fn arg(mut self, arg: S) -> Self - where - S: ToString - { + /// Append a service command line argument. + #[allow(clippy::needless_pass_by_value)] + #[must_use] + pub fn arg(mut self, arg: impl ToString) -> Self { self.args.push(arg.to_string()); self } - pub fn arg_ref(&mut self, arg: S) -> &mut Self - where - S: ToString - { + /// Append a service command line argument. + #[allow(clippy::needless_pass_by_value)] + pub fn arg_ref(&mut self, arg: impl ToString) -> &mut Self { self.args.push(arg.to_string()); self } + /// Append service command line arguments. + #[must_use] pub fn args(mut self, args: I) -> Self where I: IntoIterator, S: ToString { - for arg in args.into_iter() { + for arg in args { self.args.push(arg.to_string()); } self } + /// Append service command line arguments. pub fn args_ref(&mut self, args: I) -> &mut Self where I: IntoIterator, S: ToString { - for arg in args.into_iter() { + for arg in args { self.arg_ref(arg.to_string()); } self } + #[must_use] pub fn have_args(&self) -> bool { !self.args.is_empty() } + /// Add a service environment variable. + #[allow(clippy::needless_pass_by_value)] + #[must_use] pub fn env(mut self, key: K, val: V) -> Self where K: ToString, V: ToString { self.envs.push((key.to_string(), val.to_string())); self } + /// Add a service environment variable. + #[allow(clippy::needless_pass_by_value)] pub fn env_ref(&mut self, key: K, val: V) -> &mut Self where K: ToString, V: ToString { self.envs.push((key.to_string(), val.to_string())); self } + /// Add service environment variables. + #[must_use] pub fn envs(mut self, envs: I) -> Self where I: IntoIterator, K: ToString, V: ToString { - for (key, val) in envs.into_iter() { + for (key, val) in envs { self.envs.push((key.to_string(), val.to_string())); } self } + /// Add service environment variables. pub fn envs_ref(&mut self, args: I) -> &mut Self where I: IntoIterator, K: ToString, V: ToString { - for (key, val) in args.into_iter() { + for (key, val) in args { self.env_ref(key.to_string(), val.to_string()); } self } + #[must_use] pub fn have_envs(&self) -> bool { !self.envs.is_empty() } - pub fn autostart(mut self) -> Self { + /// Mark service to auto-start on boot. + #[must_use] + pub const fn autostart(mut self) -> Self { self.autostart = true; self } + /// Mark service to auto-start on boot. pub fn autostart_ref(&mut self) -> &mut Self { self.autostart = true; self } /// Sets the work directory that the service should start in. /// /// This is a utf-8 string rather than a `Path` or `PathBuf` because the /// directory tends to end up in places that have an utf-8 constraint. + #[allow(clippy::needless_pass_by_value)] + #[must_use] pub fn workdir(mut self, workdir: impl ToString) -> Self { self.workdir = Some(workdir.to_string()); self } /// In-place version of [`Self::workdir()`]. + #[allow(clippy::needless_pass_by_value)] pub fn workdir_ref(&mut self, workdir: impl ToString) -> &mut Self { self.workdir = Some(workdir.to_string()); self } + /// Add a service dependency. + /// + /// Has no effect on macos. + #[must_use] pub fn depend(mut self, dep: Depend) -> Self { self.deps.push(dep); self } + /// Add a service dependency. + /// + /// Has no effect on macos. pub fn depend_ref(&mut self, dep: Depend) -> &mut Self { self.deps.push(dep); self } + /// Perform the service registration. + /// + /// # Errors + /// The error may be system/service subsystem specific. pub fn register(self) -> Result<(), Error> { #[cfg(windows)] winsvc::install(self)?; #[cfg(target_os = "macos")] @@ -456,10 +589,14 @@ Ok(()) } } +/// Deregister a service from a service subsystem. +/// +/// # Errors +/// The error may be system/service subsystem specific. #[allow(unreachable_code)] pub fn uninstall(svcname: &str) -> Result<(), Error> { #[cfg(windows)] { winsvc::uninstall(svcname)?; Index: src/installer/launchd.rs ================================================================== --- src/installer/launchd.rs +++ src/installer/launchd.rs @@ -1,67 +1,72 @@ -use std::{fs, path::Path, sync::Arc}; +use std::{fs, io::ErrorKind, path::Path, sync::Arc}; use sidoc::{Builder, RenderContext}; -use crate::err::Error; +use super::Error; +/// Generate a plist file for running service under launchd. +/// +/// # Errors +/// [`Error::IO`] will be returned if file I/O failed when trying to generate +/// plist file. pub fn install(ctx: super::RegSvc) -> Result<(), Error> { let mut bldr = Builder::new(); bldr.line(r#""#); bldr.line(r#""#); bldr.scope(r#""#, Some("")); bldr.scope("", Some("" for now - bldr.line(r#"Label"#); + bldr.line("Label"); bldr.line(format!("{}", ctx.svcname())); let service_binary_path = ::std::env::current_exe()? .to_str() - .ok_or(Error::bad_format("Executable pathname is not utf-8"))? + .ok_or_else(|| Error::bad_format("Executable pathname is not utf-8"))? .to_string(); if let Some(ref username) = ctx.runas.user { - bldr.line(r#"UserName"#); - bldr.line(format!("{}", username)); + bldr.line("UserName"); + bldr.line(format!("{username}")); } if let Some(ref groupname) = ctx.runas.group { - bldr.line(r#"GroupName"#); - bldr.line(format!("{}", groupname)); + bldr.line("GroupName"); + bldr.line(format!("{groupname}")); } if ctx.runas.initgroups { - bldr.line(r#"InitGroups"#); + bldr.line("InitGroups"); bldr.line(""); } if let Some(ref umask) = ctx.runas.umask { - bldr.line(r#"Umask"#); - bldr.line(format!("{}", umask)); + bldr.line("Umask"); + bldr.line(format!("{umask}")); } if ctx.have_args() { - bldr.line(r#"ProgramArguments"#); + bldr.line("ProgramArguments"); bldr.scope("", Some("{}", service_binary_path)); + bldr.line(format!("{service_binary_path}")); for arg in &ctx.args { - bldr.line(format!("{}", arg)); + bldr.line(format!("{arg}")); } bldr.exit(); // } else { - bldr.line(r#"Program"#); - bldr.line(format!("{}", service_binary_path)); + bldr.line("Program"); + bldr.line(format!("{service_binary_path}")); } let mut envs = Vec::new(); if let Some(ll) = ctx.log_level { envs.push((String::from("LOG_LEVEL"), ll.to_string())); } - if let Some(ll) = ctx.trace_level { - envs.push((String::from("TRACE_LEVEL"), ll.to_string())); + if let Some(ref tf) = ctx.trace_filter { + envs.push((String::from("TRACE_FILTER"), tf.to_string())); } if let Some(ref fname) = ctx.trace_file { envs.push((String::from("TRACE_FILE"), fname.to_string())); } if ctx.have_envs() { @@ -69,25 +74,25 @@ envs.push((key.to_string(), value.to_string())); } } if !envs.is_empty() { - bldr.line(r#"EnvironmentVariables"#); + bldr.line("EnvironmentVariables"); bldr.scope("", Some("{}", key)); - bldr.line(format!("{}", value)); + bldr.line(format!("{key}")); + bldr.line(format!("{value}")); } bldr.exit(); // } if let Some(wd) = ctx.workdir { bldr.line("WorkingDirectory"); - bldr.line(format!("{}", wd)); + bldr.line(format!("{wd}")); } if ctx.autostart { bldr.line("RunAtLoad"); bldr.line(""); @@ -109,21 +114,24 @@ // ToDo: Set proper path let fname = format!("{}.plist", ctx.svcname); let fname = Path::new(&fname); - // ToDo: If plist file already exist then fail -- unless force flag was - // specified. - if fname.exists() { - Err(Error::io("File already exists."))?; + if fname.exists() && !ctx.force { + Err(Error::IO( + std::io::Error::new(ErrorKind::AlreadyExists, "File already exists") + .to_string() + ))?; } fs::write(fname, buf)?; Ok(()) } +#[expect(clippy::missing_errors_doc)] +#[expect(clippy::missing_const_for_fn)] pub fn uninstall(_svcname: &str) -> Result<(), Error> { Ok(()) } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/installer/systemd.rs ================================================================== --- src/installer/systemd.rs +++ src/installer/systemd.rs @@ -1,90 +1,147 @@ use std::{fs, path::Path}; use crate::err::Error; +/// Generate a systemd service unit file. +/// +/// # Errors +/// [`Error::IO`] will be returned if file I/O failed when trying to generate +/// unit file. pub fn install(ctx: super::RegSvc) -> Result<(), Error> { let service_binary_path = ::std::env::current_exe()? .to_str() - .ok_or(Error::bad_format("Executable pathname is not utf-8"))? + .ok_or_else(|| Error::bad_format("Executable pathname is not utf-8"))? .to_string(); // // [Unit] // let mut unit_lines: Vec = vec![]; unit_lines.push("[Unit]".into()); if let Some(ref desc) = ctx.description { - unit_lines.push(format!("Description={}", desc)); + unit_lines.push(format!("Description={desc}")); + } + if ctx.netservice { + // Note: The After=network.target does _not_ enure that the network is up + // and running before starting the service. It's more suited to + // ensuring that the service gets _shut down_ before the network is. + // + // Trying to use the service subsystem to wait for the network is a bad + // idea -- the service application should do it instead. + // + // See: https://www.freedesktop.org/wiki/Software/systemd/NetworkTarget/ + unit_lines.push("After=network.target".into()); } // // [Service] // let mut svc_lines: Vec = vec![]; svc_lines.push("[Service]".into()); - svc_lines.push("Type=notify".into()); + if ctx.conf_reload { + svc_lines.push("Type=notify-reload".into()); + } else { + svc_lines.push("Type=notify".into()); + } if let Some(ref username) = ctx.runas.user { - svc_lines.push(format!(r#"User="{}""#, username)); + svc_lines.push(format!(r#"User="{username}""#)); } if let Some(ref groupname) = ctx.runas.group { - svc_lines.push(format!(r#"Group="{}""#, groupname)); + svc_lines.push(format!(r#"Group="{groupname}""#)); } if let Some(ref umask) = ctx.runas.umask { - svc_lines.push(format!(r#"UMask="{}""#, umask)); + svc_lines.push(format!(r#"UMask="{umask}""#)); + } + + /* + if ctx.restart_on_failure { + svc_lines.push(format!("Restart=on-failure")); } + */ if let Some(ll) = ctx.log_level { - svc_lines.push(format!(r#"Environment="LOG_LEVEL={}""#, ll.to_string())); + svc_lines.push(format!(r#"Environment="LOG_LEVEL={ll}""#)); } - if let Some(ll) = ctx.trace_level { - svc_lines.push(format!(r#"Environment="TRACE_LEVEL={}""#, ll.to_string())); + if let Some(tf) = ctx.trace_filter { + svc_lines.push(format!(r#"Environment="TRACE_FILTER={tf}""#)); } if let Some(fname) = ctx.trace_file { - svc_lines.push(format!(r#"Environment="TRACE_FILE={}""#, fname)); + svc_lines.push(format!(r#"Environment="TRACE_FILE={fname}""#)); } for (key, value) in &ctx.envs { - svc_lines.push(format!(r#"Environment="{}={}""#, key, value)); + svc_lines.push(format!(r#"Environment="{key}={value}""#)); } if let Some(wd) = ctx.workdir { - svc_lines.push(format!("WorkingDirectory={}", wd)); + svc_lines.push(format!("WorkingDirectory={wd}")); + } + + // + // Set up service executable and command line arguments. + // + // + let mut cmdline = vec![service_binary_path]; + for arg in &ctx.args { + cmdline.push(arg.clone()); } - svc_lines.push(format!("ExecStart={}", service_binary_path)); + svc_lines.push(format!("ExecStart={}", cmdline.join(" "))); // // [Install] // - let mut inst_lines: Vec = vec![]; - inst_lines.push("[Install]".into()); - inst_lines.push("WantedBy=multi-user.target".into()); + let inst_lines = [ + String::from("[Install]"), + String::from("WantedBy=multi-user.target") + ]; // // Putting it all together // - let mut blocks: Vec = vec![]; - blocks.push(unit_lines.join("\n")); - blocks.push(svc_lines.join("\n")); - blocks.push(inst_lines.join("\n")); + let blocks = [ + unit_lines.join("\n"), + svc_lines.join("\n"), + inst_lines.join("\n") + ]; - let filebuf = blocks.join("\n\n"); + let mut filebuf = blocks.join("\n\n"); + filebuf.push('\n'); // ToDo: Set proper path let fname = format!("{}.service", ctx.svcname); let fname = Path::new(&fname); - // ToDo: If plist file already exist then fail -- unless force flag was - // specified. - if fname.exists() { + // ToDo: If file already exists, should it be assumed that the service needs + // to be stopped? + if fname.exists() && !ctx.force { Err(Error::io("File already exists."))?; } fs::write(fname, filebuf)?; + + /* + if install_location == System { + "systemctl daemon-reload" + + if autorun { + "systemctl enable {}.service" + } + } + */ Ok(()) } +#[expect(clippy::missing_errors_doc)] +#[expect(clippy::missing_const_for_fn)] pub fn uninstall(_svcname: &str) -> Result<(), Error> { + /* + if install_location == System { + systemctl stop {}.service + rm {}.service + systemctl daemon-reload + } + */ Ok(()) } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/installer/winsvc.rs ================================================================== --- src/installer/winsvc.rs +++ src/installer/winsvc.rs @@ -6,19 +6,22 @@ ServiceStartType, ServiceState, ServiceType }, service_manager::{ServiceManager, ServiceManagerAccess} }; -use crate::{ - err::Error, - rt::winsvc::{ - create_service_params, get_service_params_subkey, write_service_subkey - } +use crate::rt::winsvc::{ + create_service_params, get_service_params_subkey, write_service_subkey }; + +use super::Error; /// Register a service in the system service's subsystem. +/// +/// # Errors +/// Problems encountered in the Windows service subsystem will be reported as +/// [`Error::SubSystem`]. // ToDo: Make notes about Windows-specific semantics: // - Uses registry // - Installer // - Windows Event Log pub fn install(ctx: super::RegSvc) -> Result<(), Error> { @@ -27,21 +30,19 @@ // Create a refrence cell used to keep track of whether to keep system // motifications (or not) when leaving function. let status = RefCell::new(false); // Register an event source named by the service name. - eventlog::register(&svcname)?; + eventlog::register(svcname)?; // The event source registration was successful and is a persistent change. // If this function returns early due to an error we want to roll back the // changes it made up to that point. This scope guard is used to deregister // the event source of the function returns early. let _status = scopeguard::guard(&status, |st| { - if !*st.borrow() { - if eventlog::deregister(svcname).is_err() { - eprintln!("!!> Unable to deregister event source"); - } + if !*st.borrow() && eventlog::deregister(svcname).is_err() { + eprintln!("!!> Unable to deregister event source"); } }); let manager_access = ServiceManagerAccess::CONNECT | ServiceManagerAccess::CREATE_SERVICE; @@ -49,15 +50,14 @@ ServiceManager::local_computer(None::<&str>, manager_access)?; let service_binary_path = ::std::env::current_exe()?; // Default display name to service name - let display_name = if let Some(ref display_name) = ctx.display_name { - display_name - } else { - svcname - }; + let display_name = ctx + .display_name + .as_ref() + .map_or(svcname, |display_name| display_name); // // Generate service launch arguments // let launch_args: Vec = @@ -80,14 +80,16 @@ deps.push(OsString::from(d)); } } } } - let dependencies: Vec = deps - .into_iter() - .map(|x| ServiceDependency::Service(x)) - .collect(); + + let dependencies: Vec = + deps.into_iter().map(ServiceDependency::Service).collect(); + + // If None, then run as System + let account_name = ctx.runas.user.clone().map(OsString::from); let service_info = ServiceInfo { name: OsString::from(svcname), display_name: OsString::from(display_name), service_type: ServiceType::OWN_PROCESS, @@ -94,18 +96,18 @@ start_type: autostart, error_control: ServiceErrorControl::Normal, executable_path: service_binary_path, launch_arguments: launch_args, dependencies, - account_name: None, // run as System + account_name, account_password: None }; //println!("==> Registering service '{}' ..", svcname); let service = service_manager .create_service(&service_info, ServiceAccess::CHANGE_CONFIG)?; let _status = scopeguard::guard(&status, |st| { - if *st.borrow() == false { + if !(*st.borrow()) { let service_access = ServiceAccess::DELETE; let res = service_manager.open_service(svcname, service_access); if let Ok(service) = res { if service.delete().is_err() { eprintln!("!!> Unable to delete service"); @@ -120,15 +122,12 @@ service.set_description(desc)?; } if ctx.have_envs() { let key = write_service_subkey(svcname)?; - let envs: Vec = ctx - .envs - .iter() - .map(|(k, v)| format!("{}={}", k, v)) - .collect(); + let envs: Vec = + ctx.envs.iter().map(|(k, v)| format!("{k}={v}")).collect(); key.set_value("Environment", &envs)?; } //println!("==> Service installation successful"); @@ -142,15 +141,15 @@ } if let Some(ll) = ctx.log_level { params.set_value("LogLevel", &ll.to_string())?; } - if let Some(ll) = ctx.trace_level { - params.set_value("TraceLevel", &ll.to_string())?; + if let Some(ll) = ctx.trace_filter { + params.set_value("TraceFilter", &ll)?; } if let Some(fname) = ctx.trace_file { - params.set_value("TraceFile", &fname.to_string())?; + params.set_value("TraceFile", &fname)?; } // Give application the opportunity to create registry keys. if let Some(cb) = ctx.regconf { @@ -173,10 +172,14 @@ /// can be uninstalled. /// /// Before attempting an actual uninstallation, this function will verify that /// under the service's `Parameters` subkey there is an `Installer` key with /// the value `qsu`. +/// +/// # Errors +/// Problems encountered in the Windows service subsystem will be reported as +/// [`Error::SubSystem`]. pub fn uninstall(svcname: &str) -> Result<(), Error> { // Only allow uninstallation of services that have an Installer=qsu key in // its Parameters subkey. if let Ok(params) = get_service_params_subkey(svcname) { if let Ok(val) = params.get_value::("Installer") { @@ -199,19 +202,19 @@ let manager_access = ServiceManagerAccess::CONNECT; let service_manager = ServiceManager::local_computer(None::<&str>, manager_access)?; let service_access = ServiceAccess::QUERY_STATUS | ServiceAccess::STOP | ServiceAccess::DELETE; - let service = service_manager.open_service(&svcname, service_access)?; + let service = service_manager.open_service(svcname, service_access)?; // Make sure service is stopped before trying to delete it loop { let service_status = service.query_status()?; if service_status.current_state == ServiceState::Stopped { break; } - println!("==> Requesting service '{}' to stop ..", svcname); + println!("==> Requesting service '{svcname}' to stop .."); service.stop()?; thread::sleep(Duration::from_secs(2)); } Index: src/lib.rs ================================================================== --- src/lib.rs +++ src/lib.rs @@ -54,12 +54,15 @@ pub use async_trait::async_trait; pub use lumberjack::LumberJack; -pub use apperr; -pub use err::{AppErr, Error}; +pub use err::Error; + +#[cfg(any(feature = "rt", feature = "installer"))] +#[cfg_attr(docsrs, doc(cfg(any(feature = "rt", feature = "installer"))))] +pub use err::CbErr; #[cfg(feature = "tokio")] pub use tokio; #[cfg(feature = "rocket")] @@ -66,15 +69,21 @@ pub use rocket; pub use log; pub use tracing; + +#[cfg(feature = "clap")] +#[cfg_attr(docsrs, doc(cfg(feature = "clap")))] +pub use clap; + /// Attempt to derive a default service name based on the executable's name. /// /// The idea is to get the current executable's file name and strip it's /// extension (if there is one). The file stem name is the default service /// name. On macos the name will be prefixed by `local.`. +#[must_use] pub fn default_service_name() -> Option { let binary_path = ::std::env::current_exe().ok()?; let name = binary_path.file_name()?; let name = Path::new(name); @@ -88,9 +97,9 @@ nm.to_str().map(String::from) } #[cfg(target_os = "macos")] fn mkname(nm: &OsStr) -> Option { - nm.to_str().map(|x| format!("local.{}", x)) + nm.to_str().map(|x| format!("local.{x}")) } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/lumberjack.rs ================================================================== --- src/lumberjack.rs +++ src/lumberjack.rs @@ -45,35 +45,24 @@ /// `TRACE_FILTER` in a similar manner, but defaults to `none`. /// /// If the environment variable `TRACE_FILE` is set the value will be the /// used as the file name to write the trace logs to. fn default() -> Self { - let log_level = if let Ok(level) = std::env::var("LOG_LEVEL") { - if let Ok(level) = level.parse::() { + let log_level = + std::env::var("LOG_LEVEL").map_or(LogLevel::Warn, |level| { level - } else { - LogLevel::Warn - } - } else { - LogLevel::Warn - }; - - let trace_file = if let Ok(v) = std::env::var("TRACE_FILE") { - Some(PathBuf::from(v)) - } else { - None - }; - - let trace_filter = if let Ok(var) = env::var("TRACE_FILTER") { - Some(var.to_string()) - } else { - None - }; + .parse::() + .map_or(LogLevel::Warn, |level| level) + }); + + let trace_file = + std::env::var("TRACE_FILE").map_or(None, |v| Some(PathBuf::from(v))); + let trace_filter = env::var("TRACE_FILTER").ok(); Self { init: true, - log_out: Default::default(), + log_out: LogOut::default(), log_level, trace_filter, //log_file: None, trace_file } @@ -80,36 +69,42 @@ } } impl LumberJack { /// Create a [`LumberJack::default()`] object. + #[must_use] pub fn new() -> Self { Self::default() } /// Do not initialize logging/tracing. /// /// This is useful when running tests. + #[must_use] pub fn noinit() -> Self { Self { init: false, ..Default::default() } } - pub fn set_init(mut self, flag: bool) -> Self { + #[must_use] + pub const fn set_init(mut self, flag: bool) -> Self { self.init = flag; self } /// Load logging/tracing information from a service Parameters subkey. + /// + /// # Errors + /// `Error::SubSystem` #[cfg(windows)] pub fn from_winsvc(svcname: &str) -> Result { let params = crate::rt::winsvc::get_service_param(svcname)?; let loglevel = params .get_value::("LogLevel") - .unwrap_or(String::from("warn")) + .unwrap_or_else(|_| String::from("warn")) .parse::() .unwrap_or(LogLevel::Warn); let tracefilter = params.get_value::("TraceFilter"); let tracefile = params.get_value::("TraceFile"); @@ -126,32 +121,39 @@ Ok(this) } /// Set the `log` logging level. - pub fn log_level(mut self, level: LogLevel) -> Self { + #[must_use] + pub const fn log_level(mut self, level: LogLevel) -> Self { self.log_level = level; self } /// Set the `tracing` log level. + #[must_use] + #[allow(clippy::needless_pass_by_value)] pub fn trace_filter(mut self, filter: impl ToString) -> Self { self.trace_filter = Some(filter.to_string()); self } /// Set a file to which `tracing` log entries are written (rather than to /// write to console). + #[must_use] pub fn trace_file

(mut self, fname: P) -> Self where P: AsRef { self.trace_file = Some(fname.as_ref().to_path_buf()); self } /// Commit requested settings to `log` and `tracing`. + /// + /// # Errors + /// Any initialization error is translated into [`Error::LumberJack`]. pub fn init(self) -> Result<(), Error> { if self.init { match self.log_out { LogOut::Console => { init_console_logging()?; @@ -166,20 +168,17 @@ if let Some(fname) = self.trace_file { init_file_tracing(fname, self.trace_filter.as_deref()); } else { init_console_tracing(self.trace_filter.as_deref()); } - - Ok(()) - } else { - Ok(()) } + Ok(()) } } -#[derive(Debug, PartialEq, Eq, Clone, Copy)] +#[derive(Default, Debug, PartialEq, Eq, Clone, Copy)] #[cfg_attr(feature = "clap", derive(ValueEnum))] pub enum LogLevel { /// No logging. #[cfg_attr(feature = "clap", clap(name = "off"))] Off, @@ -188,10 +187,11 @@ #[cfg_attr(feature = "clap", clap(name = "error"))] Error, /// Log warnings and errors. #[cfg_attr(feature = "clap", clap(name = "warn"))] + #[default] Warn, /// Log info, warnings and errors. #[cfg_attr(feature = "clap", clap(name = "info"))] Info, @@ -204,60 +204,54 @@ #[cfg_attr(feature = "clap", clap(name = "trace"))] Trace } impl FromStr for LogLevel { - type Err = Error; + type Err = String; fn from_str(s: &str) -> Result { match s { - "off" => Ok(LogLevel::Off), - "error" => Ok(LogLevel::Error), - "warn" => Ok(LogLevel::Warn), - "info" => Ok(LogLevel::Info), - "debug" => Ok(LogLevel::Debug), - "trace" => Ok(LogLevel::Trace), - _ => Err(Error::BadFormat(format!("Unknown log level '{}'", s))) - } - } -} - -impl Default for LogLevel { - fn default() -> Self { - LogLevel::Warn + "off" => Ok(Self::Off), + "error" => Ok(Self::Error), + "warn" => Ok(Self::Warn), + "info" => Ok(Self::Info), + "debug" => Ok(Self::Debug), + "trace" => Ok(Self::Trace), + _ => Err(format!("Unknown log level '{s}'")) + } } } impl fmt::Display for LogLevel { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { let s = match self { - LogLevel::Off => "off", - LogLevel::Error => "error", - LogLevel::Warn => "warn", - LogLevel::Info => "info", - LogLevel::Debug => "debug", - LogLevel::Trace => "trace" + Self::Off => "off", + Self::Error => "error", + Self::Warn => "warn", + Self::Info => "info", + Self::Debug => "debug", + Self::Trace => "trace" }; - write!(f, "{}", s) + write!(f, "{s}") } } impl From for log::LevelFilter { - fn from(ll: LogLevel) -> log::LevelFilter { + fn from(ll: LogLevel) -> Self { match ll { - LogLevel::Off => log::LevelFilter::Off, - LogLevel::Error => log::LevelFilter::Error, - LogLevel::Warn => log::LevelFilter::Warn, - LogLevel::Info => log::LevelFilter::Info, - LogLevel::Debug => log::LevelFilter::Debug, - LogLevel::Trace => log::LevelFilter::Trace + LogLevel::Off => Self::Off, + LogLevel::Error => Self::Error, + LogLevel::Warn => Self::Warn, + LogLevel::Info => Self::Info, + LogLevel::Debug => Self::Debug, + LogLevel::Trace => Self::Trace } } } impl From for Option { - fn from(ll: LogLevel) -> Option { + fn from(ll: LogLevel) -> Self { match ll { LogLevel::Off => None, LogLevel::Error => Some(tracing::Level::ERROR), LogLevel::Warn => Some(tracing::Level::WARN), LogLevel::Info => Some(tracing::Level::INFO), @@ -301,23 +295,12 @@ Ok(()) } pub fn init_console_tracing(filter: Option<&str>) { - /* - let subscriber = FmtSubscriber::builder().with_max_level(lvl).finish(); - tracing::subscriber::set_global_default(subscriber) - .expect("setting default subscriber failed"); - */ - - // When running on console, then disable tracing by default - let filter = if let Some(spec) = filter { - EnvFilter::new(spec) - } else { - EnvFilter::new("none") - }; + let filter = filter.map_or_else(|| EnvFilter::new("none"), EnvFilter::new); tracing_subscriber::fmt() .with_env_filter(filter) //.with_timer(timer) .init(); @@ -329,11 +312,11 @@ /// This function will attempt to set up tracing to a file. There are three /// conditions that must be true in order for tracing to a file to be enabled: /// - A filename must be specified (either via the `fname` argument or the /// `TRACE_FILE` environment variable). /// - A trace filter must be specified (either via the `filter` argument or the -/// `TRACE_FILTER` environment variable). +/// `TRACE_FILTER` environment variable). /// - The file name must be openable for writing. /// /// Because tracing is an optional feature and intended for development only, /// this function will enable tracing if possible, and silently ignore and /// errors. @@ -346,33 +329,20 @@ // let timer = UtcTime::new(format_description!( "[year]-[month]-[day] [hour]:[minute]:[second]" )); - let f = if let Ok(f) = - fs::OpenOptions::new().create(true).append(true).open(fname) - { - f - } else { + let Ok(f) = fs::OpenOptions::new().create(true).append(true).open(fname) + else { return; }; // // If TRACING_FILE is set, then default to filter out at warn level. // Disable all tracing if TRACE_FILTER is not set // - let filter = if let Some(spec) = filter { - EnvFilter::new(spec) - } else { - EnvFilter::new("warn") - /* - EnvFilter::builder() - .with_default_directive(LevelFilter::OFF.into()) - .parse("") - .unwrap() - */ - }; + let filter = filter.map_or_else(|| EnvFilter::new("warn"), EnvFilter::new); tracing_subscriber::fmt() .with_env_filter(filter) .with_writer(f) .with_ansi(false) Index: src/rt.rs ================================================================== --- src/rt.rs +++ src/rt.rs @@ -16,31 +16,30 @@ //! +---------------------+ //! //! //! The primary goal of the _qsu_ runtime is to provide a consistent interface //! that will be mapped to whatever service subsystem it is actually running -//! on. This including no special runtime at all; such as when running the +//! on. This including no special subsystem at all; such as when running the //! server application as a regular foreground process (which is common during //! development). //! //! # How server applications are implemented and used //! _qsu_ needs to know what kind of runtime the server application expects. //! Server applications pick the runtime type by implementing a trait, of which //! there are currently three recognized types: //! - [`ServiceHandler`] is used for "non-async" server applications. //! - [`TokioServiceHandler`] is used for server applications that run under -//! the tokio executor. +//! the tokio executor. //! - [`RocketServiceHandler`] is for server applications that are built on top //! of the Rocket HTTP framework. //! //! Each of these implement three methods: //! - `init()` is for initializing the service. //! - `run()` is for running the actual server application. //! - `shutdown()` is for shutting down the server application. //! -//! The actual trait methods may look quite different, depending on the trait -//! being used. +//! The specifics of these trait methods may look quite different. //! //! Once a service trait has been implemented, the application creates a //! [`RunCtx`] object and calls its [`run()`](RunCtx::run()) method, passing in //! an service implementation object. //! @@ -53,56 +52,13 @@ //! //! The handler's `shutdown()` will be called regardless of whether `init()` or //! `run()` was successful (the only precondition for `shutdown()` to be called //! is that `init()` was called). //! -//! # Application errors -//! The _qsu_ runtime is initialized and run from an application that is called -//! back to from the _qsu_ runtime. This has the unfortunate side effect of -//! creating a kind of barrier between the application's "outside" (the part -//! that sets up and runs the _qsu_ runtime) and the "inside" (the service -//! trait callback methods). Specifically, the problem this causes is that if -//! an error occurs in the "inner" server application code, the "outer" -//! application code may want to know exactly what the inner error was. -//! -//! _qsu_ bridges this gap by providing the [`AppErr`] type for the `Err()` -//! case of the callbacks. The `AppErr` is a newtype over a boxed `Any` type. -//! In order to get at the original error value from the "inside" the `AppErr` -//! needs to be unwrapped. See the [`AppErr`] documentation for more -//! information. -//! -//! Presumably the application has it's own `Error` type. To allow the -//! callbacks to return application-defined errors using the regular question -//! mark, it may be helpful to add the following error conversion to the error -//! module: -//! -//! ``` -//! use std::fmt; -//! -//! // Application-specific Error type -//! #[derive(Debug)] -//! enum Error { -//! // .. application-defined error variants -//! } -//! impl fmt::Display for Error { -//! fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { -//! Ok(()) -//! } -//! } -//! impl std::error::Error for Error {} -//! impl apperr::Blessed for Error {} -//! -//! impl From for qsu::AppErr { -//! fn from(err: Error) -> qsu::AppErr { -//! qsu::AppErr::new(err) -//! } -//! } -//! ``` -//! -//! # Using the argument parser +//! # Argument parser //! _qsu_ offers an [argument parser](crate::argp::ArgParser), which can -//! abstract away much of the runtime management. +//! abstract away much of the runtime management and service registration. mod nosvc; mod rttype; mod signals; @@ -124,12 +80,15 @@ #[cfg(feature = "tokio")] use tokio::runtime; use tokio::sync::broadcast; +#[cfg(all(target_os = "linux", feature = "systemd"))] +use sd_notify::NotifyState; -use crate::{err::Error, lumberjack::LumberJack, AppErr}; + +use crate::{err::CbErr, lumberjack::LumberJack}; /// The run time environment. /// /// Can be used by the application callbacks to determine whether it is running /// as a service. @@ -165,12 +124,12 @@ } impl AsRef for StateMsg { fn as_ref(&self) -> &str { match self { - StateMsg::Ref(s) => s, - StateMsg::Owned(s) => s + Self::Ref(s) => s, + Self::Owned(s) => s } } } @@ -195,10 +154,11 @@ } impl InitCtx { /// Return context used to identify whether service application is running as /// a foreground process or a system service. + #[must_use] pub fn runenv(&self) -> RunEnv { self.re.clone() } /// Report startup state to the system service manager. @@ -220,10 +180,11 @@ } impl TermCtx { /// Return context used to identify whether service application is running as /// a foreground process or a system service. + #[must_use] pub fn runenv(&self) -> RunEnv { self.re.clone() } /// Report shutdown state to the system service manager. @@ -240,15 +201,32 @@ /// "Synchronous" (non-`async`) server application. /// /// Implement this for an object that wraps a server application that does not /// use an async runtime. pub trait ServiceHandler { - fn init(&mut self, ictx: InitCtx) -> Result<(), AppErr>; + type AppErr; + + /// Implement to handle service application initialization. + /// + /// # Errors + /// Application-defined error returned from callback will be wrapped in + /// [`CbErr::App`] and returned to application. + fn init(&mut self, ictx: InitCtx) -> Result<(), Self::AppErr>; - fn run(&mut self, re: &RunEnv, ser: SvcEvtReader) -> Result<(), AppErr>; + /// Implement to run service application. + /// + /// # Errors + /// Application-defined error returned from callback will be wrapped in + /// [`CbErr::App`] and returned to application. + fn run(&mut self, re: &RunEnv) -> Result<(), Self::AppErr>; - fn shutdown(&mut self, tctx: TermCtx) -> Result<(), AppErr>; + /// Implement to handle service application termination. + /// + /// # Errors + /// Application-defined error returned from callback will be wrapped in + /// [`CbErr::App`] and returned to application. + fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr>; } /// `async` server application built on the tokio runtime. /// @@ -256,19 +234,17 @@ /// tokio as an async runtime. #[cfg(feature = "tokio")] #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] #[async_trait] pub trait TokioServiceHandler { - async fn init(&mut self, ictx: InitCtx) -> Result<(), AppErr>; - - async fn run( - &mut self, - re: &RunEnv, - ser: SvcEvtReader - ) -> Result<(), AppErr>; - - async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), AppErr>; + type AppErr; + + async fn init(&mut self, ictx: InitCtx) -> Result<(), Self::AppErr>; + + async fn run(&mut self, re: &RunEnv) -> Result<(), Self::AppErr>; + + async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr>; } /// Rocket server application handler. /// @@ -294,39 +270,78 @@ /// runtime to be responsible for initiating the `Rocket` shutdown. #[cfg(feature = "rocket")] #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))] #[async_trait] pub trait RocketServiceHandler { + type AppErr; + /// Rocket service initialization. /// /// The returned `Rocket`s will be ignited and their shutdown handlers will /// be triggered on shutdown. async fn init( &mut self, ictx: InitCtx - ) -> Result>, AppErr>; + ) -> Result>, Self::AppErr>; /// Server application main entry point. /// /// If the `init()` trait method returned `Rocket` instances to the /// qsu runtime it will ignote these and pass them to `run()`. It is the /// responsibility of this method to launch the rockets and await them. async fn run( &mut self, rockets: Vec>, - re: &RunEnv, - ser: SvcEvtReader - ) -> Result<(), AppErr>; + re: &RunEnv + ) -> Result<(), Self::AppErr>; + + async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr>; +} + + +/// The means through which the service termination happened. +#[derive(Copy, Clone, Debug)] +pub enum Demise { + /// On unixy platforms, this indicates that the termination was initiated + /// via a `SIGINT` signal. When running as a foreground process on Windows + /// this indicates that Ctrl+C was issued. + Interrupted, + + /// On unixy platforms, this indicates that the termination was initiated + /// via the `SIGTERM` signal. + /// + /// On Windows, running as a service, this indicates that the service + /// subsystem requested service to be shut down. Running as a foreground + /// process, this means that Ctrl+Break was issued or that the console + /// window was closed. + Terminated, + + /// Reached the end of the service application without any external requests + /// to terminate. + ReachedEnd +} + +#[derive(Copy, Clone, Debug)] +pub enum UserSig { + /// SIGUSR1 + Sig1, - async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), AppErr>; + /// SIGUSR2 + Sig2 } /// Event notifications that originate from the service subsystem that is /// controlling the server application. -#[derive(Clone, Debug)] +#[derive(Copy, Clone, Debug)] pub enum SvcEvt { + /// User events. + /// + /// These will be generated on unixy platform if the process receives + /// SIGUSR1 or SIGUSR2. + User(UserSig), + /// Service subsystem has requested that the server application should pause /// its operations. /// /// Only the Windows service subsystem will emit these events. Pause, @@ -342,73 +357,46 @@ /// /// On Unixy platforms this is triggered by SIGHUP, and is unsupported on /// Windows. ReloadConf, - /// The service subsystem (or equivalent) has requested that the service - /// shutdown. - Shutdown, - - /// The service subsystem (or equivalent) has requested that the service - /// terminate. - Terminate -} - - -/// Channel end-point used to receive events from the service subsystem. -pub struct SvcEvtReader { - rx: broadcast::Receiver -} - -impl SvcEvtReader { - /// Block and wait for an event. - /// - /// Once `SvcEvt::Shutdown` or `SvcEvt::Terminate` has been received, this - /// method should not be called again. - pub fn recv(&mut self) -> Option { - self.rx.blocking_recv().ok() - } - - /// Attempt to get next event. - /// - /// Once `SvcEvt::Shutdown` or `SvcEvt::Terminate` has been received, this - /// method should not be called again. - pub fn try_recv(&mut self) -> Option { - self.rx.try_recv().ok() - } - - /// Async wait for an event. - /// - /// Once `SvcEvt::Shutdown` or `SvcEvt::Terminate` has been received, this - /// method should not be called again. - pub async fn arecv(&mut self) -> Option { - self.rx.recv().await.ok() - } + /// The service application is terminating. The `Demise` value indicates + /// the reason for the shutdown. + Shutdown(Demise) } /// The server application runtime type. - -pub enum SrvAppRt { +// large_enum_variant isn't relevant here because only one instance of this is +// ever created for a process. +#[allow(clippy::large_enum_variant, clippy::module_name_repetitions)] +pub enum SrvAppRt { /// A plain non-async (sometimes referred to as "blocking") server /// application. - Sync(Box), + Sync { + svcevt_handler: Box, + rt_handler: Box + Send> + }, /// Initializa a tokio runtime, and call each application handler from an /// async context. #[cfg(feature = "tokio")] #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] - Tokio( - Option, - Box - ), + Tokio { + rtbldr: Option, + svcevt_handler: Box, + rt_handler: Box + Send> + }, /// Allow Rocket to initialize the tokio runtime, and call each application /// handler from an async context. #[cfg(feature = "rocket")] #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))] - Rocket(Box) + Rocket { + svcevt_handler: Box, + rt_handler: Box + Send> + } } /// Service runner context. pub struct RunCtx { @@ -419,73 +407,131 @@ } impl RunCtx { /// Run as a systemd service. #[cfg(all(target_os = "linux", feature = "systemd"))] - fn systemd(self, st: SrvAppRt) -> Result<(), Error> { + fn systemd(self, st: SrvAppRt) -> Result<(), CbErr> + where + ApEr: Send + { LumberJack::default().set_init(self.log_init).init()?; tracing::debug!("Running service '{}'", self.svcname); - let reporter = Arc::new(systemd::ServiceReporter {}); + let sr = Arc::new(systemd::ServiceReporter {}); let re = RunEnv::Service(Some(self.svcname.clone())); - let res = match st { - SrvAppRt::Sync(handler) => { - rttype::sync_main(re, handler, reporter, None, self.test_mode) - } - SrvAppRt::Tokio(rtbldr, handler) => { - rttype::tokio_main(rtbldr, re, handler, reporter, None) - } + match st { + SrvAppRt::Sync { + svcevt_handler, + rt_handler + } => rttype::sync_main(rttype::SyncMainParams { + re, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: None, + test_mode: self.test_mode + }), + SrvAppRt::Tokio { + rtbldr, + svcevt_handler, + rt_handler + } => rttype::tokio_main( + rtbldr, + rttype::TokioMainParams { + re, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: None + } + ), #[cfg(feature = "rocket")] - SrvAppRt::Rocket(handler) => { - rttype::rocket_main(re, handler, reporter, None) - } - }; - - res + SrvAppRt::Rocket { + svcevt_handler, + rt_handler + } => rttype::rocket_main(rttype::RocketMainParams { + re, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: None + }) + } } /// Run as a Windows service. #[cfg(windows)] - fn winsvc(self, st: SrvAppRt) -> Result<(), Error> { + fn winsvc(self, st: SrvAppRt) -> Result<(), CbErr> + where + ApEr: Send + 'static + { winsvc::run(&self.svcname, st)?; Ok(()) } /// Run as a foreground server - fn foreground(self, st: SrvAppRt) -> Result<(), Error> { + fn foreground(self, st: SrvAppRt) -> Result<(), CbErr> + where + ApEr: Send + { LumberJack::default().set_init(self.log_init).init()?; tracing::debug!("Running service '{}'", self.svcname); - let reporter = Arc::new(nosvc::ServiceReporter {}); - - let re = RunEnv::Foreground; + let sr = Arc::new(nosvc::ServiceReporter {}); match st { - SrvAppRt::Sync(handler) => { - rttype::sync_main(re, handler, reporter, None, self.test_mode) - } + SrvAppRt::Sync { + svcevt_handler, + rt_handler + } => rttype::sync_main(rttype::SyncMainParams { + re: RunEnv::Foreground, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: None, + test_mode: self.test_mode + }), #[cfg(feature = "tokio")] - SrvAppRt::Tokio(rtbldr, handler) => { - rttype::tokio_main(rtbldr, re, handler, reporter, None) - } + SrvAppRt::Tokio { + rtbldr, + svcevt_handler, + rt_handler + } => rttype::tokio_main( + rtbldr, + rttype::TokioMainParams { + re: RunEnv::Foreground, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: None + } + ), #[cfg(feature = "rocket")] - SrvAppRt::Rocket(handler) => { - rttype::rocket_main(re, handler, reporter, None) - } + SrvAppRt::Rocket { + svcevt_handler, + rt_handler + } => rttype::rocket_main(rttype::RocketMainParams { + re: RunEnv::Foreground, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: None + }) } } } impl RunCtx { /// Create a new service running context. + #[must_use] pub fn new(name: &str) -> Self { Self { service: false, svcname: name.into(), log_init: true, @@ -499,11 +545,12 @@ /// /// qsu performs a few global initialization that will fail if run repeatedly /// within the same process. This causes some problem when running tests, /// because rust may run tests in threads within the same process. #[doc(hidden)] - pub fn test_mode(mut self) -> Self { + #[must_use] + pub const fn test_mode(mut self) -> Self { self.log_init = false; self.test_mode = true; self } @@ -510,11 +557,12 @@ /// Disable logging/tracing initialization. /// /// This is useful in tests because tests may run in different threads within /// the same process, causing the log/tracing initialization to panic. #[doc(hidden)] - pub fn log_init(mut self, flag: bool) -> Self { + #[must_use] + pub const fn log_init(mut self, flag: bool) -> Self { self.log_init = flag; self } /// Reference version of [`RunCtx::log_init()`]. @@ -524,11 +572,12 @@ self } /// Mark this run context to run under the operating system's subservice, if /// one is available on this platform. - pub fn service(mut self) -> Self { + #[must_use] + pub const fn service(mut self) -> Self { self.service = true; self } /// Mark this run context to run under the operating system's subservice, if @@ -536,11 +585,12 @@ pub fn service_ref(&mut self) -> &mut Self { self.service = true; self } - pub fn is_service(&self) -> bool { + #[must_use] + pub const fn is_service(&self) -> bool { self.service } /// Launch the application. /// @@ -548,11 +598,19 @@ /// appropriate service subsystem integration before running the actual /// server application code. /// /// This function must only be called from the main thread of the process, /// and must be called before any other threads are started. - pub fn run(self, st: SrvAppRt) -> Result<(), Error> { + /// + /// # Errors + /// [`CbErr::App`] is returned, containing application-specific error, if n + /// application callback returned an error. [`CbErr::Lib`] indicates that an + /// error occurred in the qsu runtime. + pub fn run(self, st: SrvAppRt) -> Result<(), CbErr> + where + ApEr: Send + 'static + { if self.service { #[cfg(all(target_os = "linux", feature = "systemd"))] self.systemd(st)?; #[cfg(windows)] @@ -570,35 +628,120 @@ Ok(()) } /// Convenience method around [`Self::run()`] using [`SrvAppRt::Sync`]. - pub fn run_sync( + #[allow(clippy::missing_errors_doc)] + pub fn run_sync( self, - handler: Box - ) -> Result<(), Error> { - self.run(SrvAppRt::Sync(handler)) + svcevt_handler: Box, + rt_handler: Box + Send> + ) -> Result<(), CbErr> + where + ApEr: Send + 'static + { + self.run(SrvAppRt::Sync { + svcevt_handler, + rt_handler + }) } /// Convenience method around [`Self::run()`] using [`SrvAppRt::Tokio`]. #[cfg(feature = "tokio")] #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] - pub fn run_tokio( + #[allow(clippy::missing_errors_doc)] + pub fn run_tokio( self, rtbldr: Option, - handler: Box - ) -> Result<(), Error> { - self.run(SrvAppRt::Tokio(rtbldr, handler)) + svcevt_handler: Box, + rt_handler: Box + Send> + ) -> Result<(), CbErr> + where + ApEr: Send + 'static + { + self.run(SrvAppRt::Tokio { + rtbldr, + svcevt_handler, + rt_handler + }) } /// Convenience method around [`Self::run()`] using [`SrvAppRt::Rocket`]. #[cfg(feature = "rocket")] #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))] - pub fn run_rocket( + #[allow(clippy::missing_errors_doc)] + pub fn run_rocket( self, - handler: Box - ) -> Result<(), Error> { - self.run(SrvAppRt::Rocket(handler)) + svcevt_handler: Box, + rt_handler: Box + Send> + ) -> Result<(), CbErr> + where + ApEr: Send + 'static + { + self.run(SrvAppRt::Rocket { + svcevt_handler, + rt_handler + }) + } +} + +/// Internal thread used to run service event handler. +#[tracing::instrument(name = "svcevtthread", skip_all)] +fn svcevt_thread( + mut rx: broadcast::Receiver, + mut evt_handler: Box +) { + while let Ok(msg) = rx.blocking_recv() { + tracing::debug!("Received {:?}", msg); + + #[cfg(all(target_os = "linux", feature = "systemd"))] + if matches!(msg, SvcEvt::ReloadConf) { + // + // Reload has been requested -- report RELOADING=1 and MONOTONIC_USEC to + // systemd before calling the application callback. + // + let ts = + nix::time::clock_gettime(nix::time::ClockId::CLOCK_MONOTONIC).unwrap(); + let s = format!( + "RELOADING=1\nMONOTONIC_USEC={}{:06}", + ts.tv_sec(), + ts.tv_nsec() / 1000 + ); + tracing::trace!("Sending notification to systemd: {}", s); + + let custom = NotifyState::Custom(&s); + if let Err(e) = sd_notify::notify(false, &[custom]) { + log::error!( + "Unable to send RELOADING=1 notification to systemd; {}", + e + ); + } + } + + // + // Call the application callback + // + evt_handler(msg); + + #[cfg(all(target_os = "linux", feature = "systemd"))] + if matches!(msg, SvcEvt::ReloadConf) { + // + // This is a reload; report READY=1 to systemd after the application + // callback has been called. + // + tracing::trace!("Sending notification to systemd: READY=1"); + if let Err(e) = sd_notify::notify(false, &[NotifyState::Ready]) { + log::error!("Unable to send READY=1 notification to systemd; {}", e); + } + } + + // If the event message was either shutdown or terminate, break out of loop + // so the thread will terminate + if let SvcEvt::Shutdown(_) = msg { + tracing::debug!("Terminating thread"); + // break out of loop when the service shutdown has been rquested + break; + } } } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/rt/rttype.rs ================================================================== --- src/rt/rttype.rs +++ src/rt/rttype.rs @@ -2,25 +2,25 @@ //! //! This module is used to collect the various supported "runtime types" or //! "run contexts". mod sync; + +#[cfg(feature = "tokio")] +#[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] +mod tokio; #[cfg(feature = "rocket")] #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))] mod rocket; +pub use sync::{main as sync_main, MainParams as SyncMainParams}; + #[cfg(feature = "tokio")] #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] -mod tokio; - -pub(crate) use sync::sync_main; +pub use tokio::{main as tokio_main, MainParams as TokioMainParams}; #[cfg(feature = "rocket")] #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))] -pub(crate) use rocket::rocket_main; - -#[cfg(feature = "tokio")] -#[cfg_attr(docsrs, doc(cfg(feature = "tokio")))] -pub(crate) use tokio::tokio_main; +pub use rocket::{main as rocket_main, MainParams as RocketMainParams}; // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/rt/rttype/rocket.rs ================================================================== --- src/rt/rttype/rocket.rs +++ src/rt/rttype/rocket.rs @@ -12,168 +12,173 @@ //! //! Server applications do not need to use this feature and should return an //! empty vector from `init()` in this case. This also requires the //! application code to trigger a shutdown of each instance itself. -use std::sync::{atomic::AtomicU32, Arc}; +use std::{ + sync::{atomic::AtomicU32, Arc}, + thread +}; use tokio::{sync::broadcast, task}; use killswitch::KillSwitch; use crate::{ - err::{AppErrors, Error}, + err::{AppErrors, CbErr}, rt::{ - signals, InitCtx, RocketServiceHandler, RunEnv, StateReporter, SvcEvt, - SvcEvtReader, TermCtx + signals, Demise, InitCtx, RocketServiceHandler, RunEnv, StateReporter, + SvcEvt, TermCtx } }; +#[cfg(unix)] +use crate::rt::UserSig; + + +pub struct MainParams +where + ApEr: Send +{ + pub(crate) re: RunEnv, + pub(crate) svcevt_handler: Box, + pub(crate) rt_handler: Box + Send>, + pub(crate) sr: Arc, + pub(crate) svcevt_ch: + Option<(broadcast::Sender, broadcast::Receiver)> +} -/// Internal "main()" routine for server applications that run one or more +/// Internal `main()`-like routine for server applications that run one or more /// Rockets as their main application. -pub(crate) fn rocket_main( - re: RunEnv, - handler: Box, - sr: Arc, - rx_svcevt: Option> -) -> Result<(), Error> { - rocket::execute(rocket_async_main(re, handler, sr, rx_svcevt))?; +pub fn main(params: MainParams) -> Result<(), CbErr> +where + ApEr: Send +{ + rocket::execute(rocket_async_main(params))?; Ok(()) } -async fn rocket_async_main( - re: RunEnv, - mut handler: Box, - sr: Arc, - rx_svcevt: Option> -) -> Result<(), Error> { +async fn rocket_async_main( + MainParams { + re, + svcevt_handler, + mut rt_handler, + sr, + svcevt_ch + }: MainParams +) -> Result<(), CbErr> +where + ApEr: Send +{ let ks = KillSwitch::new(); - // If a SvcEvt receiver end-point was handed to us, then use it. Otherwise - // create our own and spawn the monitoring tasks that will generate events - // for it. - let rx_svcevt = if let Some(rx_svcevt) = rx_svcevt { - rx_svcevt + // If a SvcEvt receiver end-point was handed to us, then use it. The + // presumption is that there's a service subsystem somewhere that has + // created the channel and is holding one of the end-points. + // + // Otherwise create our own channel and spawn the monitoring tasks that will + // generate events for it. + let (tx_svcevt, rx_svcevt) = if let Some((tx_svcevt, rx_svcevt)) = svcevt_ch + { + (tx_svcevt, rx_svcevt) } else { - // Create channel used to signal events to application - let (tx, rx) = broadcast::channel(16); - - let ks2 = ks.clone(); - - // SIGINT (on unix) and Ctrl+C on Windows should trigger a Shutdown event. - let txc = tx.clone(); - task::spawn(signals::wait_shutdown( - move || { - if let Err(e) = txc.send(SvcEvt::Shutdown) { - log::error!("Unable to send SvcEvt::Shutdown event; {}", e); - } - }, - ks2 - )); - - // SIGTERM (on unix) and Ctrl+Break/Close on Windows should trigger a - // Terminate event. - let txc = tx.clone(); - let ks2 = ks.clone(); - task::spawn(signals::wait_term( - move || { - if let Err(e) = txc.send(SvcEvt::Terminate) { - log::error!("Unable to send SvcEvt::Terminate event; {}", e); - } - }, - ks2 - )); - - // There doesn't seem to be anything equivalent to SIGHUP for Windows - // (Services) - #[cfg(unix)] - { - let ks2 = ks.clone(); - - let txc = tx.clone(); - task::spawn(signals::wait_reload( - move || { - if let Err(e) = txc.send(SvcEvt::ReloadConf) { - log::error!("Unable to send SvcEvt::ReloadConf event; {}", e); - } - }, - ks2 - )); - } - - rx + init_svc_channels(&ks) }; - let mut rx_svcevt2 = rx_svcevt.resubscribe(); - - let set = Box::new(SvcEvtReader { rx: rx_svcevt }); // Call application's init() method. let ictx = InitCtx { re: re.clone(), sr: Arc::clone(&sr), cnt: Arc::new(AtomicU32::new(2)) // 1 is used by the runtime, so start at 2 }; - let (rockets, init_apperr) = match handler.init(ictx).await { + let (rockets, init_apperr) = match rt_handler.init(ictx).await { Ok(rockets) => (rockets, None), Err(e) => (Vec::new(), Some(e)) }; // Ignite rockets so we can get Shutdown contexts for each of the instances. + // There are two cases where the rockets vector will be empty: + // - if init() returned error. + // - the application doesn't want to use the built-in rocket shutdown + // support. let mut ignited = vec![]; let mut rocket_shutdowns = vec![]; for rocket in rockets { let rocket = rocket.ignite().await?; rocket_shutdowns.push(rocket.shutdown()); ignited.push(rocket); } - // Set the service's state to "started" - sr.started(); - - // Launch a task that waits for the SvtEvt::Shutdown event. Once it - // arrives, tell all rocket instances to gracefully shutdown. - // - // Note: We don't want to use the killswitch for this because the killswitch - // isn't triggered until run() has returned, and we might want the graceful - // shutdown to be the cause of the graceful shutdowns. - let jh_graceful_landing = task::spawn(async move { - loop { - match rx_svcevt2.recv().await { - Ok(SvcEvt::Shutdown) => { - tracing::trace!("Ask rocket instances to shut down gracefully"); - for shutdown in rocket_shutdowns { - // Tell this rocket instance to shut down gracefully. - shutdown.notify(); - } - break; - } - Ok(SvcEvt::Terminate) => { + // If init() was successful, then do some prepartations and then run the + // application run() callback. + let run_apperr = if init_apperr.is_none() { + // Set the service's state to "started" + sr.started(); + + let mut rx_svcevt2 = rx_svcevt.resubscribe(); + + // Launch a task that waits for the SvtEvt::Shutdown event. Once it + // arrives, tell all rocket instances to gracefully shutdown. + // + // Note: We don't want to use the killswitch for this because the + // killswitch isn't triggered until run() has returned. + let jh_graceful_landing = task::spawn(async move { + loop { + let msg = match rx_svcevt2.recv().await { + Ok(msg) => msg, + Err(e) => { + log::error!("Unable to receive broadcast SvcEvt message, {}", e); + break; + } + }; + if let SvcEvt::Shutdown(_) = msg { tracing::trace!("Ask rocket instances to shut down gracefully"); for shutdown in rocket_shutdowns { // Tell this rocket instance to shut down gracefully. shutdown.notify(); } break; } - Ok(_) => { - tracing::trace!("Ignored message in wask waiting for shutdown"); - continue; - } - Err(e) => { - log::error!("Unable to receive broadcast SvcEvt message, {}", e); - break; - } - } - } - }); - - let run_apperr = if init_apperr.is_none() { - sr.started(); - handler.run(ignited, &re, *set).await.err() + tracing::trace!("Ignored message in task waiting for shutdown"); + } + }); + + sr.started(); + + // Kick off service event monitoring thread before running main app + // callback + let jh = thread::Builder::new() + .name("svcevt".into()) + .spawn(|| crate::rt::svcevt_thread(rx_svcevt, svcevt_handler)) + .unwrap(); + + // Run the main service application callback. + // + // This is basically the service application's "main()". + let res = rt_handler.run(ignited, &re).await.err(); + + // Shut down svcevent thread + // + // Tell it that an (implicit) shutdown event has occurred. + // Duplicates don't matter, because once the first one is processed the + // thread will terminate. + let _ = tx_svcevt.send(SvcEvt::Shutdown(Demise::ReachedEnd)); + + // Wait for all task that is waiting for a shutdown event to complete + if let Err(e) = jh_graceful_landing.await { + log::warn!( + "An error was returned from the graceful landing task; {}", + e + ); + } + + // Wait for service event monitoring thread to terminate + let _ = task::spawn_blocking(|| jh.join()).await; + + res } else { None }; @@ -182,19 +187,10 @@ sr.stopping(1, None); // Now that the main application has terminated kill off any remaining // auxiliary tasks (read: signal waiters) ks.trigger(); - - // .. and wait for all task that is waiting for a shutdown event to complete - if let Err(e) = jh_graceful_landing.await { - log::warn!( - "An error was returned from the graceful landing task; {}", - e - ); - } - if (ks.finalize().await).is_err() { log::warn!("Attempted to finalize KillSwitch that wasn't triggered yet"); } // Call the application's shutdown() function. @@ -201,11 +197,11 @@ let tctx = TermCtx { re, sr: Arc::clone(&sr), cnt: Arc::new(AtomicU32::new(2)) // 1 is used by the runtime, so start at 2 }; - let term_apperr = handler.shutdown(tctx).await.err(); + let term_apperr = rt_handler.shutdown(tctx).await.err(); // Inform the service subsystem that the the shutdown is complete sr.stopped(); // There can be multiple failures, and we don't want to lose information @@ -215,12 +211,100 @@ let apperrs = AppErrors { init: init_apperr, run: run_apperr, shutdown: term_apperr }; - Err(Error::SrvApp(apperrs))?; + Err(CbErr::SrvApp(apperrs))?; } Ok(()) } + +fn init_svc_channels( + ks: &KillSwitch +) -> (broadcast::Sender, broadcast::Receiver) { + // Create channel used to signal events to application + let (tx, rx) = broadcast::channel(16); + + // ToDo: autoclone + let ks2 = ks.clone(); + + // SIGINT (on unix) and Ctrl+C on Windows should trigger a Shutdown event. + // ToDo: autoclone + let txc = tx.clone(); + task::spawn(signals::wait_shutdown( + move || { + if let Err(e) = txc.send(SvcEvt::Shutdown(Demise::Interrupted)) { + log::error!("Unable to send SvcEvt::Shutdown event; {}", e); + } + }, + ks2 + )); + + // SIGTERM (on unix) and Ctrl+Break/Close on Windows should trigger a + // Terminate event. + // ToDo: autoclone + let txc = tx.clone(); + // ToDo: autoclone + let ks2 = ks.clone(); + task::spawn(signals::wait_term( + move || { + if let Err(e) = txc.send(SvcEvt::Shutdown(Demise::Terminated)) { + log::error!("Unable to send SvcEvt::Terminate event; {}", e); + } + }, + ks2 + )); + + // There doesn't seem to be anything equivalent to SIGHUP for Windows + // (Services) + #[cfg(unix)] + { + // ToDo: autoclone + let ks2 = ks.clone(); + + // ToDo: autoclone + let txc = tx.clone(); + task::spawn(signals::wait_reload( + move || { + if let Err(e) = txc.send(SvcEvt::ReloadConf) { + log::error!("Unable to send SvcEvt::ReloadConf event; {}", e); + } + }, + ks2 + )); + } + + #[cfg(unix)] + { + let ks2 = ks.clone(); + + let txc = tx.clone(); + task::spawn(signals::wait_user1( + move || { + if let Err(e) = txc.send(SvcEvt::User(UserSig::Sig1)) { + log::error!("Unable to send SvcEvt::User(Sig1) event; {}", e); + } + }, + ks2 + )); + } + + #[cfg(unix)] + { + let ks2 = ks.clone(); + + let txc = tx.clone(); + task::spawn(signals::wait_user2( + move || { + if let Err(e) = txc.send(SvcEvt::User(UserSig::Sig2)) { + log::error!("Unable to send SvcEvt::User(Sig2) event; {}", e); + } + }, + ks2 + )); + } + + (tx, rx) +} // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/rt/rttype/sync.rs ================================================================== --- src/rt/rttype/sync.rs +++ src/rt/rttype/sync.rs @@ -1,84 +1,141 @@ -use std::sync::{atomic::AtomicU32, Arc}; +use std::{ + sync::{atomic::AtomicU32, Arc}, + thread +}; use tokio::sync::broadcast; use crate::{ - err::{AppErrors, Error}, + err::{AppErrors, CbErr}, rt::{ - signals, InitCtx, RunEnv, ServiceHandler, StateReporter, SvcEvt, - SvcEvtReader, TermCtx + signals, Demise, InitCtx, RunEnv, ServiceHandler, StateReporter, SvcEvt, + TermCtx } }; #[cfg(unix)] -use crate::rt::signals::SigType; +use crate::rt::{signals::SigType, UserSig}; + +pub struct MainParams { + pub(crate) re: RunEnv, + pub(crate) svcevt_handler: Box, + pub(crate) rt_handler: Box>, + pub(crate) sr: Arc, + pub(crate) svcevt_ch: + Option<(broadcast::Sender, broadcast::Receiver)>, + pub(crate) test_mode: bool +} -/// Internal "main()" routine for server applications that run plain old +/// Internal `main()`-like routine for server applications that run plain old /// non-`async` code. -pub(crate) fn sync_main( - re: RunEnv, - mut handler: Box, - sr: Arc, - rx_svcevt: Option>, - test_mode: bool -) -> Result<(), Error> { +pub fn main( + MainParams { + re, + svcevt_handler, + mut rt_handler, + sr, + svcevt_ch, + test_mode + }: MainParams +) -> Result<(), CbErr> { // Get rid of unused variable warning #[cfg(unix)] let _ = test_mode; - let rx_svcevt = if let Some(rx_svcevt) = rx_svcevt { + // If a channel was passed from caller, then use it. The assumption is + // that there's some other runtime that is monitoring for system + // (service-related) events that is holding a sending end-point. + // + // Otherwise create a new unbounded channel and kick off the appropriate + // system event monitoring. + let (tx_svcevt, rx_svcevt) = if let Some((tx_svcevt, rx_svcevt)) = svcevt_ch + { // Use the broadcast receiver supplied by caller (it likely originates from // a service runtime integration). - rx_svcevt + (tx_svcevt, rx_svcevt) } else { let (tx, rx) = broadcast::channel(16); + let tx2 = tx.clone(); + #[cfg(unix)] signals::sync_sigmon(move |st| match st { + SigType::Usr1 => { + if let Err(e) = tx2.send(SvcEvt::User(UserSig::Sig1)) { + log::error!("Unable to send SvcEvt::Info event; {}", e); + } + } + SigType::Usr2 => { + if let Err(e) = tx2.send(SvcEvt::User(UserSig::Sig2)) { + log::error!("Unable to send SvcEvt::Info event; {}", e); + } + } SigType::Int => { - if let Err(e) = tx.send(SvcEvt::Shutdown) { + if let Err(e) = tx2.send(SvcEvt::Shutdown(Demise::Interrupted)) { log::error!("Unable to send SvcEvt::Shutdown event; {}", e); } } SigType::Term => { - if let Err(e) = tx.send(SvcEvt::Terminate) { + if let Err(e) = tx2.send(SvcEvt::Shutdown(Demise::Terminated)) { log::error!("Unable to send SvcEvt::Terminate event; {}", e); } } SigType::Hup => { - if let Err(e) = tx.send(SvcEvt::ReloadConf) { + if let Err(e) = tx2.send(SvcEvt::ReloadConf) { log::error!("Unable to send SvcEvt::ReloadConf event; {}", e); } } })?; // On Windows, if rx_svcevt is None, means we're not running under the // service subsystem (i.e. we're running as a foreground process), so // register a Ctrl+C handler. #[cfg(windows)] - signals::sync_kill_to_event(tx, test_mode)?; + signals::sync_kill_to_event(tx2, test_mode)?; - rx + (tx, rx) }; - let set = Box::new(SvcEvtReader { rx: rx_svcevt }); // Call server application's init() method, passing along a startup state // reporter object. let ictx = InitCtx { re: re.clone(), sr: Arc::clone(&sr), cnt: Arc::new(AtomicU32::new(2)) // 1 is used by the runtime, so start at 2 }; - let init_apperr = handler.init(ictx).err(); + let init_apperr = rt_handler.init(ictx).err(); // If init() was successful, set the service's state to "started" and then // call the server application's run() method. let run_apperr = if init_apperr.is_none() { sr.started(); - handler.run(&re, *set).err() + + // Kick off service event monitoring thread before running main app + // callback + let jh = thread::Builder::new() + .name("svcevt".into()) + .spawn(|| crate::rt::svcevt_thread(rx_svcevt, svcevt_handler)) + .unwrap(); + + // Run the main service application callback. + // + // This is basically the service application's "main()". + let ret = rt_handler.run(&re).err(); + + // Shut down svcevent thread + // + // Tell it that an (implicit) shutdown event has occurred. + // Duplicates don't matter, because once the first one is processed the + // thread will terminate. + let _ = tx_svcevt.send(SvcEvt::Shutdown(Demise::ReachedEnd)); + + // Wait for service event monitoring thread to terminate + let _ = jh.join(); + + ret } else { None }; // Always send the first shutdown checkpoint here. Either init() failed or @@ -90,14 +147,15 @@ let tctx = TermCtx { re, sr: Arc::clone(&sr), cnt: Arc::new(AtomicU32::new(2)) // 1 is used by the runtime, so start at 2 }; - let term_apperr = handler.shutdown(tctx).err(); + let term_apperr = rt_handler.shutdown(tctx).err(); // Inform the service subsystem that the the shutdown is complete sr.stopped(); + // There can be multiple failures, and we don't want to lose information // about what went wrong, so return an error context that can contain all // callback errors. if init_apperr.is_some() || run_apperr.is_some() || term_apperr.is_some() { @@ -104,12 +162,12 @@ let apperrs = AppErrors { init: init_apperr, run: run_apperr, shutdown: term_apperr }; - Err(Error::SrvApp(apperrs))?; + Err(CbErr::SrvApp(apperrs))?; } Ok(()) } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/rt/rttype/tokio.rs ================================================================== --- src/rt/rttype/tokio.rs +++ src/rt/rttype/tokio.rs @@ -1,128 +1,125 @@ -use std::sync::{atomic::AtomicU32, Arc}; +use std::{ + sync::{atomic::AtomicU32, Arc}, + thread +}; use tokio::{runtime, sync::broadcast, task}; use crate::{ - err::{AppErrors, Error}, + err::{AppErrors, CbErr}, rt::{ - signals, InitCtx, RunEnv, StateReporter, SvcEvt, SvcEvtReader, TermCtx, + signals, Demise, InitCtx, RunEnv, StateReporter, SvcEvt, TermCtx, TokioServiceHandler } }; use killswitch::KillSwitch; +#[cfg(unix)] +use crate::rt::UserSig; + + +pub struct MainParams { + pub(crate) re: RunEnv, + pub(crate) svcevt_handler: Box, + pub(crate) rt_handler: Box + Send>, + pub(crate) sr: Arc, + pub(crate) svcevt_ch: + Option<(broadcast::Sender, broadcast::Receiver)> +} + -/// Internal "main()" routine for server applications that run the tokio +/// Internal `main()`-like routine for server applications that run the tokio /// runtime for `async` code. -pub(crate) fn tokio_main( +pub fn main( rtbldr: Option, - re: RunEnv, - handler: Box, - sr: Arc, - rx_svcevt: Option> -) -> Result<(), Error> { + params: MainParams +) -> Result<(), CbErr> +where + ApEr: Send +{ let rt = if let Some(mut bldr) = rtbldr { bldr.build()? } else { tokio::runtime::Runtime::new()? }; - rt.block_on(tokio_async_main(re, handler, sr, rx_svcevt))?; + rt.block_on(async_main(params))?; Ok(()) } /// The `async` main function for tokio servers. /// /// If `rx_svcevt` is `Some(_)` it means the channel was created elsewhere /// (implied: The transmitting endpoint lives somewhere else). If it is `None` /// the channel needs to be created. -async fn tokio_async_main( - re: RunEnv, - mut handler: Box, - sr: Arc, - rx_svcevt: Option> -) -> Result<(), Error> { +async fn async_main( + MainParams { + re, + svcevt_handler, + mut rt_handler, + sr, + svcevt_ch + }: MainParams +) -> Result<(), CbErr> +where + ApEr: Send +{ let ks = KillSwitch::new(); // If a SvcEvt receiver end-point was handed to us, then use it. Otherwise // create our own and spawn the monitoring tasks that will generate events // for it. - let rx_svcevt = if let Some(rx_svcevt) = rx_svcevt { - rx_svcevt + let (tx_svcevt, rx_svcevt) = if let Some((tx_svcevt, rx_svcevt)) = svcevt_ch + { + (tx_svcevt, rx_svcevt) } else { - // Create channel used to signal events to application - let (tx, rx) = broadcast::channel(16); - - let ks2 = ks.clone(); - - // SIGINT (on unix) and Ctrl+C on Windows should trigger a Shutdown event. - let txc = tx.clone(); - task::spawn(signals::wait_shutdown( - move || { - if let Err(e) = txc.send(SvcEvt::Shutdown) { - log::error!("Unable to send SvcEvt::ReloadConf event; {}", e); - } - }, - ks2 - )); - - // SIGTERM (on unix) and Ctrl+Break/Close on Windows should trigger a - // Terminate event. - let txc = tx.clone(); - let ks2 = ks.clone(); - task::spawn(signals::wait_term( - move || { - if let Err(e) = txc.send(SvcEvt::Terminate) { - log::error!("Unable to send SvcEvt::Terminate event; {}", e); - } - }, - ks2 - )); - - // There doesn't seem to be anything equivalent to SIGHUP for Windows - // (Services) - #[cfg(unix)] - { - let ks2 = ks.clone(); - - let txc = tx.clone(); - task::spawn(signals::wait_reload( - move || { - if let Err(e) = txc.send(SvcEvt::ReloadConf) { - log::error!("Unable to send SvcEvt::ReloadConf event; {}", e); - } - }, - ks2 - )); - } - - rx + init_svc_channels(&ks) }; - let set = Box::new(SvcEvtReader { rx: rx_svcevt }); - // Call application's init() method. let ictx = InitCtx { re: re.clone(), sr: Arc::clone(&sr), cnt: Arc::new(AtomicU32::new(2)) // 1 is used by the runtime, so start at 2 }; - let init_apperr = handler.init(ictx).await.err(); + let init_apperr = rt_handler.init(ictx).await.err(); let run_apperr = if init_apperr.is_none() { sr.started(); - handler.run(&re, *set).await.err() + + // Kick off service event monitoring thread before running main app + // callback + let jh = thread::Builder::new() + .name("svcevt".into()) + .spawn(|| crate::rt::svcevt_thread(rx_svcevt, svcevt_handler)) + .unwrap(); + + // Run the main service application callback. + // + // This is basically the service application's "main()". + let ret = rt_handler.run(&re).await.err(); + + // Shut down svcevent thread + // + // Tell it that an (implicit) shutdown event has occurred. + // Duplicates don't matter, because once the first one is processed the + // thread will terminate. + let _ = tx_svcevt.send(SvcEvt::Shutdown(Demise::ReachedEnd)); + + // Wait for service event monitoring thread to terminate + let _ = task::spawn_blocking(|| jh.join()).await; + + ret } else { None }; // Always send the first shutdown checkpoint here. Either init() failed or // run retuned. Either way, we're shutting down. sr.stopping(1, None); - // Now that the main application has terminated kill off any remaining // auxiliary tasks (read: signal waiters) ks.trigger(); @@ -134,11 +131,11 @@ let tctx = TermCtx { re, sr: Arc::clone(&sr), cnt: Arc::new(AtomicU32::new(2)) // 1 is used by the runtime, so start at 2 }; - let term_apperr = handler.shutdown(tctx).await.err(); + let term_apperr = rt_handler.shutdown(tctx).await.err(); // Inform the service subsystem that the the shutdown is complete sr.stopped(); // There can be multiple failures, and we don't want to lose information @@ -148,12 +145,95 @@ let apperrs = AppErrors { init: init_apperr, run: run_apperr, shutdown: term_apperr }; - Err(Error::SrvApp(apperrs))?; + Err(CbErr::SrvApp(apperrs))?; } Ok(()) } + +fn init_svc_channels( + ks: &KillSwitch +) -> (broadcast::Sender, broadcast::Receiver) { + // Create channel used to signal events to application + let (tx, rx) = broadcast::channel(16); + + let ks2 = ks.clone(); + + // SIGINT (on unix) and Ctrl+C on Windows should trigger a Shutdown event. + let txc = tx.clone(); + task::spawn(signals::wait_shutdown( + move || { + if let Err(e) = txc.send(SvcEvt::Shutdown(Demise::Interrupted)) { + log::error!("Unable to send SvcEvt::ReloadConf event; {}", e); + } + }, + ks2 + )); + + // SIGTERM (on unix) and Ctrl+Break/Close on Windows should trigger a + // Terminate event. + let txc = tx.clone(); + let ks2 = ks.clone(); + task::spawn(signals::wait_term( + move || { + let svcevt = SvcEvt::Shutdown(Demise::Terminated); + if let Err(e) = txc.send(svcevt) { + log::error!("Unable to send SvcEvt::Terminate event; {}", e); + } + }, + ks2 + )); + + // There doesn't seem to be anything equivalent to SIGHUP for Windows + // (Services) + #[cfg(unix)] + { + let ks2 = ks.clone(); + + let txc = tx.clone(); + task::spawn(signals::wait_reload( + move || { + if let Err(e) = txc.send(SvcEvt::ReloadConf) { + log::error!("Unable to send SvcEvt::ReloadConf event; {}", e); + } + }, + ks2 + )); + } + + #[cfg(unix)] + { + let ks2 = ks.clone(); + + let txc = tx.clone(); + task::spawn(signals::wait_user1( + move || { + if let Err(e) = txc.send(SvcEvt::User(UserSig::Sig1)) { + log::error!("Unable to send SvcEvt::User(Sig1) event; {}", e); + } + }, + ks2 + )); + } + + #[cfg(unix)] + { + let ks2 = ks.clone(); + + let txc = tx.clone(); + task::spawn(signals::wait_user2( + move || { + if let Err(e) = txc.send(SvcEvt::User(UserSig::Sig2)) { + log::error!("Unable to send SvcEvt::User(Sig2) event; {}", e); + } + }, + ks2 + )); + } + + (tx, rx) +} // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/rt/signals.rs ================================================================== --- src/rt/signals.rs +++ src/rt/signals.rs @@ -8,14 +8,16 @@ #[cfg(unix)] pub use unix::{sync_sigmon, SigType}; #[cfg(all(unix, feature = "tokio"))] -pub use unix::{wait_reload, wait_shutdown, wait_term}; +pub use unix::{ + wait_reload, wait_shutdown, wait_term, wait_user1, wait_user2 +}; -#[cfg(windows)] +#[cfg(all(windows, feature = "tokio"))] pub use win::{wait_shutdown, wait_term}; #[cfg(windows)] -pub(crate) use win::sync_kill_to_event; +pub use win::sync_kill_to_event; // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: src/rt/signals/unix.rs ================================================================== --- src/rt/signals/unix.rs +++ src/rt/signals/unix.rs @@ -16,91 +16,113 @@ /// Whenever a SIGINT or SIGTERM is signalled the closure in `f` is called and /// the task is terminated. #[cfg(feature = "tokio")] pub async fn wait_shutdown(f: F, ks: KillSwitch) where - F: FnOnce() -{ - tracing::trace!("SIGINT task launched"); - - let Ok(mut sigint) = signal(SignalKind::interrupt()) else { - log::error!("Unable to create SIGINT Future"); - return; - }; - - // Wait for SIGINT. - tokio::select! { - _ = sigint.recv() => { - tracing::debug!("Received SIGINT -- running closure"); - f(); - }, - _ = ks.wait() => { - tracing::debug!("killswitch triggered"); - } - } - - tracing::trace!("wait_shutdown() terminating"); + F: FnOnce() + Send +{ + wait_oneshot_signal(SignalKind::interrupt(), f, ks).await; } #[cfg(feature = "tokio")] pub async fn wait_term(f: F, ks: KillSwitch) where - F: FnOnce() -{ - tracing::trace!("SIGTERM task launched"); - - let Ok(mut sigterm) = signal(SignalKind::terminate()) else { - log::error!("Unable to create SIGTERM Future"); - return; - }; - - // Wait for either SIGTERM. - tokio::select! { - _ = sigterm.recv() => { - tracing::debug!("Received SIGTERM -- running closure"); - f(); - } - _ = ks.wait() => { - tracing::debug!("killswitch triggered"); - } - } - - tracing::trace!("wait_term() terminating"); + F: FnOnce() + Send +{ + wait_oneshot_signal(SignalKind::terminate(), f, ks).await; } /// Async task used to wait for SIGHUP /// /// Whenever a SIGHUP is signalled the closure in `f` is called. #[cfg(feature = "tokio")] pub async fn wait_reload(f: F, ks: KillSwitch) where - F: Fn() + F: Fn() + Send +{ + wait_repeating_signal(SignalKind::hangup(), f, ks).await; +} + +#[cfg(feature = "tokio")] +pub async fn wait_user1(f: F, ks: KillSwitch) +where + F: Fn() + Send +{ + wait_repeating_signal(SignalKind::user_defined1(), f, ks).await; +} + +#[cfg(feature = "tokio")] +pub async fn wait_user2(f: F, ks: KillSwitch) +where + F: Fn() + Send +{ + wait_repeating_signal(SignalKind::user_defined2(), f, ks).await; +} + + +#[cfg(feature = "tokio")] +pub async fn wait_repeating_signal( + sigkind: SignalKind, + f: F, + ks: KillSwitch +) where + F: Fn() + Send { - tracing::trace!("SIGHUP task launched"); + tracing::trace!("Repeating {:?} task launched", sigkind); - let Ok(mut sighup) = signal(SignalKind::hangup()) else { - log::error!("Unable to create SIGHUP Future"); + let Ok(mut sig) = signal(sigkind) else { + log::error!("Unable to create {:?} Future", sigkind); return; }; + loop { tokio::select! { - _ = sighup.recv() => { - tracing::debug!("Received SIGHUP"); + _ = sig.recv() => { + tracing::debug!("Received {:?} -- running closure", sigkind); f(); - }, - _ = ks.wait() => { + } + () = ks.wait() => { tracing::debug!("killswitch triggered"); break; } } } - tracing::trace!("wait_reload() terminating"); + tracing::trace!("{:?} terminating", sigkind); +} + + +#[cfg(feature = "tokio")] +pub async fn wait_oneshot_signal(sigkind: SignalKind, f: F, ks: KillSwitch) +where + F: FnOnce() + Send +{ + tracing::trace!("One-shot {:?} task launched", sigkind); + + let Ok(mut sig) = signal(sigkind) else { + log::error!("Unable to create {:?} Future", sigkind); + return; + }; + + // Wait for either SIGUSR1. + tokio::select! { + _ = sig.recv() => { + tracing::debug!("Received {:?} -- running closure", sigkind); + f(); + } + () = ks.wait() => { + tracing::debug!("killswitch triggered"); + } + } + + tracing::trace!("{:?} terminating", sigkind); } pub enum SigType { + Usr1, + Usr2, Int, Term, Hup } @@ -113,10 +135,12 @@ // let mut ss = SigSet::empty(); ss.add(Signal::SIGINT); ss.add(Signal::SIGTERM); ss.add(Signal::SIGHUP); + ss.add(Signal::SIGUSR1); + ss.add(Signal::SIGUSR2); let mut oldset = SigSet::empty(); nix::sys::signal::pthread_sigmask( SigmaskHow::SIG_SETMASK, Some(&ss), @@ -133,19 +157,27 @@ let mut mask: libc::sigset_t = std::mem::zeroed(); libc::sigemptyset(&mut mask); libc::sigaddset(&mut mask, libc::SIGINT); libc::sigaddset(&mut mask, libc::SIGTERM); libc::sigaddset(&mut mask, libc::SIGHUP); + libc::sigaddset(&mut mask, libc::SIGUSR1); + libc::sigaddset(&mut mask, libc::SIGUSR2); mask }; loop { let mut sig: libc::c_int = 0; let ret = unsafe { libc::sigwait(&mask, &mut sig) }; if ret == 0 { let signal = Signal::try_from(sig).unwrap(); match signal { + Signal::SIGUSR1 => { + f(SigType::Usr1); + } + Signal::SIGUSR2 => { + f(SigType::Usr2); + } Signal::SIGINT => { f(SigType::Int); break; } Signal::SIGTERM => { Index: src/rt/signals/win.rs ================================================================== --- src/rt/signals/win.rs +++ src/rt/signals/win.rs @@ -1,31 +1,36 @@ use std::sync::OnceLock; -use tokio::{signal, sync::broadcast}; +use tokio::sync::broadcast; + +#[cfg(feature = "tokio")] +use {killswitch::KillSwitch, tokio::signal}; use windows_sys::Win32::{ Foundation::{BOOL, FALSE, TRUE}, System::Console::{ SetConsoleCtrlHandler, CTRL_BREAK_EVENT, CTRL_CLOSE_EVENT, CTRL_C_EVENT } }; -use killswitch::KillSwitch; - -use crate::{err::Error, rt::SvcEvt}; +use crate::{ + err::Error, + rt::{Demise, SvcEvt} +}; static CELL: OnceLock BOOL + Send + Sync>> = OnceLock::new(); /// Async task used to wait for Ctrl+C to be signalled. /// /// Whenever a Ctrl+C is signalled the closure in `f` is called and /// the task is terminated. +#[cfg(feature = "tokio")] pub async fn wait_shutdown(f: F, ks: KillSwitch) where - F: FnOnce() + F: FnOnce() + Send { tracing::trace!("CTRL+C task launched"); tokio::select! { _ = signal::ctrl_c() => { @@ -32,21 +37,22 @@ tracing::debug!("Received Ctrl+C"); // Once any process termination signal has been received post call the // callback. f(); }, - _ = ks.wait() => { + () = ks.wait() => { tracing::debug!("killswitch triggered"); } } tracing::trace!("wait_shutdown() terminating"); } +#[cfg(feature = "tokio")] pub async fn wait_term(f: F, ks: KillSwitch) where - F: FnOnce() + F: FnOnce() + Send { tracing::trace!("CTRL+Break/Close task launched"); let Ok(mut cbreak) = signal::windows::ctrl_break() else { log::error!("Unable to create Ctrl+Break monitor"); @@ -69,20 +75,20 @@ tracing::debug!("Received Close"); // Once any process termination signal has been received post call the // callback. f(); }, - _ = ks.wait() => { + () = ks.wait() => { tracing::debug!("killswitch triggered"); } } tracing::trace!("wait_term() terminating"); } -pub(crate) fn sync_kill_to_event( +pub fn sync_kill_to_event( tx: broadcast::Sender, test_mode: bool ) -> Result<(), Error> { setup_sync_fg_kill_handler( move |ty| { @@ -89,11 +95,11 @@ match ty { CTRL_C_EVENT => { tracing::trace!( "Received some kind of event that should trigger a shutdown." ); - if tx.send(SvcEvt::Shutdown).is_ok() { + if tx.send(SvcEvt::Shutdown(Demise::Interrupted)).is_ok() { // We handled this event TRUE } else { FALSE } @@ -100,11 +106,11 @@ } CTRL_BREAK_EVENT | CTRL_CLOSE_EVENT => { tracing::trace!( "Received some kind of event that should trigger a termination." ); - if tx.send(SvcEvt::Terminate).is_ok() { + if tx.send(SvcEvt::Shutdown(Demise::Terminated)).is_ok() { // We handled this event TRUE } else { FALSE } @@ -116,11 +122,11 @@ )?; Ok(()) } -pub(crate) fn setup_sync_fg_kill_handler( +pub fn setup_sync_fg_kill_handler( f: F, test_mode: bool ) -> Result<(), Error> where F: Fn(u32) -> BOOL + Send + Sync + 'static @@ -142,11 +148,11 @@ let rc = unsafe { SetConsoleCtrlHandler(Some(ctrlhandler), 1) }; // Returns non-zero on success (rc != 0) .then_some(()) - .ok_or(Error::internal("SetConsoleCtrlHandler failed"))?; + .ok_or_else(|| Error::internal("SetConsoleCtrlHandler failed"))?; Ok(()) } unsafe extern "system" fn ctrlhandler(ty: u32) -> BOOL { Index: src/rt/systemd.rs ================================================================== --- src/rt/systemd.rs +++ src/rt/systemd.rs @@ -9,15 +9,14 @@ /// A service reporter that sends notifications to systemd. pub struct ServiceReporter {} impl super::StateReporter for ServiceReporter { fn starting(&self, checkpoint: u32, status: Option) { - let text = if let Some(msg) = status { - format!("Starting[{}] {}", checkpoint, msg.as_ref()) - } else { - format!("Startup checkpoint {}", checkpoint) - }; + let text = status.map_or_else( + || format!("Startup checkpoint {checkpoint}"), + |msg| format!("Starting[{checkpoint}] {}", msg.as_ref()) + ); if let Err(e) = sd_notify::notify(false, &[NotifyState::Status(&text)]) { log::error!("Unable to report service started state; {}", e); } } @@ -36,15 +35,14 @@ if let Err(e) = sd_notify::notify(false, &[NotifyState::Stopping]) { log::error!("Unable to report service started state; {}", e); } } - let text = if let Some(msg) = status { - format!("Stopping[{}] {}", checkpoint, msg.as_ref()) - } else { - format!("Stopping checkpoint {}", checkpoint) - }; + let text = status.map_or_else( + || format!("Stopping checkpoint {checkpoint}"), + |msg| format!("Stopping[{checkpoint}] {}", msg.as_ref()) + ); // ToDo: Is it okay to set status after "Stopping" has been set? if let Err(e) = sd_notify::notify(false, &[NotifyState::Status(&text)]) { log::error!("Unable to report service started state; {}", e); } Index: src/rt/winsvc.rs ================================================================== --- src/rt/winsvc.rs +++ src/rt/winsvc.rs @@ -1,6 +1,16 @@ //! Windows service module. +//! +//! While Windows services have a normal `main()` entry point, the way they +//! work is that a (blocking) runtime is called, which calls an application +//! callback where the service application logic is normally implemented. +//! +//! qsu does it a little differently; it runs the service application logic in +//! a thread (called `svcapp`) and runs the Windows service subsystem on the +//! main thread. Its (the service subsystem's) callback is used to monitor the +//! subsystem for events, which are passed on to the qsu event handler +//! application callback. use std::{ ffi::OsString, sync::{Arc, OnceLock}, thread, @@ -25,20 +35,20 @@ self, ServiceControlHandlerResult, ServiceStatusHandle }, service_dispatcher }; -use winreg::{enums::*, RegKey}; +use winreg::{enums::HKEY_LOCAL_MACHINE, RegKey}; #[cfg(feature = "wait-for-debugger")] use dbgtools_win::debugger; use crate::{ - err::Error, + err::{CbErr, Error}, lumberjack::LumberJack, - rt::{RunEnv, SrvAppRt, SvcEvt} + rt::{rttype, Demise, RunEnv, SrvAppRt, SvcEvt} }; use super::StateMsg; @@ -73,13 +83,17 @@ /// Buffer passed back to the application thread from the service subsystem /// thread. struct HandshakeMsg { /// Channel end-point used to send messages to the service subsystem. tx: UnboundedSender, + + /// Channel end-point used to send service event messages to the application + /// callback. + tx_svcevt: broadcast::Sender, /// Channel end-point used to receive messages from the service subsystem. - rx: broadcast::Receiver + rx_svcevt: broadcast::Receiver } /// A service reporter that forwards application state information to the /// windows service subsystem. @@ -120,11 +134,20 @@ log::trace!("Stopped"); } } -pub fn run(svcname: &str, st: SrvAppRt) -> Result<(), Error> { +/// Run a service application under the Windows service subsystem. +/// +/// # Errors +/// `Error::SubSystem` menas the service could not be started. `Error::IO` +/// means the internal worker could not be launched. +#[allow(clippy::missing_panics_doc)] +pub fn run(svcname: &str, st: SrvAppRt) -> Result<(), Error> +where + ApEr: Send + 'static +{ #[cfg(feature = "wait-for-debugger")] { debugger::wait_for_then_break(); debugger::output("Hello, debugger"); } @@ -156,52 +179,90 @@ // Register generated `ffi_service_main` with the system and start the // service, blocking this thread until the service is stopped. service_dispatcher::start(svcname, ffi_service_main)?; + // The return value should be hard-coded to `Result<(), Error>`, so this + // unwrap should be okay. match jh.join() { Ok(_) => Ok(()), Err(e) => *e .downcast::>() .expect("Unable to downcast error from svcapp thread") } } -fn srvapp_thread( - st: SrvAppRt, + +/// Internal server application wrapper thread. +fn srvapp_thread( + st: SrvAppRt, svcname: String, rx_fromsvc: oneshot::Receiver> -) -> Result<(), Error> { +) -> Result<(), CbErr> +where + ApEr: Send +{ // Wait for the service subsystem to report that it has initialized. // It passes along a channel end-point that can be used to send events to // the service manager. - let Ok(res) = rx_fromsvc.blocking_recv() else { panic!("Unable to receive handshake"); }; - let Ok(HandshakeMsg { tx, rx }) = res else { + let Ok(HandshakeMsg { + tx, + tx_svcevt, + rx_svcevt + }) = res + else { panic!("Unable to receive handshake"); }; - let reporter = Arc::new(ServiceReporter { tx: tx.clone() }); + let sr = Arc::new(ServiceReporter { tx }); let re = RunEnv::Service(Some(svcname)); match st { - SrvAppRt::Sync(handler) => { + SrvAppRt::Sync { + svcevt_handler, + rt_handler + } => rttype::sync_main(rttype::SyncMainParams { + re, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: Some((tx_svcevt, rx_svcevt)), // Don't support test mode when running as a windows service - crate::rt::rttype::sync_main(re, handler, reporter, Some(rx), false) - } + test_mode: false + }), #[cfg(feature = "tokio")] - SrvAppRt::Tokio(rtbldr, handler) => { - crate::rt::rttype::tokio_main(rtbldr, re, handler, reporter, Some(rx)) - } + SrvAppRt::Tokio { + rtbldr, + svcevt_handler, + rt_handler + } => rttype::tokio_main( + rtbldr, + rttype::TokioMainParams { + re, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: Some((tx_svcevt, rx_svcevt)) + } + ), + #[cfg(feature = "rocket")] - SrvAppRt::Rocket(handler) => { - crate::rt::rttype::rocket_main(re, handler, reporter, Some(rx)) - } + SrvAppRt::Rocket { + svcevt_handler, + rt_handler + } => rttype::rocket_main(rttype::RocketMainParams { + re, + svcevt_handler, + rt_handler, + sr, + svcevt_ch: Some((tx_svcevt, rx_svcevt)) + }) } } // Generate the windows service boilerplate. The boilerplate contains the @@ -225,10 +286,13 @@ rx_tosvc: UnboundedReceiver, status_handle: ServiceStatusHandle } + +/// Windows Service main entry point. +#[allow(clippy::needless_pass_by_value)] fn my_service_main(_arguments: Vec) { // Start by pulling out the service name and the channel sender. let Xfer { svcname, tx_fromsvc @@ -248,13 +312,11 @@ log::error!("Unable to send handshake message"); return; } // Enter a loop that waits to receive a service termination event. - if let Err(e) = svcloop(rx_tosvc, status_handle) { - log::error!("The service loop failed; {}", e); - } + svcloop(rx_tosvc, status_handle); } Err(e) => { // If svcinit() returns Err() we don't actually know if logging has been // enabled yet -- but we can't do much other than hope that it is and try // to output an error log. @@ -281,11 +343,11 @@ // - If the Paramters subkey can not be loaded, do we abort? // - If the cwd can not be changed to the WorkDir we should abort. if let Ok(svcparams) = get_service_params_subkey(svcname) { if let Ok(wd) = svcparams.get_value::("WorkDir") { std::env::set_current_dir(wd).map_err(|e| { - Error::internal(format!("Unable to switch to WorkDir; {}", e)) + Error::internal(format!("Unable to switch to WorkDir; {e}")) })?; } } // Create channel that will be used to receive messages from the application. @@ -295,10 +357,12 @@ let (tx_svcevt, rx_svcevt) = broadcast::channel(16); // // Define system service event handler that will be receiving service events. // + // ToDo: autoclone + let tx_svcevt2 = tx_svcevt.clone(); let event_handler = move |control_event| -> ServiceControlHandlerResult { match control_event { ServiceControl::Interrogate => { log::debug!("svc signal recieved: interrogate"); // Notifies a service to report its current status information to the @@ -308,11 +372,11 @@ } ServiceControl::Stop => { log::debug!("svc signal recieved: stop"); // Message application that it's time to shutdown - if let Err(e) = tx_svcevt.send(SvcEvt::Shutdown) { + if let Err(e) = tx_svcevt2.send(SvcEvt::Shutdown(Demise::Terminated)) { log::error!("Unable to send SvcEvt::Shutdown from winsvc; {}", e); } ServiceControlHandlerResult::NoError } @@ -352,126 +416,138 @@ } Ok(InitRes { handshake_reply: HandshakeMsg { tx: tx_tosvc, - rx: rx_svcevt + tx_svcevt, + rx_svcevt }, rx_tosvc, status_handle }) } +/// Internal service state loop. +/// +/// Receives the current service application state from an internal channel and +/// uses it to report the state to the windows service subsystem. +/// +/// Once a _Stopped_ state is received, the state will be reported to winsvc +/// subsystem, and the loop will be broken out of so the thread exits. fn svcloop( mut rx_tosvc: UnboundedReceiver, status_handle: ServiceStatusHandle -) -> Result<(), Error> { +) { // // Enter loop that waits for application state changes that should be // reported to the service subsystem. // Once the application reports that it has stopped, then break out of the // loop. // tracing::trace!("enter app state monitoring loop"); loop { - match rx_tosvc.blocking_recv() { - Some(ev) => { - match ev { - ToSvcMsg::Starting(checkpoint) => { - log::debug!("app reported that it is running"); - if let Err(e) = status_handle.set_service_status(ServiceStatus { - service_type: SERVICE_TYPE, - current_state: ServiceState::StartPending, - controls_accepted: ServiceControlAccept::empty(), - exit_code: ServiceExitCode::Win32(0), - checkpoint, - wait_hint: SERVICE_STARTPENDING_TIME, - process_id: None - }) { - log::error!( - "Unable to set service status to 'start pending {}'; {}", - checkpoint, - e - ); - } - } - ToSvcMsg::Started => { - if let Err(e) = status_handle.set_service_status(ServiceStatus { - service_type: SERVICE_TYPE, - current_state: ServiceState::Running, - controls_accepted: ServiceControlAccept::STOP, - exit_code: ServiceExitCode::Win32(0), - checkpoint: 0, - wait_hint: Duration::default(), - process_id: None - }) { - log::error!("Unable to set service status to 'started'; {}", e); - } - } - ToSvcMsg::Stopping(checkpoint) => { - log::debug!("app is shutting down"); - if let Err(e) = status_handle.set_service_status(ServiceStatus { - service_type: SERVICE_TYPE, - current_state: ServiceState::StopPending, - controls_accepted: ServiceControlAccept::empty(), - exit_code: ServiceExitCode::Win32(0), - checkpoint, - wait_hint: SERVICE_STOPPENDING_TIME, - process_id: None - }) { - log::error!( - "Unable to set service status to 'stop pending {}'; {}", - checkpoint, - e - ); - } - } - ToSvcMsg::Stopped => { - if let Err(e) = status_handle.set_service_status(ServiceStatus { - service_type: SERVICE_TYPE, - current_state: ServiceState::Stopped, - controls_accepted: ServiceControlAccept::empty(), - exit_code: ServiceExitCode::Win32(0), - checkpoint: 0, - wait_hint: Duration::default(), - process_id: None - }) { - log::error!("Unable to set service status to 'stopped'; {}", e); - } - - // Break out of loop to terminate service subsystem - break; - } - } - } - None => { - // All the sender halves have been deallocated - log::error!("Sender endpoints unexpectedly disappeared"); + let Some(ev) = rx_tosvc.blocking_recv() else { + // All the sender halves have been deallocated + log::error!("Sender endpoints unexpectedly disappeared"); + break; + }; + match ev { + ToSvcMsg::Starting(checkpoint) => { + log::debug!("app reported that it is running"); + if let Err(e) = status_handle.set_service_status(ServiceStatus { + service_type: SERVICE_TYPE, + current_state: ServiceState::StartPending, + controls_accepted: ServiceControlAccept::empty(), + exit_code: ServiceExitCode::Win32(0), + checkpoint, + wait_hint: SERVICE_STARTPENDING_TIME, + process_id: None + }) { + log::error!( + "Unable to set service status to 'start pending {checkpoint}'; \ + {e}" + ); + } + } + ToSvcMsg::Started => { + if let Err(e) = status_handle.set_service_status(ServiceStatus { + service_type: SERVICE_TYPE, + current_state: ServiceState::Running, + controls_accepted: ServiceControlAccept::STOP, + exit_code: ServiceExitCode::Win32(0), + checkpoint: 0, + wait_hint: Duration::default(), + process_id: None + }) { + log::error!("Unable to set service status to 'started'; {e}"); + } + } + ToSvcMsg::Stopping(checkpoint) => { + log::debug!("app is shutting down"); + if let Err(e) = status_handle.set_service_status(ServiceStatus { + service_type: SERVICE_TYPE, + current_state: ServiceState::StopPending, + controls_accepted: ServiceControlAccept::empty(), + exit_code: ServiceExitCode::Win32(0), + checkpoint, + wait_hint: SERVICE_STOPPENDING_TIME, + process_id: None + }) { + log::error!( + "Unable to set service status to 'stop pending {checkpoint}'; {e}" + ); + } + } + ToSvcMsg::Stopped => { + if let Err(e) = status_handle.set_service_status(ServiceStatus { + service_type: SERVICE_TYPE, + current_state: ServiceState::Stopped, + controls_accepted: ServiceControlAccept::empty(), + exit_code: ServiceExitCode::Win32(0), + checkpoint: 0, + wait_hint: Duration::default(), + process_id: None + }) { + log::error!("Unable to set service status to 'stopped'; {e}"); + } + + // Break out of loop to terminate service subsystem break; } } } tracing::trace!("service terminated"); - - Ok(()) } const SVCPATH: &str = "SYSTEM\\CurrentControlSet\\Services"; const PARAMS: &str = "Parameters"; +/// Create a read-only handle service's registry subkey. +/// +/// `HKLM:SYSTEM\CurrentControlSet\Services\[servicename]` +/// +/// # Errors +/// Registry errors are returned as `Error::SubSystem`. pub fn read_service_subkey( service_name: &str ) -> Result { let hklm = RegKey::predef(HKEY_LOCAL_MACHINE); let services = hklm.open_subkey(SVCPATH)?; let subkey = services.open_subkey(service_name)?; Ok(subkey) } + +/// Create a read/write handle for service's registry subkey. +/// +/// `HKLM:SYSTEM\CurrentControlSet\Services\[servicename]` +/// +/// # Errors +/// Registry errors are returned as `Error::SubSystem`. pub fn write_service_subkey( service_name: &str ) -> Result { let hklm = RegKey::predef(HKEY_LOCAL_MACHINE); let services = hklm.open_subkey(SVCPATH)?; @@ -478,11 +554,16 @@ let subkey = services.open_subkey_with_flags(service_name, winreg::enums::KEY_WRITE)?; Ok(subkey) } -/// Create a Parameters subkey for a service. +/// Create a `Parameters` subkey for a service, and return a handle to it +/// +/// `HKLM:SYSTEM\CurrentControlSet\Services\[servicename]\Parameters` +/// +/// # Errors +/// Registry errors are returned as `Error::SubSystem`. pub fn create_service_params( service_name: &str ) -> Result { let hklm = RegKey::predef(HKEY_LOCAL_MACHINE); let services = hklm.open_subkey(SVCPATH)?; @@ -490,11 +571,16 @@ let (subkey, _disp) = asrv.create_subkey(PARAMS)?; Ok(subkey) } -/// Create a Parameters subkey for a service. +/// Get a read/write handle for the `Parameters` subkey for a service. +/// +/// `HKLM:SYSTEM\CurrentControlSet\Services\[servicename]\Parameters` +/// +/// # Errors +/// Registry errors are returned as `Error::SubSystem`. pub fn get_service_params_subkey( service_name: &str ) -> Result { let hklm = RegKey::predef(HKEY_LOCAL_MACHINE); let services = hklm.open_subkey(SVCPATH)?; @@ -502,11 +588,16 @@ let subkey = asrv.open_subkey(PARAMS)?; Ok(subkey) } -/// Load a service Parameter from the registry. +/// Load a service `Parameter` from the registry. +/// +/// `HKLM:SYSTEM\CurrentControlSet\Services\[servicename]\Parameters` +/// +/// # Errors +/// Registry errors are returned as `Error::SubSystem`. pub fn get_service_param(service_name: &str) -> Result { let hklm = RegKey::predef(HKEY_LOCAL_MACHINE); let services = hklm.open_subkey(SVCPATH)?; let asrv = services.open_subkey(service_name)?; let params = asrv.open_subkey(PARAMS)?; Index: tests/apperr.rs ================================================================== --- tests/apperr.rs +++ tests/apperr.rs @@ -18,29 +18,33 @@ let runctx = RunCtx::new(SVCNAME).test_mode(); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MySyncService { visited: Arc::clone(&visited), ..Default::default() } .fail_init() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_sync(handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_sync(svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only init() failed assert!(errs.init.is_some()); assert!(errs.run.is_none()); assert!(errs.shutdown.is_none()); - let Error::Hello(s) = errs.init.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.init.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Sync::init()"); // Failing init should cause run() not to be called, but shutdown() should @@ -60,29 +64,32 @@ let runctx = RunCtx::new(SVCNAME).test_mode(); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MySyncService { visited: Arc::clone(&visited), ..Default::default() } .fail_run() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_sync(handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_sync(svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only run() failed assert!(errs.init.is_none()); assert!(errs.run.is_some()); assert!(errs.shutdown.is_none()); - let Error::Hello(s) = errs.run.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.run.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Sync::run()"); // Failing run should not hinder shutdown() from being called. @@ -101,29 +108,32 @@ let runctx = RunCtx::new(SVCNAME).test_mode(); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MySyncService { visited: Arc::clone(&visited), ..Default::default() } .fail_shutdown() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_sync(handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_sync(svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only shutdown() failed assert!(errs.init.is_none()); assert!(errs.run.is_none()); assert!(errs.shutdown.is_some()); - let Error::Hello(s) = errs.shutdown.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.shutdown.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Sync::shutdown()"); // All callbacks should have been visited @@ -144,29 +154,32 @@ let runctx = RunCtx::new(SVCNAME).test_mode(); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MyTokioService { visited: Arc::clone(&visited), ..Default::default() } .fail_init() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_tokio(None, handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_tokio(None, svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only init() failed assert!(errs.init.is_some()); assert!(errs.run.is_none()); assert!(errs.shutdown.is_none()); - let Error::Hello(s) = errs.init.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.init.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Tokio::init()"); // Failing init should cause run() not to be called, but shutdown() should @@ -187,29 +200,32 @@ let runctx = RunCtx::new(SVCNAME).test_mode(); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MyTokioService { visited: Arc::clone(&visited), ..Default::default() } .fail_run() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_tokio(None, handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_tokio(None, svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only run() failed assert!(errs.init.is_none()); assert!(errs.run.is_some()); assert!(errs.shutdown.is_none()); - let Error::Hello(s) = errs.run.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.run.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Tokio::run()"); // Failing run should not hinder shutdown() from being called. @@ -230,29 +246,32 @@ let runctx = RunCtx::new(SVCNAME).test_mode(); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MyTokioService { visited: Arc::clone(&visited), ..Default::default() } .fail_shutdown() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_tokio(None, handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_tokio(None, svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only shutdown() failed assert!(errs.init.is_none()); assert!(errs.run.is_none()); assert!(errs.shutdown.is_some()); - let Error::Hello(s) = errs.shutdown.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.shutdown.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Tokio::shutdown()"); // All callbacks should have been visited @@ -273,29 +292,32 @@ let runctx = RunCtx::new(SVCNAME).test_mode(); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MyRocketService { visited: Arc::clone(&visited), ..Default::default() } .fail_init() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_rocket(handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_rocket(svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only init() failed assert!(errs.init.is_some()); assert!(errs.run.is_none()); assert!(errs.shutdown.is_none()); - let Error::Hello(s) = errs.init.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.init.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Rocket::init()"); // Failing init should cause run() not to be called, but shutdown() should @@ -316,29 +338,32 @@ let runctx = RunCtx::new(SVCNAME).log_init(false); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MyRocketService { visited: Arc::clone(&visited), ..Default::default() } .fail_run() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_rocket(handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_rocket(svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only run() failed assert!(errs.init.is_none()); assert!(errs.run.is_some()); assert!(errs.shutdown.is_none()); - let Error::Hello(s) = errs.run.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.run.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Rocket::run()"); // Failing run should not hinder shutdown() from being called. @@ -358,29 +383,32 @@ let runctx = RunCtx::new(SVCNAME).log_init(false); // Prepare a server application context which keeps track of which callbacks // have been called let visited = Arc::new(Mutex::new(apps::Visited::default())); - let handler = Box::new( + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new( apps::MyRocketService { visited: Arc::clone(&visited), ..Default::default() } .fail_shutdown() ); // Call RunCtx::run(), expecting an server application callback error. - let Err(qsu::Error::SrvApp(errs)) = runctx.run_rocket(handler) else { + let Err(qsu::CbErr::SrvApp(errs)) = + runctx.run_rocket(svcevt_handler, rt_handler) + else { panic!("Not expected Err(qsu::Error::SrvApp(_))"); }; // Verify that only shutdown() failed assert!(errs.init.is_none()); assert!(errs.run.is_none()); assert!(errs.shutdown.is_some()); - let Error::Hello(s) = errs.shutdown.unwrap().unwrap_inner::() else { + let Error::Hello(s) = errs.shutdown.unwrap() else { panic!("Not expected Error::Hello"); }; assert_eq!(s, "From Rocket::shutdown()"); // All callbacks should have been visited Index: tests/apps/mod.rs ================================================================== --- tests/apps/mod.rs +++ tests/apps/mod.rs @@ -1,10 +1,10 @@ use std::sync::Arc; use parking_lot::Mutex; -use qsu::rt::{InitCtx, RunEnv, ServiceHandler, SvcEvtReader, TermCtx}; +use qsu::rt::{InitCtx, RunEnv, ServiceHandler, TermCtx}; #[cfg(feature = "tokio")] use qsu::rt::TokioServiceHandler; @@ -68,31 +68,29 @@ } } impl ServiceHandler for MySyncService { - fn init(&mut self, _ictx: InitCtx) -> Result<(), qsu::AppErr> { + type AppErr = Error; + + fn init(&mut self, _ictx: InitCtx) -> Result<(), Self::AppErr> { self.visited.lock().init = true; if self.fail.init { Err(Error::hello("From Sync::init()"))?; } Ok(()) } - fn run( - &mut self, - _re: &RunEnv, - _set: SvcEvtReader - ) -> Result<(), qsu::AppErr> { + fn run(&mut self, _re: &RunEnv) -> Result<(), Self::AppErr> { self.visited.lock().run = true; if self.fail.run { Err(Error::hello("From Sync::run()"))?; } Ok(()) } - fn shutdown(&mut self, _tctx: TermCtx) -> Result<(), qsu::AppErr> { + fn shutdown(&mut self, _tctx: TermCtx) -> Result<(), Self::AppErr> { self.visited.lock().shutdown = true; if self.fail.shutdown { Err(Error::hello("From Sync::shutdown()"))?; } Ok(()) @@ -125,31 +123,29 @@ } #[cfg(feature = "tokio")] #[qsu::async_trait] impl TokioServiceHandler for MyTokioService { - async fn init(&mut self, _ictx: InitCtx) -> Result<(), qsu::AppErr> { + type AppErr = Error; + + async fn init(&mut self, _ictx: InitCtx) -> Result<(), Self::AppErr> { self.visited.lock().init = true; if self.fail.init { Err(Error::hello("From Tokio::init()"))?; } Ok(()) } - async fn run( - &mut self, - _re: &RunEnv, - _set: SvcEvtReader - ) -> Result<(), qsu::AppErr> { + async fn run(&mut self, _re: &RunEnv) -> Result<(), Self::AppErr> { self.visited.lock().run = true; if self.fail.run { Err(Error::hello("From Tokio::run()"))?; } Ok(()) } - async fn shutdown(&mut self, _tctx: TermCtx) -> Result<(), qsu::AppErr> { + async fn shutdown(&mut self, _tctx: TermCtx) -> Result<(), Self::AppErr> { self.visited.lock().shutdown = true; if self.fail.shutdown { Err(Error::hello("From Tokio::shutdown()"))?; } Ok(()) @@ -182,14 +178,16 @@ } #[cfg(feature = "rocket")] #[qsu::async_trait] impl RocketServiceHandler for MyRocketService { + type AppErr = Error; + async fn init( &mut self, _ictx: InitCtx - ) -> Result>, qsu::AppErr> { + ) -> Result>, Self::AppErr> { self.visited.lock().init = true; if self.fail.init { Err(Error::hello("From Rocket::init()"))?; } Ok(Vec::new()) @@ -196,25 +194,24 @@ } async fn run( &mut self, _rockets: Vec>, - _re: &RunEnv, - _set: SvcEvtReader - ) -> Result<(), qsu::AppErr> { + _re: &RunEnv + ) -> Result<(), Self::AppErr> { self.visited.lock().run = true; if self.fail.run { Err(Error::hello("From Rocket::run()"))?; } Ok(()) } - async fn shutdown(&mut self, _tctx: TermCtx) -> Result<(), qsu::AppErr> { + async fn shutdown(&mut self, _tctx: TermCtx) -> Result<(), Self::AppErr> { self.visited.lock().shutdown = true; if self.fail.shutdown { Err(Error::hello("From Rocket::shutdown()"))?; } Ok(()) } } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: tests/err/mod.rs ================================================================== --- tests/err/mod.rs +++ tests/err/mod.rs @@ -6,43 +6,37 @@ IO(String), Qsu(String) } impl std::error::Error for Error {} -impl apperr::Blessed for Error {} impl Error { + #[allow(clippy::needless_pass_by_value)] pub fn hello(msg: impl ToString) -> Self { - Error::Hello(msg.to_string()) + Self::Hello(msg.to_string()) } } impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match self { - Error::Hello(s) => { - write!(f, "Hello error; {}", s) - } - Error::IO(s) => { - write!(f, "I/O error; {}", s) - } - Error::Qsu(s) => { - write!(f, "qsu error; {}", s) - } + Self::Hello(s) => write!(f, "Hello error; {s}"), + Self::IO(s) => write!(f, "I/O error; {s}"), + Self::Qsu(s) => write!(f, "qsu error; {s}") } } } impl From for Error { fn from(err: io::Error) -> Self { - Error::IO(err.to_string()) + Self::IO(err.to_string()) } } -impl From for Error { - fn from(err: qsu::Error) -> Self { - Error::Qsu(err.to_string()) +impl From> for Error { + fn from(err: qsu::CbErr) -> Self { + Self::Qsu(err.to_string()) } } /* /// Convenience converter used to pass an application-defined errors from the @@ -52,12 +46,12 @@ qsu::Error::app(err) } } */ -impl From for qsu::AppErr { - fn from(err: Error) -> qsu::AppErr { - qsu::AppErr::new(err) +impl From for qsu::CbErr { + fn from(err: Error) -> Self { + Self::App(err) } } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: tests/initrunshutdown.rs ================================================================== --- tests/initrunshutdown.rs +++ tests/initrunshutdown.rs @@ -21,16 +21,16 @@ assert!(!visited.run); assert!(!visited.shutdown); let visited = Arc::new(Mutex::new(visited)); - let handler = Box::new(apps::MySyncService { + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new(apps::MySyncService { visited: Arc::clone(&visited), ..Default::default() }); - - let Ok(_) = runctx.run_sync(handler) else { + let Ok(()) = runctx.run_sync(svcevt_handler, rt_handler) else { panic!("run_sync() unexpectedly failed"); }; let visited = Arc::into_inner(visited) .expect("Unable to into_inner Arc") @@ -53,16 +53,16 @@ assert!(!visited.run); assert!(!visited.shutdown); let visited = Arc::new(Mutex::new(visited)); - let handler = Box::new(apps::MyTokioService { + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new(apps::MyTokioService { visited: Arc::clone(&visited), ..Default::default() }); - - let Ok(_) = runctx.run_tokio(None, handler) else { + let Ok(()) = runctx.run_tokio(None, svcevt_handler, rt_handler) else { panic!("run_sync() unexpectedly failed"); }; let visited = Arc::into_inner(visited) .expect("Unable to into_inner Arc") @@ -86,16 +86,17 @@ assert!(!visited.run); assert!(!visited.shutdown); let visited = Arc::new(Mutex::new(visited)); - let handler = Box::new(apps::MyRocketService { + let svcevt_handler = Box::new(move |_msg| {}); + let rt_handler = Box::new(apps::MyRocketService { visited: Arc::clone(&visited), ..Default::default() }); - let Ok(_) = runctx.run_rocket(handler) else { + let Ok(()) = runctx.run_rocket(svcevt_handler, rt_handler) else { panic!("run_sync() unexpectedly failed"); }; let visited = Arc::into_inner(visited) .expect("Unable to into_inner Arc") Index: www/changelog.md ================================================================== --- www/changelog.md +++ www/changelog.md @@ -6,13 +6,43 @@ [Details](/vdiff?from=qsu-0.5.0&to=trunk) ### Added +- The reload service event has been implmented. + - It is not supported on Windows. (The Windows service subsystem offers no + "reload" event). + - On unixy systems it is triggered through the SIGHUP signal. + - When using this feature on systemd, the server type should be set to + `notify-reload`, and qsu will automatically report `RELOADING`, + `MONOTONIC_USEC` or `READY` to systemd as needed. +- Translate SIGUSR1/SIGUSR2 signals to `SvcEvt`'s on unixy platforms. + ### Changed + +- ⚠️ Rename instances of "trace level" to "trace filter". +- ⚠️ `SvcEvt::Shutdown` and `SvcEvt::Terminate` have been merged into the + variant `SvcEvt::Shutdown(Demise)`, where `Demise` can be used to determine + if the service application is terminating via interruption/Ctrl+C, service + shutdown request or if the service application reached reached its end + without any external termination requests. +- ⚠️ Introduce `CbErr` to carry application-specific error type, and use + `Error` to only store library errors. +- Installer: + - Allow display name and description to be set on unsupported platforms, to + allow application code to avoid platform gates. + - On Windows, implicitly add `Tcpip` as a service dependency for services + marked as a netservice. ### Removed + +- ⚠️ The `SvcEvtReader` channel end-point has been removed. Previously this was + passed to the service handlers' `run()` method. Now, instead, the + application passes a service event handler closure when creating the runtime. + To simulate the old behavior, an application can pass a closure that + forwards all incoming `SvcEvt` events to a channel, which has its receiving + end-point stored in the service handler. --- ## [0.5.0] - 2024-05-20 Index: www/design-notes.md ================================================================== --- www/design-notes.md +++ www/design-notes.md @@ -1,71 +1,60 @@ # Design Notes -_qsu_'s primary function is to provide a service runtime laywer that sits +_qsu_'s primary function is to provide a service runtime layer that sits between the operating system's service subsystem and the application server code. It provides abstractions that allow a server to behave in a unified way, regardless of the actual service subsystem type (it even strives to support the same interface and semantics when running as a foreground process). -In addition to the service wrapper runtime, _qsu_ includes: -- Initialization of the `log` and `tracing` crates. -- System service installer/uninstaller wrappers. -- A command line argument parser, based on clap, which defines some common - semantics for registering, deregistering and running services. - - -## Service subsystem integration - -Traditional Unix server processes "daemonize", which involves steps to make -it into an isolated background process. - -Many modern operating systems instead have service subsystems that are -meant to control when and how server processes are launched. These -subsystems tend to actively discourage server applications from -"deamonizing", and instead let the service manager perform whatever -configuration/isolation it needs/wants. - -qsu (currently) does not support deamonizing. This is less due to an -opposition to it, and more due to the need not having arisen (yet). -Instead, it assumes that services are somewhat regular processes, at least on -unixes. - -When built on linux with the `systemd` feature, it is assumed that the -service is a `notify` systemd service type. - -On Windows, services behave very differently. They have a special runtime -that needs to be set up and run. One of the goals of _qsu_ is to hide -this machinery from the server application developer as much as possible. - -As a side-effect attempting to support these widely different systems, but -allowing the same server code to be used to compiled for these different -service subsystem there are interfaces that applications need/should call that -are no-operations on some platforms. - - -## Non-service servers - -When a process is running in an environment that does not have any special -service subsystem integration, _qsu_ will take steps in order to emulate -it. In practical terms, this means: - -- On unixy platforms, signal handlers are installed to intercept SIGINT/SIGTERM - (to signal a server termination event to the application) and SIGHUP (to - signal "reload configuration" event). -- On Windows a Ctrl+C/Break handler is installed (to signal a server - termination event to the application). - - -## Shut down - -A shutdown occurs when the application's `run()` implementation returns to its -caller. The _qsu_ runtime can send an event to the application to inform it -that it should shut down. - -On unixy platforms the shutdown event is originates when a SIGINT/SIGTERM -signals has been intercepted. On Windows when running as a foreground process -the shutdown even originates from a Ctrl+C/Break handlers. When running a -Windows Service, the shutdown event originates from the system's service -dispatcher. - -The server application _must_ immediately obey the shutdown request. +In order to write a service application, the developer first chooses a runtime +to integrate against. This basically refers to choosing whether the service +runs async or non-async code. Next they implement a runtime type specific +trait, which has three methods `init()`, `run()` and `shutdown()`. + +Some service subsystems have special "startup" and "shutdown" phases. The +service handler's trait methods `init()` and `shutdown()` correspond to these +phases, and qsu will report (as appropriate) to the service subsystem that +these phases have been entered as the appropriate callback methods are called. + +An application also needs to set up a service event handler. This is a closure +that will receive events from the service subsystem. When running as a +foreground process, qsu simulates the same behavior so the application will +receive the same events. Specifically, when a Window Service is stopped, a +`SvcEvt::Shutdown(_)` event will be received by the event handler. When a +service application is run in foreground mode, pressing Ctrl+C or Ctrl+Break +will cause the same `SvcEvt::Shutdown(_)` to be sent. + +Once the service handler trait has been implemented and the service event +closure has been created, the qsu service runtime is called, passing along the +two handlers. At this point qsu will initialize logging and then call the +service handler's `init()` implementation, and if successful it will call its +`run()` implementation and wait until it returns before calling the +`shutdown()` impelementation and then finally terminate the runtime. + +Applications must themselves translate shutdown events received by the service +event handler to an actual termination of the service application. This is +typically done using channels, or cancellation tokens. + + +## Errors + +There are several application callbacks in qsu; primarily in the service +application runtime, but also in the argument parser. It is recognized that +service applications that encounter fatal errors may want to return the +application-specific error back to the original caller (for instance to return +different appropriate exit codes to the calling shell). To allow this +callbacks return an error type `CbErr` which is generic over an +application-defined type used to carry the application error. + +However, there are parts of qsu that doesn't have need for this, even parts +that become needlessly complicated having to support the generic parameter, so +there's an `Error` type that does not carry the `ApEr` generic. + +`CbErr` can carry the `Error` type within one of its variants. + + +## ToDo + +- Document argument parser +- Document the service installer facilities Index: www/index.md ================================================================== --- www/index.md +++ www/index.md @@ -1,104 +1,54 @@ # qsu -Portable service utilities with an (opinionated) service wrapper runtime. - - -## The "What?" - -The _qsu_ ("kazoo") crate is a: - -- service runtime which acts as a layer betwen the server application code - and an operating system service subsystem (launchd, systemd, Windows - Services), and emulates what it needs when running in a non-service (a.k.a. - foreground) environment in order to avoid implementation divergence for the - different environemtns. -- set of utility functions for working with services. - -_qsu_ is somewhat opinionated and is designed to make certain types of -server software easier to write, but it does not aim to cover every -use-case. - - -## The "Why?" - -The original motivating factors for this crate was the observation that -a fair amount of time was spent on making server applications work in different -service and non-service environments. This time could be better spent working -on the actual server application code. - - -## The "How?" - -See [design notes](./design-notes.md) for a high-level overview of how the -_qsu_ wrapper runtime is implemented. - - -## What qsu is opinionated about - -It should be noted that the word _opinionated_ does not necessary mean -that qsu is forever locked to these behaviors -- it's just to say -what the current version does. - - -### All platforms -- Uses both the `log` crate and `tracing` crate. - - `log` is intended for production logging. - - `tracing` is intended for developer and debugging logging. -- The optional built-in command line parser assumes that there are at least - three subcommands (used to register, deregister and run service). - -### Unixy platforms and running as a foreground process in Windows -- The following environment variables control logging/tracing: - - `LOG_LEVEL` is used to control the logging level for the `log` crate. - - `TRACE_LEVEL` is used to control the tracing level for the `tracing` - crate. - - If `TRACE_FILE` is set tracing will be directed to a file instead of - being printed to the console. - -### Unixy platforms -- Signal management: - - SIGINT/SIGTERM is translated to a server termination request that the - server application must honor. - - SIGHUP is interpreted as a request to "reload configuration". - -### Windows foreground process -- Captures Ctrl+C and Ctrl+Break and translates them to service event - messages. - -### Windows Service -- The service installation registers the service name as an event source - for the Windows Events Log. -- Application-specific configuration is stored in the registry under - `HKLM:\SYSTEM\CurrentControlSet\Services\[service_name]\Parameters` -- Within the service's registry `Parameters` subkey: - - The key `LogLevel` takes the role of the `LOG_LEVEL` environment - vairable. - - If key `TraceFile` and `TraceLevel` correspond to the environment - variables `TRACE_FILE` and `TRACE_LEVEL`. Both these must be - configured in the registry to enable tracing. -- Logging through `log` will log to the Windows Events Log. -- Logging using `trace` will write trace logs to a file. - - -## When to use it, and when not to use it - -To be frank, if you're writing a systemd-only service, then the value of using -_qsu_ is negligible (or it might even be wasteful to pull in _qsu_). The -benefits of using _qsu_ will be noticed mostly when targeting the Windows -Services subsystem. But mostly the benefits become apparent when targetting -multiple service subsystems in the same project, and wanting to have a similar -API when developing non-async and async services. - - -## General tips - -- The logging/tracing facilities aren't initialized until the server - application's runtime has been initialized, because the runtime type may - affect the logging/tracing backends. As an implication of this, services - that use the `ArgParser` should defer operations that need logging/tracking - until the service handler's `init()` trait method is called. +The _qsu_ ("kazoo") crate offers portable service utilities with an +([opinionated](./opinionated.md)) service wrapper runtime. + +_qsu_'s primary objective is to allow a service developer to focus on the +actual service application code, without having to bother with service +subsystem-specific integrations -- while at the same time allowing the +service application to run as a regular foreground process, without the code +needing to diverge between the two. + + +## Features + +- _qsu_ offers two primary runtime types: + - `Sync` is for "non-async" service applications. + - `Tokio` is for [tokio](https://tokio.rs/)-based async service applications. +- In addition is offers special-purpose service runtimes for: + - `Rocket` is an tokio/async runtime environment with helpers for + [Rocket](https://rocket.rs/) applications. +- The runtime uses two logging systems; [`log`](https://crates.io/crates/log) + and [`tracing`](https://crates.io/crates/tracing) -- it assumes that `log` is + used for production logs, and `tracing` is used for developer logs. +- It offers an optional command line argument parser, based on + [clap](https://crates.io/crates/clap), that provides a (configurable) + command line interface for registering, running and deregistering the + service. +- It provides an installation module that can register/deregister services on + Windows, and generate system service unit files for systemd or launcher plist + files for launchd. + + +## Known limitations + +- There are several assumptions made in _qsu_ about paths being utf-8. + Even if a public interface takes a `Path` or `PathBuf`, the function may + return an error if the path isn't utf-8 compliant. + + +## Further reading + +- [How to use the qsu runtime](./qsurt.md) - This should be read by both + both developers and administrators using a service written on top of qsu. +- See [design notes](./design-notes.md) for a high-level overview of how + the _qsu_ wrapper runtime is implemented. +- Examples: + - The source repository contains several qsu examples. + - The [staticrocket](https://crates.io/crates/staticrocket) crate uses qsu. + ## Feature labels in documentation The crate's documentation uses automatically generated feature labels, which currently requires nightly featuers. To build the documentation locally use: @@ -105,31 +55,10 @@ ``` RUSTFLAGS="--cfg docsrs" RUSTDOCFLAGS="--cfg docsrs" cargo +nightly doc --all-features ``` - -## Known limitations - -- There are several assumptions made in _qsu_ about paths being utf-8. - Even if a public interface takes a `Path` or `PathBuf`, the function may - return an error if the path isn't utf-8 compliant. - - -## Examples - -- The repository contains three different in-tree examples: - - `hellosvc` is a "sync" (read: non-`async`) server application which dumps - logs and traces every 30 seconds until the service is terminated. - - `hellosvc-tokio` is the same as `hellosvc`, but is an `async` server that - runs on top of tokio. - - `hellosvc-rocket` is a Rocket server that writes logs and traces each time - a request it made to the index page. -- The [staticrocket](https://crates.io/crates/staticrocket) crate uses qsu. - In particular it implements the `Rocket` service handler, and it adds a - custom application-specific subcommand to the `ArgParser`. - ## Change log The details of changes can always be found in the timeline, but for a high-level view of changes between released versions there's a manually ADDED www/opinionated.md Index: www/opinionated.md ================================================================== --- /dev/null +++ www/opinionated.md @@ -0,0 +1,46 @@ +# Why is qsu opinonated, and what is it opinonated about? + +Service subsystems can be complicated and offer a lot of features. Even +systemd services, which are relatively simple compared to the Windows Service +subsystem, has a lot of features and service integration possibilities. + +It is not the goal of qsu to be able to express every type of service, its goal +is merely to facilitate the development of some hand-wavy "normal" service, +where the developer should minimally need to worry about platform/subsystem +specific integrations. In order to accomplish this it makes a few assumptions: + +- Some features, like configuration reload events, are available on all + platforms, but they will never be generated on some (for instance Windows + service subsystem will never generate them). +- Linux has different subsystems to manage services. qsu only allows special + integration against systemd, and is limited to the `notify` and + `notify-reload` service types. +- There are two types of logs; "production logs" are for end-users. "developer + logs" are for developers and troubleshooting only. qsu uses the `log` crate + for "production logs" and `tracing` crate for "developer logs". It assumes + ownership of these runtimes, in the sense that it will initialize them and + assumes no one else does so. +- While developing service applications the developer may not want to have to + install the program as a service, but run it as a regular foreground program. + Doing this should require minimal diverging code. +- qsu claims ownership of certain signals on unixy platforms. On Windows + it claims ownership of Ctrl+C and Ctrl+Break when running in foreground mode. + + +## Agnostic to the developer, but idiomatic to the end-user + +While it's a goal to provide a simple and unified developer experience when +writing service applications (regardless of the service subsystem the +application is run on top of), the same is not true from the end-user +administrator's perspective. It can be very annoying when foreign idioms are +brought into a system -- it's not unreasonable to assume that people want new +programs to behave like all the other programs they are accustomed to on their +platform. + +As an example, Windows administrators will likely expect certain service +configuration parameters to be stored in the registry, under the service's +reguistry subkey. qsu caters to this, and this will inevitably bleed into the +developer's domain -- but the goal to limit subsystem-specifics as much as +possible for the developer, while enabling idiomatic behaviors for the system +administrators. + ADDED www/qsurt.md Index: www/qsurt.md ================================================================== --- /dev/null +++ www/qsurt.md @@ -0,0 +1,87 @@ +# Using the qsu runtime + +A service using the qsu runtime can run in two different contexts. One is +(perhaps not optimally) referred to as running in the _foreground_, and the +other is referred to as running _as a service_. + +The intended use-case for running a service in foreground mode is during +development and initial testing and troubleshooting of a service application. +Normally when running in this mode, the service is run from the command line, +in a terminal window, and all its logs are written to the terminal. + +Running _as a service_ means that the service application is assumed to be +running under a service subsystem, such a the Windows Service subsystem, or +systemd on linux. Logs will be sent to the appropriate system-specific logging +system (Windows Event Log on Windows, the systemd journal on linux with +systemd). Starting and stopping a qsu application running as a service should +be done using the regular system service management tools. + +When a qsu application uses the qsu argument parser, running the command +without any subcommand will cause it to run in foreground mode. Running it +with the subcommand `run-service` (though this is configurable) the service +application will integrate into the platform's native service subsystem as +appropriate. + + +## Configuring logging + +When running in foreground mode, production logging is controlled using the +`LOG_LEVEL` environment variable. It can be set to `none`, `error`, `warn`, +`info`, `debug` and `trace` (in order of increasing verbosity). If the +variable is not set, it will default to `warn`. + +To configure the development logs, use the `TRACE_FILTER` environment +variable instead. This log uses the +[`tracing_subscriber`](https://crates.io/crates/tracing_subscriber) crate's +environment filter. The filter format is feature rich and is is out of scope +in this document, but in its simplest form it can be set to the same values +as `LOG_LEVEL`. In order to reduce noise it can be configured to only observe +certain modules. This is accomplished by disabling all logs, and then only +enabling a subset: `TRACE_FILTER="none,qsu::rt=trace"`. The development log +defaults to `none`. + +In some cases it may be helpful to write the development logs to a file. This +can be done by setting the `TRACE_FILE` environment variable to the path and +filename of the logfile to write developer logs to. + +On unixy platforms, the same environment variables (`LOG_LEVEL`, +`TRACE_FILTER`, `TRACE_FILE`) control logging when running as a service. + +On Windows, when running as a service, production logging will log to the +Windows Events Log. The development logs will be lost unless written to a +file. The logging is controlled in the registry under +`HKLM:\SYSTEM\CurrentControlSet\Services\[service_name]\Parameters`, where: + +- The key `LogLevel` takes the role of the `LOG_LEVEL` environment + vairable. +- The keys `TraceFile` and `TraceFilter` correspond to the environment + variables `TRACE_FILE` and `TRACE_LEVEL`. Both these must be configured in + the registry to enable development logging. + + +## Unixy platforms & signals + +On unixy platforms, the qsu runtime claims ownership of certain signals. +`SIGINT` and `SIGTERM` are used to indicate that the service should shut down. +`SIGHUP` can be used to instruct the service application to reload its +configuration. `SIGUSR1` and `SIGUSR2` are translated to custom +application-specific evnts. + +Applications/libraries must avoid disrupting qsu's signal handlers. + + +## Windows foreground + +When running in foreground mode, the qsu runtime on Windows claims +ownership of the `Ctrl+C` and `Ctrl+Break` handlers. Applications/libraries +must avoid disrupting these. + + +## Windows service + +When running under the Windows Service subsystem, qsu will check if the +service's `Parameters` registry subkey contains a string called `WorkDir`. If +this is set it will set the process current working directory to that directory +before calling the service handler's `init()`. + +