ruststream 0.7.0-rc.12

//! The process entry point generated by [`#[ruststream::app]`](macro@crate::app).
//!
//! An app crate exposes a builder function (`fn app() -> impl App`, or a concrete
//! `RustStream<_>`) and marks it with the attribute macro, which expands to a `main` that calls
//! [`run_main`]. That gives a zero-boilerplate binary which understands two commands:
//!
//! - `run` (the default) builds the Tokio runtime the attribute names (a multi-threaded one
//!   unless `flavor = "current_thread"` or `worker_threads = n` says otherwise, see
//!   [`AppRuntime`]) and runs the service until an interrupt, mirroring [`RustStream::run`](crate::runtime::RustStream::run). With the `logging`
//!   feature enabled it first installs
//!   the colored console logger ([`crate::logging`]), so a scaffolded service prints logs without
//!   any setup; an app that installs its own subscriber keeps it.
//! - `asyncapi gen [-o <file>] [--yaml]` builds the [`AsyncAPI`](crate::asyncapi) document for the
//!   service and writes it to stdout or a file. Requires the crate to enable the `asyncapi` feature.
//!
//! The standalone `ruststream` CLI drives these by shelling out to `cargo run -- <command>`, so the
//! same dispatch serves both `cargo run` and `ruststream run`.
//!
//! # The `ruststream` command
//!
//! `cargo install ruststream --features cli` installs the standalone command. It drives
//! `cargo run` for the target crate, so a service needs nothing beyond the attribute:
//!
//! ```text
//! ruststream run                         # cargo run -- run, against ./Cargo.toml
//! ruststream run -p ./my-service         # against another crate
//! ruststream run --release               # release build
//! ruststream asyncapi gen                # print the AsyncAPI document
//! ruststream asyncapi gen -o spec.json   # write it to a file
//! ruststream asyncapi gen --yaml         # YAML instead of JSON
//! ```
//!
//! A new project is scaffolded with `cargo generate` from a template the broker crate ships,
//! typically one per transport or topology. The starter template for the in-memory broker lives
//! in this repository, and the quick start has the command:
//! <https://powersemmi.github.io/ruststream/latest/getting-started/quickstart/>.

use std::io;
use std::num::NonZeroUsize;
use std::process::ExitCode;

use thiserror::Error;
use tokio::runtime::{Builder, Runtime};

use super::app::{App, RustStreamError};

