qsu/

rt.rs

//! Server application wrapper runtime.
//!
//! # Overview
//! The _qsu_'s runtime lives in a layer between the actual server application
//! and the "service subsystems" that servier applications can integrate
//! against:
//!
//! <pre>
//! +---------------------+
//! |  Server Application |  (HTTP server, Matrix server, etc)
//! +---------------------+
//! |       qsu rt        |
//! +---------------------+
//! |  Platform service   |  (Windows Service subsystem, systemd, etc)
//! |      subsystem      |
//! +---------------------+
//! </pre>
//!
//! 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 subsystem at all; such as when running the
//! server application as a regular foreground process (which is common during
//! development).
//!
//! # How service applications are implemented and used
//! 1. Choose service application type.
//!
//!    _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.
//!    - [`RocketServiceHandler`] is for server applications that are built on
//!      top of the Rocket HTTP framework.
//! 2. Fot the selected service handler type, implement the three methods:
//!    - `init()` for initializing the service.
//!    - `run()` is for running the actual server application.
//!    - `shutdown()` is for shutting down the server application.
//! 3. Once a service trait has been implemented, the application should create
//!    a [`RunCtx`] object and call its [`run()`](RunCtx::run()) method,
//!    passing in the service implementation object.
//!
//! ## Service Handler semantics
//! When a handler is run through the [`RunCtx`] it will first call the
//! handler's `init()` method.  If it returns `Ok()`, its `run()` method will
//! be run.
//!
//! 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).
//!
//! # Passing data to `init()` and `shutdown()`
//! It's not uncommon that an application may have data that it wishes to pass
//! to the service handler's `init()` or `shutdown()` handlers.  This can be
//! accomplished by storing the data in the type that implements the service
//! handler trait.  However, if there's a lot of data this can cause it to
//! become cluttered with things that don't really belong there (they only use
//! the type as a temporary transport mechanism).  As a way to avoid the
//! clutter, the [`RunCtx`] can be handed types that can be picked up in
//! `init()`/`shutdown()` (using `InitCtx` and `TermCtx`).
//!
//! Use [`RunCtx::init_passthrough()`]/[`RunCtx::init_passthrough_r()`] to pass
//! data to the `init()` handler, and use
//! [`RunCtx::term_passthrough()`]/[`RunCtx::term_passthrough_r()`] to pass
//! data to the `shutdown()` handler.
//!
//! # Logging
//! The runtime will implicitly initialize production logging and development
//! logging.  To see how these can be configured, see
//! [lumberjack](crate::lumberjack).
//!
//! # Argument parser
//! _qsu_ offers an [argument parser](crate::argp::ArgParser), which can
//! abstract away much of the runtime management and service registration.

mod nosvc;
mod rttype;
mod signals;

#[cfg(all(target_os = "linux", feature = "systemd"))]
#[cfg_attr(docsrs, doc(cfg(feature = "systemd")))]
mod systemd;

#[cfg(windows)]
pub mod winsvc;

use std::{
  any::{Any, TypeId},
  sync::{
    Arc, OnceLock,
    atomic::{AtomicU32, Ordering}
  }
};

use hashbrown::HashMap;

#[cfg(any(feature = "tokio", feature = "rocket"))]
use async_trait::async_trait;

#[cfg(feature = "tokio")]
use tokio::runtime;

use tokio::sync::broadcast;

#[cfg(all(target_os = "linux", feature = "systemd"))]
use sd_notify::NotifyState;


use crate::{err::CbErr, lumberjack::LumberJack};


#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum RunAs {
  Foreground,
  SvcSubsys
}

/// Keep track of whether the service application is running as foreground
/// process or running under a service subsystem.
static RUNAS: OnceLock<RunAs> = OnceLock::new();

pub fn runas() -> Option<RunAs> {
  RUNAS.get().copied()
}


/// The run time environment.
///
/// Can be used by the application callbacks to determine whether it is running
/// as a service.
#[derive(Debug, Clone)]
pub enum RunEnv {
  /// Running as a foreground process.
  Foreground,

