From 6240e304f74a7572e9d73243632248a0f0e7388b Mon Sep 17 00:00:00 2001 From: Cory Forsstrom Date: Tue, 27 Jan 2026 07:50:27 -0800 Subject: [PATCH] Add example of an output / emit API --- Cargo.lock | 10 ++++ Cargo.toml | 1 + moss/Cargo.toml | 1 + moss/src/cli/mod.rs | 21 +++++++- moss/src/lib.rs | 1 + moss/src/output.rs | 89 +++++++++++++++++++++++++++++++ moss/src/output/tracing.rs | 51 ++++++++++++++++++ moss/src/output/tui.rs | 95 ++++++++++++++++++++++++++++++++++ moss/src/repository/manager.rs | 79 ++++++++++++++++------------ 9 files changed, 314 insertions(+), 34 deletions(-) create mode 100644 moss/src/output.rs create mode 100644 moss/src/output/tracing.rs create mode 100644 moss/src/output/tui.rs diff --git a/Cargo.lock b/Cargo.lock index 31d03ef88..1589007eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -112,6 +112,15 @@ dependencies = [ "windows-sys 0.60.2", ] +[[package]] +name = "arc-swap" +version = "1.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51d03449bb8ca2cc2ef70869af31463d1ae5ccc8fa3e334b307203fbf815207e" +dependencies = [ + "rustversion", +] + [[package]] name = "arrayref" version = "0.3.9" @@ -1892,6 +1901,7 @@ dependencies = [ name = "moss" version = "0.26.1" dependencies = [ + "arc-swap", "astr", "blsforme", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 79fb4309c..9907cd8e6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,6 +97,7 @@ tempfile = "3.20.0" kdl = "6.5.0" libc = "0.2.62" cbindgen = "0.29.2" +arc-swap = "1.8.0" [workspace.lints.rust] rust_2018_idioms = { level = "warn", priority = -1 } diff --git a/moss/Cargo.toml b/moss/Cargo.toml index 0104cf9d5..e08ebd429 100644 --- a/moss/Cargo.toml +++ b/moss/Cargo.toml @@ -15,6 +15,7 @@ triggers = { path = "../crates/triggers" } tui = { path = "../crates/tui" } vfs = { path = "../crates/vfs" } +arc-swap.workspace = true blsforme.workspace = true bytes.workspace = true camino.workspace = true diff --git a/moss/src/cli/mod.rs b/moss/src/cli/mod.rs index 03706474c..df0286cdd 100644 --- a/moss/src/cli/mod.rs +++ b/moss/src/cli/mod.rs @@ -10,7 +10,10 @@ use clap_complete::{ shells::{Bash, Fish, Zsh}, }; use clap_mangen::Man; -use moss::{Installation, installation}; +use moss::{ + Installation, installation, + output::{self, DefaultOutput, TracingOutput}, +}; use thiserror::Error; use tracing_common::{self, logging::LogConfig, logging::init_log_with_config}; use tui::Styled; @@ -101,6 +104,14 @@ fn command() -> Command { .value_name("DIR") .hide(true), ) + .arg( + Arg::new("silent") + .short('s') + .long("silent") + .global(true) + .help("Suppress all output") + .action(ArgAction::SetTrue), + ) .arg_required_else_help(true) .subcommand(boot::command()) .subcommand(cache::command()) @@ -159,11 +170,19 @@ pub fn process() -> Result<(), Error> { let matches = command().get_matches_from(args); let show_version = matches.get_one::("version").is_some_and(|v| *v); + let silent = matches.get_one::("silent").is_some_and(|v| *v); if show_version { println!("moss {}", tools_buildinfo::get_full_version()); } + if silent { + // We still output logging, that's controlled w/ a separate flag + output::install_emitter(TracingOutput::default()); + } else { + output::install_emitter(DefaultOutput::default()); + } + if let Some(log_config) = matches.get_one::("log") { init_log_with_config(log_config.clone()); } diff --git a/moss/src/lib.rs b/moss/src/lib.rs index 754045773..9b954ae93 100644 --- a/moss/src/lib.rs +++ b/moss/src/lib.rs @@ -17,6 +17,7 @@ pub mod db; pub mod dependency; pub mod environment; pub mod installation; +pub mod output; pub mod package; pub mod registry; pub mod repository; diff --git a/moss/src/output.rs b/moss/src/output.rs new file mode 100644 index 000000000..78b15ca2d --- /dev/null +++ b/moss/src/output.rs @@ -0,0 +1,89 @@ +// SPDX-FileCopyrightText: Copyright © 2020-2026 Serpent OS Developers +// +// SPDX-License-Identifier: MPL-2.0 + +use std::sync::{Arc, OnceLock}; + +use crate::repository; + +pub use self::tracing::TracingOutput; +pub use self::tui::TuiOutput; + +mod tracing; +mod tui; + +/// Default emitter used if [`install_emitter`] isn't called with a +/// custom [`Emit`] implementation. +pub type DefaultOutput = Chain; + +/// Global emitter either defaulted or installed via [`install_emitter`] +static EMITTER: OnceLock> = OnceLock::new(); + +/// Install a global emitter that can be used with [`emit`]. If not called, +/// [`DefaultOutput`] is used. +/// +/// This can only be called once. Future calls have no effect. +pub fn install_emitter(emitter: impl Emitter + 'static) { + let _ = EMITTER.set(Arc::new(emitter)); +} + +/// Get access to the global emitter +pub fn emitter() -> &'static dyn Emitter { + EMITTER.get_or_init(|| Arc::new(DefaultOutput::default())).as_ref() +} + +/// Emit an event for output +#[macro_export] +macro_rules! emit { + ($($tt:tt)*) => { + $crate::output::Event::emit(($($tt)*), $crate::output::emitter()); + }; +} + +/// Defines how events are emitted to some output +pub trait Emitter: Send + Sync { + fn emit(&self, _event: &InternalEvent) {} +} + +/// An emittable event +pub trait Event: Sized { + fn emit(self, _emitter: &dyn Emitter) {} +} + +/// An internal `moss` library event +pub enum InternalEvent { + RepositoryManager(repository::manager::OutputEvent), +} + +pub trait EmitExt: Emitter + Sized { + fn chain(self, other: U) -> Chain + where + U: Emitter + Sized, + { + Chain { a: self, b: other } + } +} + +/// Do nothing with / suppress all output +#[derive(Debug, Clone, Copy)] +pub struct NoopOutput; + +impl Emitter for NoopOutput {} + +/// Chains multiple emitters together +#[derive(Clone, Default)] +pub struct Chain { + a: A, + b: B, +} + +impl Emitter for Chain +where + A: Emitter, + B: Emitter, +{ + fn emit(&self, event: &InternalEvent) { + self.a.emit(event); + self.b.emit(event); + } +} diff --git a/moss/src/output/tracing.rs b/moss/src/output/tracing.rs new file mode 100644 index 000000000..0ae34de52 --- /dev/null +++ b/moss/src/output/tracing.rs @@ -0,0 +1,51 @@ +use tracing::info; + +use crate::{output, repository}; + +/// Tracing output +#[derive(Debug, Clone, Default)] +pub struct TracingOutput { + _tracing: TracingState, +} + +impl output::Emitter for TracingOutput { + fn emit(&self, event: &output::InternalEvent) { + match event { + output::InternalEvent::RepositoryManager(event) => match event { + repository::manager::OutputEvent::RefreshStarted { num_to_refresh } => { + info!( + target: "repository_manager", + num_repositories = %num_to_refresh, + "Refreshing repositories" + ); + } + repository::manager::OutputEvent::RefreshRepoStarted(id) => { + info!( + target: "repository_manager", + repo_id = %id, + "Refreshing repository" + ); + } + repository::manager::OutputEvent::RefreshRepoFinished(id) => { + info!( + target: "repository_manager", + repo_id = %id, + "Repository refreshed" + ); + } + repository::manager::OutputEvent::RefreshFinished { elapsed } => { + info!( + target: "repository_manager", + elapsed_seconds = %elapsed.as_secs_f32(), + "All repositories refreshed" + ); + } + }, + } + } +} + +#[derive(Debug, Clone, Default)] +struct TracingState { + // spans: Arc>>, +} diff --git a/moss/src/output/tui.rs b/moss/src/output/tui.rs new file mode 100644 index 000000000..92f7e5181 --- /dev/null +++ b/moss/src/output/tui.rs @@ -0,0 +1,95 @@ +use std::{collections::HashMap, fmt, sync::Arc, time::Duration}; + +use arc_swap::ArcSwap; +use tui::{MultiProgress, ProgressBar, ProgressStyle, Styled}; + +use crate::{output, repository}; + +/// Textual output to stdout / stderr +#[derive(Debug, Clone, Default)] +pub struct TuiOutput { + progress: ProgressState, +} + +impl output::Emitter for TuiOutput { + fn emit(&self, event: &output::InternalEvent) { + match event { + output::InternalEvent::RepositoryManager(event) => self.emit_repository_manager(event), + } + } +} + +impl TuiOutput { + fn emit_repository_manager(&self, event: &repository::manager::OutputEvent) { + match event { + repository::manager::OutputEvent::RefreshStarted { .. } => { + self.progress.multi_start(); + } + repository::manager::OutputEvent::RefreshRepoStarted(id) => { + let id = id.to_string(); + let pb = self.progress.multi_add_pb( + &id, + ProgressBar::new_spinner() + .with_style( + ProgressStyle::with_template(" {spinner} {wide_msg}") + .unwrap() + .tick_chars("--=≡■≡=--"), + ) + .with_message(format!("{} {id}", "Refreshing".blue())), + ); + pb.enable_steady_tick(Duration::from_millis(150)); + } + repository::manager::OutputEvent::RefreshRepoFinished(id) => { + let id = id.to_string(); + self.progress + .multi_pb_println(&id, format_args!("{} {id}", "Refreshed".green())); + self.progress.multi_remove_pb(&id); + } + repository::manager::OutputEvent::RefreshFinished { .. } => { + self.progress.multi_finish(); + } + } + } +} + +#[derive(Debug, Clone, Default)] +struct ProgressState { + // ArcSwap provides lock-free safe usage in sync & async environments + mpb: Arc>, + pbs: Arc>>, +} + +impl ProgressState { + fn multi_start(&self) { + self.mpb.store(Arc::new(MultiProgress::new())); + self.pbs.store(Arc::new(HashMap::new())); + } + + fn multi_add_pb(&self, id: &str, pb: ProgressBar) -> ProgressBar { + let pb = self.mpb.load().add(pb); + self.pbs.rcu(|pbs| { + let mut pbs = (**pbs).clone(); + pbs.insert(id.to_owned(), pb.clone()); + Arc::new(pbs) + }); + pb + } + + fn multi_pb_println(&self, id: &str, args: fmt::Arguments<'_>) { + let pbs = self.pbs.load(); + pbs.get(id).expect("pb exists").suspend(|| println!("{args}")); + } + + fn multi_remove_pb(&self, id: &str) { + self.pbs.rcu(|pbs| { + let mut pbs = (**pbs).clone(); + pbs.remove(id); + Arc::new(pbs) + }); + } + + fn multi_finish(&self) { + self.mpb.store(Arc::new(MultiProgress::new())); + self.pbs.store(Arc::new(HashMap::new())); + } +} diff --git a/moss/src/repository/manager.rs b/moss/src/repository/manager.rs index 137ededa2..c0d176b3b 100644 --- a/moss/src/repository/manager.rs +++ b/moss/src/repository/manager.rs @@ -14,12 +14,9 @@ use stone::{StoneDecodedPayload, StonePayloadMetaTag, StoneReadError}; use thiserror::Error; use xxhash_rust::xxh3::xxh3_64; -use tui::{MultiProgress, ProgressBar, ProgressStyle, Styled}; - use crate::db::meta; use crate::repository::{self, Repository}; -use crate::{Installation, package}; -use crate::{environment, runtime}; +use crate::{Installation, emit, environment, output, package, runtime}; enum Source { System(config::Manager), @@ -42,11 +39,21 @@ pub struct Manager { repositories: BTreeMap, } -impl Manager { - pub fn is_explicit(&self) -> bool { - matches!(self.source, Source::Explicit { .. }) +#[derive(Debug, Clone)] +pub enum OutputEvent { + RefreshStarted { num_to_refresh: usize }, + RefreshRepoStarted(repository::Id), + RefreshRepoFinished(repository::Id), + RefreshFinished { elapsed: Duration }, +} + +impl output::Event for OutputEvent { + fn emit(self, emitter: &dyn output::Emitter) { + emitter.emit(&output::InternalEvent::RepositoryManager(self)); } +} +impl Manager { /// Create a [`Manager`] for the supplied [`Installation`] using system configurations pub fn system(config: config::Manager, installation: Installation) -> Result { Self::new(Source::System(config), installation) @@ -100,6 +107,10 @@ impl Manager { }) } + pub fn is_explicit(&self) -> bool { + matches!(self.source, Source::Explicit { .. }) + } + /// Add a [`Repository`] pub fn add_repository(&mut self, id: repository::Id, repository: Repository) -> Result<(), Error> { let Source::System(config) = &self.source else { @@ -139,32 +150,37 @@ impl Manager { /// Refresh all [`Repository`]'s by fetching it's latest index /// file and updating it's associated meta database pub async fn refresh_all(&mut self) -> Result<(), Error> { - let mpb = MultiProgress::new(); + let active_repos = self + .repositories + .iter() + .filter(|(_, r)| r.repository.active) + .collect::>(); + + let now = tokio::time::Instant::now(); + + emit!(OutputEvent::RefreshStarted { + num_to_refresh: active_repos.len(), + }); // Fetch index files asynchronously and then // update to DB - stream::iter(self.repositories.iter().filter(|(_, r)| r.repository.active)) + let res = stream::iter(&active_repos) .map(|(id, _)| async { - let pb = mpb.add( - ProgressBar::new_spinner() - .with_style( - ProgressStyle::with_template(" {spinner} {wide_msg}") - .unwrap() - .tick_chars("--=≡■≡=--"), - ) - .with_message(format!("{} {}", "Refreshing".blue(), *id)), - ); - pb.enable_steady_tick(Duration::from_millis(150)); + emit!(OutputEvent::RefreshRepoStarted((*id).clone())); self.refresh(id).await?; - pb.suspend(|| println!("{} {}", "Refreshed".green(), *id)); + emit!(OutputEvent::RefreshRepoFinished((*id).clone())); Ok(()) }) .buffer_unordered(environment::MAX_NETWORK_CONCURRENCY) .try_collect() - .await + .await; + + emit!(OutputEvent::RefreshFinished { elapsed: now.elapsed() }); + + res } /// Ensures all repositories are initialized - index file downloaded and meta db @@ -189,26 +205,21 @@ impl Manager { return Ok(0); } - let mpb = MultiProgress::new(); + let now = tokio::time::Instant::now(); + + emit!(OutputEvent::RefreshStarted { + num_to_refresh: uninitialized.len(), + }); // Fetch index files asynchronously and then // update to DB stream::iter(&uninitialized) .map(|id| async { - let pb = mpb.add( - ProgressBar::new_spinner() - .with_style( - ProgressStyle::with_template(" {spinner} {wide_msg}") - .unwrap() - .tick_chars("--=≡■≡=--"), - ) - .with_message(format!("{} {}", "Refreshing".blue(), *id)), - ); - pb.enable_steady_tick(Duration::from_millis(150)); + emit!(OutputEvent::RefreshRepoStarted((*id).clone())); self.refresh(id).await?; - pb.suspend(|| println!("{} {}", "Refreshed".green(), *id)); + emit!(OutputEvent::RefreshRepoFinished((*id).clone())); Ok(()) as Result<_, Error> }) @@ -216,6 +227,8 @@ impl Manager { .try_collect::<()>() .await?; + emit!(OutputEvent::RefreshFinished { elapsed: now.elapsed() }); + Ok(uninitialized.len()) }