/// Errors surfaced by the generated CLI entry point.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum CliError {
    /// The Tokio runtime could not be built.
    #[error("failed to build the async runtime: {0}")]
    Runtime(#[source] io::Error),
    /// The service returned an error while running (see [`RustStreamError`]).
    #[error(transparent)]
    Run(#[from] RustStreamError),
    /// The first argument was not a recognized command.
    #[error("unknown command {0:?}; expected `run` or `asyncapi gen`")]
    UnknownCommand(String),
    /// `asyncapi` was followed by something other than `gen`.
    #[error("unknown `asyncapi` subcommand {0:?}; expected `gen`")]
    UnknownAsyncApi(String),
    /// An unrecognized flag was passed to `asyncapi gen`.
    #[error("unknown option {0:?} for `asyncapi gen`")]
    UnknownOption(String),
    /// `asyncapi gen` was requested but the crate does not enable the `asyncapi` feature.
    #[error("`asyncapi gen` requires the crate to enable ruststream's `asyncapi` feature")]
    AsyncApiDisabled,
    /// The generated spec could not be written to the requested file.
    #[error("could not write the spec to {path}: {source}")]
    WriteSpec {
        /// The output path that failed.
        path: String,
        /// The underlying I/O error.
        #[source]
        source: io::Error,
    },
    /// The spec could not be serialized to JSON.
    #[cfg(feature = "asyncapi")]
    #[error("could not serialize the AsyncAPI spec to JSON: {0}")]
    SerializeJson(#[from] serde_json::Error),
    /// The spec could not be serialized to YAML.
    #[cfg(feature = "asyncapi")]
    #[error("could not serialize the AsyncAPI spec to YAML: {0}")]
    SerializeYaml(#[source] serde_norway::Error),
}

/// The command parsed from the process arguments.
enum Command {
    /// Run the service until interrupted.
    Run,
    /// Generate the `AsyncAPI` document.
    AsyncApiGen {
        /// Output file, or stdout when `None`.
        out: Option<String>,
        /// Emit YAML instead of JSON.
        yaml: bool,
    },
}

/// The Tokio runtime the generated `main` runs the service on: the `flavor` and `worker_threads`
/// arguments of [`#[ruststream::app]`](macro@crate::app), which mirror `#[tokio::main]`'s.
///
/// A service whose CPU-bound subscriptions run on dedicated threads (`threads(n)`) sizes the app's
/// runtime for its I/O alone: the dedicated threads and the runtime's workers share the machine's
/// cores. A current-thread runtime next to dedicated threads gives the rest of the service the
/// lowest latency.
///
/// # Examples
///
/// ```no_run
/// # #[cfg(feature = "memory")]
/// # {
/// use ruststream::memory::MemoryBroker;
/// use ruststream::runtime::cli::{AppRuntime, run_main_on};
/// use ruststream::runtime::{App, AppInfo, RustStream};
///
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new())
/// }
///
/// // What `#[ruststream::app(flavor = "current_thread")]` expands to.
/// fn main() -> std::process::ExitCode {
///     run_main_on(AppRuntime::CurrentThread, app)
/// }
/// # }
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum AppRuntime {
    /// A multi-threaded runtime with Tokio's default of one worker per core.
    #[default]
    MultiThread,
    /// A multi-threaded runtime with this many workers (`worker_threads = n`).
    MultiThreadWorkers(NonZeroUsize),
    /// A runtime on the thread that calls `main` (`flavor = "current_thread"`).
    CurrentThread,
}

impl AppRuntime {
    /// Builds the runtime, with every driver the framework uses enabled.
    fn build(self) -> io::Result<Runtime> {
        let mut builder = match self {
            Self::MultiThread => Builder::new_multi_thread(),
            Self::MultiThreadWorkers(count) => {
                let mut builder = Builder::new_multi_thread();
                builder.worker_threads(count.get());
                builder
            }
            Self::CurrentThread => Builder::new_current_thread(),
        };
        builder.enable_all().build()
    }
}

/// Runs the generated CLI for `build`'s service, returning a process exit code.
///
/// Called by the [`#[ruststream::app]`](macro@crate::app) expansion; you rarely call it directly.
/// It reads the process arguments, dispatches to `run` or `asyncapi gen`, and prints any error to
/// stderr.
///
/// # Examples
///
/// ```no_run
/// # #[cfg(feature = "memory")]
/// # {
/// use ruststream::memory::MemoryBroker;
/// use ruststream::runtime::{App, AppInfo, RustStream};
///
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new())
/// }
///
/// fn main() -> std::process::ExitCode {
///     ruststream::runtime::cli::run_main(app)
/// }
/// # }
/// ```
#[must_use]
pub fn run_main<A, F>(build: F) -> ExitCode
where
    A: App,
    F: FnOnce() -> A,
{
    run_main_on(AppRuntime::default(), build)
}

/// [`run_main`] on the runtime `runtime` names: what `#[ruststream::app]` expands to when it
/// carries `flavor` or `worker_threads`.
///
/// # Examples
///
/// ```no_run
/// # #[cfg(feature = "memory")]
/// # {
/// use ruststream::memory::MemoryBroker;
/// use ruststream::nonzero;
/// use ruststream::runtime::cli::{AppRuntime, run_main_on};
/// use ruststream::runtime::{App, AppInfo, RustStream};
///
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new())
/// }
///
/// fn main() -> std::process::ExitCode {
///     run_main_on(AppRuntime::MultiThreadWorkers(nonzero!(2)), app)
/// }
/// # }
/// ```
#[must_use]
pub fn run_main_on<A, F>(runtime: AppRuntime, build: F) -> ExitCode
where
    A: App,
    F: FnOnce() -> A,
{
    let args: Vec<String> = std::env::args().skip(1).collect();
    report(execute(&args, runtime, build))
}