  /// Running as a service.
  ///
  /// If the service subsystem has named services, this will contain the
  /// service name.
  Service(Option<String>)
}


/// Report the current startup/shutdown state to the platform service
/// subsystem.
pub(crate) trait StateReporter {
  /// Inform the service subsystem that initialization is running.
  fn starting(&self, checkpoint: u32, msg: &str);

  /// Inform the service subsystem that initialization has completed.
  fn started(&self);

  /// Inform the service subsystem that termination is running.
  fn stopping(&self, checkpoint: u32, msg: &str);

  /// Inform the service subsystem that termination has completed.
  fn stopped(&self);
}


/// A wrapper around a service state reporter backend implementation.
#[derive(Clone)]
#[repr(transparent)]
pub(crate) struct ServiceReporter(Arc<dyn StateReporter + Send + Sync>);

impl ServiceReporter {
  pub(crate) fn new<SR>(sr: SR) -> Self
  where
    SR: StateReporter + Send + Sync + 'static
  {
    Self(Arc::new(sr))
  }

  pub(crate) fn starting(&self, checkpoint: u32, status: Option<&str>) {
    let text = status.as_ref().map_or_else(
      || format!("Starting[{checkpoint}]"),
      |msg| format!("Starting[{checkpoint}] {msg}")
    );
    self.0.starting(checkpoint, &text);
    log::debug!("{text}");
  }

  pub(crate) fn started(&self) {
    self.0.started();
    log::debug!(
      "Service initialization has finished and is entering running state"
    );
  }

  pub(crate) fn stopping(&self, checkpoint: u32, status: Option<&str>) {
    let text = status.as_ref().map_or_else(
      || format!("Stopping[{checkpoint}]"),
      |msg| format!("Stopping[{checkpoint}] {msg}")
    );
    self.0.stopping(checkpoint, &text);
    log::debug!("{text}");
  }

  pub(crate) fn stopped(&self) {
    self.0.stopped();
    log::debug!("Service termination has finished");
  }
}


/// Context passed to `init()` service application callback.
///
/// This context can be used to query whether the service application is
/// running as foreground process or running within a service subsystem.
///
/// It can also be used to report progress status back to the service
/// subsystems, for platforms that support it.
pub struct InitCtx {
  re: RunEnv,
  sr: ServiceReporter,
  cnt: Arc<AtomicU32>,
  passthrough: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
  passthrough_term: HashMap<TypeId, Box<dyn Any + Send + Sync>>
}

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.
  ///
  /// For foreground processes and services that do not support startup state
  /// notifications this method has no effect.
  pub fn report(&self, status: Option<&str>) {
    let checkpoint = self.cnt.fetch_add(1, Ordering::SeqCst);
    if let Some(msg) = status {
      tracing::trace!("Reached init checkpoint {checkpoint}; {}", msg);
    } else {
      tracing::trace!("Reached init checkpoint {checkpoint}");
    }
    self.sr.starting(checkpoint, status);
  }

  #[must_use]
  pub fn get<T>(&self) -> Option<&T>
  where
    T: Send + 'static
  {
    self
      .passthrough
      .get(&TypeId::of::<T>())
      .and_then(|boxed| boxed.downcast_ref::<T>())
  }

  /// Extract passthrough buffer.
  pub fn take<T>(&mut self) -> Option<T>
  where
    T: Send + 'static
  {
    self
      .passthrough
      .remove(&TypeId::of::<T>())
      .and_then(|boxed| boxed.downcast::<T>().ok())
      .map(|v| *v)
  }

  pub fn term_passthrough<T>(&mut self, value: T)
  where
    T: Send + Sync + 'static
  {
    self
      .passthrough_term
      .insert(TypeId::of::<T>(), Box::new(value));
  }
}

impl Drop for InitCtx {
  fn drop(&mut self) {
    let checkpoint = self.cnt.fetch_add(1, Ordering::SeqCst);
    self
      .sr
      .starting(checkpoint, Some("Initialization phase finished"));
  }
}