/// Maps a dispatch outcome onto the process exit code, printing the error to stderr. Split from
/// [`run_main`] so the mapping is reachable without the process arguments.
fn report(outcome: Result<(), CliError>) -> ExitCode {
    match outcome {
        Ok(()) => ExitCode::SUCCESS,
        Err(err) => {
            eprintln!("ruststream: {err}");
            ExitCode::FAILURE
        }
    }
}

fn execute<A, F>(args: &[String], runtime: AppRuntime, build: F) -> Result<(), CliError>
where
    A: App,
    F: FnOnce() -> A,
{
    match parse(args)? {
        Command::Run => {
            // Install the colored console logger so a freshly scaffolded service prints logs out
            // of the box. Ignore the error: it only fails if the app already installed its own
            // subscriber, which we must not replace.
            #[cfg(feature = "logging")]
            let _ = crate::logging::init();
            let app = build();
            let runtime = runtime.build().map_err(CliError::Runtime)?;
            runtime.block_on(app.run())?;
            Ok(())
        }
        Command::AsyncApiGen { out, yaml } => generate_spec(&build(), out.as_deref(), yaml),
    }
}

fn parse(args: &[String]) -> Result<Command, CliError> {
    match args.first().map(String::as_str) {
        None | Some("run") => Ok(Command::Run),
        Some("asyncapi") => parse_asyncapi(&args[1..]),
        Some(other) => Err(CliError::UnknownCommand(other.to_owned())),
    }
}

fn parse_asyncapi(args: &[String]) -> Result<Command, CliError> {
    if args.first().map(String::as_str) != Some("gen") {
        let found = args.first().cloned().unwrap_or_default();
        return Err(CliError::UnknownAsyncApi(found));
    }
    let mut out = None;
    let mut yaml = false;
    let mut rest = args[1..].iter();
    while let Some(arg) = rest.next() {
        match arg.as_str() {
            "-o" | "--out" => out = rest.next().cloned(),
            "--yaml" => yaml = true,
            other => return Err(CliError::UnknownOption(other.to_owned())),
        }
    }
    Ok(Command::AsyncApiGen { out, yaml })
}

#[cfg(feature = "asyncapi")]
fn generate_spec<A: App>(app: &A, out: Option<&str>, yaml: bool) -> Result<(), CliError> {
    let spec = crate::asyncapi::build_spec(app);
    let text = if yaml {
        spec.to_yaml().map_err(CliError::SerializeYaml)?
    } else {
        spec.to_json()?
    };
    if let Some(path) = out {
        std::fs::write(path, text).map_err(|source| CliError::WriteSpec {
            path: path.to_owned(),
            source,
        })
    } else {
        println!("{text}");
        Ok(())
    }
}