/// Context passed to `term()` service application callback.
///
/// This context can be used to query whether the service application is
/// running as foreground process or running within a service subsystem.
///
/// It can also be used to report progress status back to the service
/// subsystems, for platforms that support it.
pub struct TermCtx {
  re: RunEnv,
  sr: ServiceReporter,
  cnt: Arc<AtomicU32>,
  passthrough: HashMap<TypeId, Box<dyn Any + Send + Sync>>
}

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.
  ///
  /// For foreground processes and services that do not support shutdown state
  /// notifications this method has no effect.
  pub fn report(&self, status: Option<&str>) {
    let checkpoint = self.cnt.fetch_add(1, Ordering::SeqCst);
    if let Some(msg) = status {
      tracing::trace!("Reached term checkpoint {checkpoint}; {msg}");
    } else {
      tracing::trace!("Reached term checkpoint {checkpoint}");
    }
    self.sr.stopping(checkpoint, status);
  }

  #[must_use]
  pub fn get<T: 'static>(&self) -> Option<&T> {
    self
      .passthrough
      .get(&TypeId::of::<T>())
      .and_then(|boxed| boxed.downcast_ref::<T>())
  }

  /// Extract passthrough buffer.
  pub fn take<T: 'static>(&mut self) -> Option<T> {
    self
      .passthrough
      .remove(&TypeId::of::<T>())
      .and_then(|boxed| boxed.downcast::<T>().ok())
      .map(|v| *v)
  }
}

impl Drop for TermCtx {
  fn drop(&mut self) {
    let checkpoint = self.cnt.fetch_add(1, Ordering::SeqCst);
    self
      .sr
      .stopping(checkpoint, Some("Termination phase finished"));
  }
}


/// "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 {
  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: &mut InitCtx) -> Result<(), Self::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>;

  /// 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: &mut TermCtx) -> Result<(), Self::AppErr>;
}


/// `async` server application built on the tokio runtime.
///
/// Implement this for an object that wraps a server application that uses
/// tokio as an async runtime.
#[cfg(feature = "tokio")]
#[cfg_attr(docsrs, doc(cfg(feature = "tokio")))]
#[async_trait]
pub trait TokioServiceHandler {
  type AppErr;

  async fn init(&mut self, ictx: &mut InitCtx) -> Result<(), Self::AppErr>;

  async fn run(&mut self, re: &RunEnv) -> Result<(), Self::AppErr>;

  async fn shutdown(&mut self, tctx: &mut TermCtx)
  -> Result<(), Self::AppErr>;
}


/// Rocket server application handler.
///
/// While Rocket is built on top of tokio, it \[Rocket\] wants to initialize
/// tokio itself.
///
/// There are two major ways to write Rocket services using qsu; either the
/// application can let qsu be aware of the server applications' `Rocket`
/// instances.  It does this by creating the `Rocket` instances in
/// `RocketServiceHandler::init()` and returns them.  _qsu_ will ignite these
/// rockets and pass them to `RocketServiceHandler::run()`.  The application is
/// responsible for launching the rockets at this point.
///
/// The other way to do it is to completely manage the `Rocket` instances in
/// application code (by not returning rocket instances from `init()`).
///
/// Allowing _qsu_ to manage the `Rocket` instances will cause _qsu_ to request
/// graceful shutdown of all `Rocket` instances once a `SvcEvt::Shutdown` is
/// sent by the runtime.
///
/// It is recommended that `ctrlc` shutdown and termination signals are
/// disabled in each `Rocket` instance's configuration, and allow the _qsu_
/// 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: &mut InitCtx
  ) -> Result<Vec<rocket::Rocket<rocket::Build>>, Self::AppErr>;

  /// Server application main entry point.
  ///
  /// If the `init()` trait method returned `Rocket<Build>` 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<rocket::Rocket<rocket::Ignite>>,
    re: &RunEnv
  ) -> Result<(), Self::AppErr>;

  async fn shutdown(&mut self, tctx: &mut 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,

  /// SIGUSR2
  Sig2
}