#[cfg(not(feature = "asyncapi"))]
fn generate_spec<A: App>(_app: &A, _out: Option<&str>, _yaml: bool) -> Result<(), CliError> {
    Err(CliError::AsyncApiDisabled)
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "memory")]
    use std::io;
    use std::process::ExitCode;

    #[cfg(feature = "memory")]
    use crate::memory::MemoryBroker;
    #[cfg(feature = "memory")]
    use crate::runtime::{App, AppInfo, RustStream, RustStreamError};

    // `execute` drives a built service, so only the tests that can build one reach it.
    #[cfg(feature = "memory")]
    use super::execute;
    use super::{CliError, Command, parse, report};

    /// `execute` on the runtime `run_main` builds.
    #[cfg(feature = "memory")]
    fn execute_default<A, F>(args: &[String], build: F) -> Result<(), CliError>
    where
        A: App,
        F: FnOnce() -> A,
    {
        execute(args, super::AppRuntime::default(), build)
    }

    fn args(parts: &[&str]) -> Vec<String> {
        parts.iter().map(|s| (*s).to_owned()).collect()
    }

    /// `ExitCode` is opaque (no `PartialEq`), so compare the rendered form.
    fn code_of(outcome: Result<(), CliError>) -> String {
        format!("{:?}", report(outcome))
    }

    #[test]
    fn empty_args_default_to_run() {
        assert!(matches!(parse(&args(&[])).unwrap(), Command::Run));
        assert!(matches!(parse(&args(&["run"])).unwrap(), Command::Run));
    }

    #[test]
    fn asyncapi_gen_parses_output_and_format() {
        let cmd = parse(&args(&["asyncapi", "gen", "-o", "spec.yaml", "--yaml"])).unwrap();
        let Command::AsyncApiGen { out, yaml } = cmd else {
            panic!("expected asyncapi gen");
        };
        assert_eq!(out.as_deref(), Some("spec.yaml"));
        assert!(yaml);
    }

    #[test]
    fn asyncapi_gen_defaults_to_json_stdout() {
        let cmd = parse(&args(&["asyncapi", "gen"])).unwrap();
        let Command::AsyncApiGen { out, yaml } = cmd else {
            panic!("expected asyncapi gen");
        };
        assert!(out.is_none());
        assert!(!yaml);
    }

    #[test]
    fn unknown_command_is_rejected() {
        assert!(parse(&args(&["frobnicate"])).is_err());
        assert!(parse(&args(&["asyncapi", "lint"])).is_err());
        assert!(parse(&args(&["asyncapi", "gen", "--nope"])).is_err());
    }

    #[test]
    fn the_exit_code_follows_the_dispatch_outcome() {
        assert_eq!(code_of(Ok(())), format!("{:?}", ExitCode::SUCCESS));
        assert_eq!(
            code_of(Err(CliError::UnknownCommand("frobnicate".to_owned()))),
            format!("{:?}", ExitCode::FAILURE),
        );
    }

    /// The service the dispatch tests generate a spec for: one broker, no handlers.
    #[cfg(feature = "memory")]
    fn demo_app() -> impl App {
        RustStream::new(AppInfo::new("cli-demo", "0.1.0")).register_broker(MemoryBroker::new())
    }

    #[cfg(feature = "memory")]
    #[test]
    fn an_unknown_command_surfaces_before_the_service_runs() {
        let outcome = execute_default(&args(&["frobnicate"]), demo_app);
        let err = outcome.expect_err("an unrecognized command must not run the service");
        assert!(matches!(err, CliError::UnknownCommand(cmd) if cmd == "frobnicate"));
        assert_eq!(
            code_of(execute_default(&args(&["asyncapi", "lint"]), demo_app)),
            format!("{:?}", ExitCode::FAILURE),
        );
    }

    /// The `run` arm builds its own runtime and drives the service to completion, so a plain
    /// `#[test]` (no ambient runtime) is required. A failing startup is what makes the arm
    /// observable without a termination signal.
    #[cfg(feature = "memory")]
    #[test]
    fn run_surfaces_a_startup_failure() {
        let build = || {
            RustStream::new(AppInfo::new("cli-broken", "0.1.0"))
                .on_startup(async move |()| {
                    Err::<(), _>(io::Error::other("state producer refused"))
                })
                .with_broker(MemoryBroker::new(), |_b| {})
        };
        let err = execute_default(&args(&["run"]), build).expect_err("the failing state producer");
        assert!(
            matches!(err, CliError::Run(RustStreamError::Startup(_))),
            "got: {err:?}",
        );
    }

    #[cfg(all(feature = "memory", feature = "asyncapi"))]
    #[test]
    fn asyncapi_gen_writes_json_and_yaml_to_a_file() {
        let dir = std::env::temp_dir();
        let json_path = dir.join(format!("ruststream-cli-{}-spec.json", std::process::id()));
        let json_out = json_path.to_str().expect("utf-8 temp path").to_owned();
        execute_default(
            &args(&["asyncapi", "gen", "-o", json_out.as_str()]),
            demo_app,
        )
        .expect("the spec must be written");
        let text = std::fs::read_to_string(&json_path).expect("the spec file exists");
        let spec: serde_json::Value = serde_json::from_str(&text).expect("valid JSON");
        assert_eq!(spec["info"]["title"], "cli-demo");
        std::fs::remove_file(&json_path).expect("cleanup");

        let yaml_path = dir.join(format!("ruststream-cli-{}-spec.yaml", std::process::id()));
        let yaml_out = yaml_path.to_str().expect("utf-8 temp path").to_owned();
        execute_default(
            &args(&["asyncapi", "gen", "-o", yaml_out.as_str(), "--yaml"]),
            demo_app,
        )
        .expect("the spec must be written");
        let text = std::fs::read_to_string(&yaml_path).expect("the spec file exists");
        let spec: serde_norway::Value = serde_norway::from_str(&text).expect("valid YAML");
        assert_eq!(spec["info"]["title"].as_str(), Some("cli-demo"));
        std::fs::remove_file(&yaml_path).expect("cleanup");
    }

    #[cfg(all(feature = "memory", feature = "asyncapi"))]
    #[test]
    fn asyncapi_gen_defaults_to_stdout() {
        // Nothing to read back (the document goes to stdout), so the assertion is that the
        // stdout arm reports success rather than falling through to a write error.
        execute_default(&args(&["asyncapi", "gen"]), demo_app).expect("the spec must reach stdout");
    }

    #[cfg(all(feature = "memory", feature = "asyncapi"))]
    #[test]
    fn asyncapi_gen_reports_the_path_it_could_not_write() {
        let missing = std::env::temp_dir()
            .join("ruststream-cli-absent-directory")
            .join("spec.json");
        let out = missing.to_str().expect("utf-8 temp path").to_owned();
        let err = execute_default(&args(&["asyncapi", "gen", "-o", out.as_str()]), demo_app)
            .expect_err("a spec cannot be written into a missing directory");
        let CliError::WriteSpec { path, .. } = err else {
            panic!("expected a write error, got: {err:?}");
        };
        assert_eq!(path, out, "the error names the output path");
    }

    #[cfg(all(feature = "memory", not(feature = "asyncapi")))]
    #[test]
    fn asyncapi_gen_without_the_feature_is_rejected() {
        let err = execute_default(&args(&["asyncapi", "gen"]), demo_app)
            .expect_err("the spec needs the asyncapi feature");
        assert!(matches!(err, CliError::AsyncApiDisabled), "got: {err:?}");
    }

    /// The runtime the generated `main` builds, per `#[ruststream::app]` argument.
    mod runtimes {
        use std::sync::{Arc, Barrier, mpsc};
        use std::time::Duration;

        use tokio::runtime::RuntimeFlavor;

        use super::super::AppRuntime;
        use crate::nonzero;

        #[test]
        fn the_default_is_multi_threaded() {
            let runtime = AppRuntime::default().build().expect("the runtime builds");
            assert_eq!(
                runtime.handle().runtime_flavor(),
                RuntimeFlavor::MultiThread
            );
        }

        #[test]
        fn a_current_thread_flavor_builds_a_current_thread_runtime() {
            let runtime = AppRuntime::CurrentThread
                .build()
                .expect("the runtime builds");
            assert_eq!(
                runtime.handle().runtime_flavor(),
                RuntimeFlavor::CurrentThread
            );
        }

        /// Three tasks that each hold a worker until all three arrive pass only with three
        /// workers.
        #[test]
        fn a_worker_count_starts_that_many_workers() {
            let runtime = AppRuntime::MultiThreadWorkers(nonzero!(3))
                .build()
                .expect("the runtime builds");
            assert_eq!(
                runtime.handle().runtime_flavor(),
                RuntimeFlavor::MultiThread
            );
            let barrier = Arc::new(Barrier::new(3));
            let (passed, arrivals) = mpsc::channel();
            for _ in 0..3 {
                let (barrier, passed) = (Arc::clone(&barrier), passed.clone());
                runtime.spawn(async move {
                    barrier.wait();
                    let _ = passed.send(());
                });
            }
            let arrived = (0..3).all(|_| arrivals.recv_timeout(Duration::from_secs(5)).is_ok());
            // Workers stuck on the barrier would hold a dropped runtime forever.
            runtime.shutdown_background();
            assert!(arrived, "three workers hold the barrier at once");
        }
    }
}