/// Event notifications that originate from the service subsystem that is
/// controlling the server application.
#[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,

  /// Service subsystem has requested that the server application should
  /// resume its operations.
  ///
  /// Only the Windows service subsystem will emit these events.
  Resume,

  /// Service subsystem has requested that the services configuration should
  /// be reread.
  ///
  /// On Unixy platforms this is triggered by SIGHUP, and is unsupported on
  /// Windows.
  ReloadConf,

  /// The service application is terminating.  The `Demise` value indicates
  /// the reason for the shutdown.
  Shutdown(Demise)
}


/// The server application runtime type.
// 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<ApEr> {
  /// A plain non-async (sometimes referred to as "blocking") server
  /// application.
  Sync {
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn ServiceHandler<AppErr = ApEr> + 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 {
    rtbldr: Option<runtime::Builder>,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn TokioServiceHandler<AppErr = ApEr> + 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 {
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn RocketServiceHandler<AppErr = ApEr> + Send>
  }
}


/// Service runner context.
pub struct RunCtx {
  service: bool,
  svcname: String,
  log_init: bool,
  passthrough_init: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
  passthrough_term: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
  test_mode: bool
}

impl RunCtx {
  /// Run as a systemd service.
  #[cfg(all(target_os = "linux", feature = "systemd"))]
  fn systemd<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + std::fmt::Debug
  {
    LumberJack::default()
      .set_init(self.log_init)
      .service()
      .init()?;

    tracing::debug!("Running service '{}'", self.svcname);

    let sr = systemd::ServiceReporter {};
    let sr = ServiceReporter::new(sr);

    let re = RunEnv::Service(Some(self.svcname.clone()));

    match st {
      SrvAppRt::Sync {
        svcevt_handler,
        rt_handler
      } => rttype::sync_main(rttype::SyncMainParams {
        re,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None,
        passthrough_init: self.passthrough_init,
        passthrough_term: self.passthrough_term,
        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,
          passthrough_init: self.passthrough_init,
          passthrough_term: self.passthrough_term
        }
      ),
      #[cfg(feature = "rocket")]
      SrvAppRt::Rocket {
        svcevt_handler,
        rt_handler
      } => rttype::rocket_main(rttype::RocketMainParams {
        re,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None,
        passthrough_init: self.passthrough_init,
        passthrough_term: self.passthrough_term
      })
    }
  }

  /// Run as a Windows service.
  #[cfg(windows)]
  fn winsvc<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    winsvc::run(
      &self.svcname,
      st,
      self.passthrough_init,
      self.passthrough_term
    )?;

    Ok(())
  }

  /// Run as a foreground server
  fn foreground<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + std::fmt::Debug
  {
    LumberJack::default().set_init(self.log_init).init()?;

    tracing::debug!("Running service '{}'", self.svcname);

    let sr = nosvc::ServiceReporter {};
    let sr = ServiceReporter::new(sr);

    match st {
      SrvAppRt::Sync {
        svcevt_handler,
        rt_handler
      } => rttype::sync_main(rttype::SyncMainParams {
        re: RunEnv::Foreground,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None,
        passthrough_init: self.passthrough_init,
        passthrough_term: self.passthrough_term,
        test_mode: self.test_mode
      }),

      #[cfg(feature = "tokio")]
      SrvAppRt::Tokio {
        rtbldr,
        svcevt_handler,
        rt_handler
      } => rttype::tokio_main(
        rtbldr,
        rttype::TokioMainParams {
          re: RunEnv::Foreground,
          svcevt_handler,
          rt_handler,
          sr,
          svcevt_ch: None,
          passthrough_init: self.passthrough_init,
          passthrough_term: self.passthrough_term
        }
      ),

      #[cfg(feature = "rocket")]
      SrvAppRt::Rocket {
        svcevt_handler,
        rt_handler
      } => rttype::rocket_main(rttype::RocketMainParams {
        re: RunEnv::Foreground,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None,
        passthrough_init: self.passthrough_init,
        passthrough_term: self.passthrough_term
      })
    }
  }
}

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,
      passthrough_init: HashMap::new(),
      passthrough_term: HashMap::new(),
      test_mode: false
    }
  }

  /// Set an initialization passthrough buffer.
  ///
  /// The `data` can be extracted in the service handler's `init()` method by
  /// calling `InitCtx::take_passthrough()`.
  #[must_use]
  pub fn init_passthrough<T>(mut self, data: T) -> Self
  where
    T: Send + Sync + 'static
  {
    self.init_passthrough_r(data);
    self
  }

  /// Same as [`RunCtx::init_passthrough()`], but works with reference to
  /// `Self`.
  pub fn init_passthrough_r<T>(&mut self, data: T) -> &mut Self
  where
    T: Send + Sync + 'static
  {
    self
      .passthrough_init
      .insert(TypeId::of::<T>(), Box::new(data));
    self
  }

  /// Set an termination passthrough buffer.
  ///
  /// The `data` can be extracted in the service handler's `init()` method by
  /// calling `InitCtx::take_passthrough()`.
  #[must_use]
  pub fn term_passthrough<T>(mut self, data: T) -> Self
  where
    T: Send + Sync + 'static
  {
    self.term_passthrough_r(data);
    self
  }

  /// Same as [`RunCtx::term_passthrough()`], but works with reference to
  /// `Self`.
  pub fn term_passthrough_r<T>(&mut self, data: T) -> &mut Self
  where
    T: Send + Sync + 'static
  {
    self
      .passthrough_term
      .insert(TypeId::of::<T>(), Box::new(data));
    self
  }

  /// Enable test mode.
  ///
  /// This method is intended for tests only.
  ///
  /// 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)]
  #[must_use]
  pub const fn test_mode(mut self) -> Self {
    self.log_init = false;
    self.test_mode = true;
    self
  }

  /// 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)]
  #[must_use]
  pub const fn log_init(mut self, flag: bool) -> Self {
    self.log_init = flag;
    self
  }

  /// Reference version of [`RunCtx::log_init()`].
  #[doc(hidden)]
  pub const fn log_init_ref(&mut self, flag: bool) -> &mut Self {
    self.log_init = flag;
    self
  }

  /// Mark this run context to run under the operating system's subservice, if
  /// one is available on this platform.
  #[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
  /// one is available on this platform.
  pub const fn service_ref(&mut self) -> &mut Self {
    self.service = true;
    self
  }

  #[must_use]
  pub const fn is_service(&self) -> bool {
    self.service
  }

  /// Launch the application.
  ///
  /// If this `RunCtx` has been marked as a _service_ then it will perform the
  /// 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.
  ///
  /// # 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<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    if self.service {
      let _ = RUNAS.set(RunAs::SvcSubsys);

      #[cfg(all(target_os = "linux", feature = "systemd"))]
      self.systemd(st)?;

      #[cfg(windows)]
      self.winsvc(st)?;

      // ToDo: We should check for other platforms here (like macOS/launchd)
    } else {
      let _ = RUNAS.set(RunAs::Foreground);

      // Do not run against any specific service subsystem.  Despite its name
      // this isn't necessarily running as a foreground process; some service
      // subsystems do not make a distinction.  Perhaps a better mental model
      // is that certain service subsystems expects to run regular "foreground"
      // processes.
      self.foreground(st)?;
    }

    Ok(())
  }

  /// Convenience method around [`Self::run()`] using [`SrvAppRt::Sync`].
  #[allow(clippy::missing_errors_doc)]
  pub fn run_sync<ApEr>(
    self,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn ServiceHandler<AppErr = ApEr> + Send>
  ) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    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")))]
  #[allow(clippy::missing_errors_doc)]
  pub fn run_tokio<ApEr>(
    self,
    rtbldr: Option<runtime::Builder>,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn TokioServiceHandler<AppErr = ApEr> + Send>
  ) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    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")))]
  #[allow(clippy::missing_errors_doc)]
  pub fn run_rocket<ApEr>(
    self,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn RocketServiceHandler<AppErr = ApEr> + Send>
  ) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    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<SvcEvt>,
  mut evt_handler: Box<dyn FnMut(SvcEvt) + Send>
) {
  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(&[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(&[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 :