dist: move Notifier into DownloadCfg

Author: djc

src/cli/common.rs

@@ -4,7 +4,7 @@ use std::fmt::Display;
 use std::fs;
 use std::io::{BufRead, Write};
 use std::path::{Path, PathBuf};
-use std::sync::{Arc, LazyLock};
+use std::sync::LazyLock;
 use std::{cmp, env};
 
 use anyhow::{Context, Result, anyhow};
@@ -13,7 +13,6 @@ use termcolor::Color;
 use tracing::{error, info, warn};
 use tracing_subscriber::{EnvFilter, Registry, reload::Handle};
 
-use crate::dist::download::Notifier;
 use crate::{
     config::Cfg,
     dist::{TargetTriple, ToolchainDesc},
@@ -122,8 +121,7 @@ pub(crate) fn read_line(process: &Process) -> Result<String> {
 
 #[tracing::instrument(level = "trace", skip(process))]
 pub(crate) fn set_globals(current_dir: PathBuf, quiet: bool, process: &Process) -> Result<Cfg<'_>> {
-    let notifier = Notifier::new(quiet, process);
-    Cfg::from_env(current_dir, Arc::new(move |n| notifier.handle(n)), process)
+    Cfg::from_env(current_dir, quiet, process)
 }
 
 pub(crate) fn show_channel_update(

src/cli/rustup_mode.rs

@@ -33,8 +33,7 @@ use crate::{
     command, component_for_bin,
     config::{ActiveReason, Cfg},
     dist::{
-        AutoInstallMode, PartialToolchainDesc, Profile, TargetTriple,
-        manifest::{Component, ComponentStatus},
+        AutoInstallMode, PartialToolchainDesc, Profile, TargetTriple, download::DownloadCfg, manifest::{Component, ComponentStatus}
     },
     errors::RustupError,
     install::{InstallMethod, UpdateStatus},
@@ -909,7 +908,7 @@ async fn check_updates(cfg: &Cfg<'_>, opts: CheckOpts) -> Result<utils::ExitCode
         && self_update_mode == SelfUpdateMode::Enable
         && !opts.no_self_update;
 
-    if self_update && check_rustup_update(cfg.process).await? {
+    if self_update && check_rustup_update(&DownloadCfg::new(cfg)).await? {
         update_available = true;
     }
 
@@ -937,10 +936,13 @@ async fn update(
     if let Some(p) = opts.profile {
         cfg.set_profile_override(p);
     }
+
     let cfg = &cfg;
     if cfg.get_profile()? == Profile::Complete {
         warn!("{}", common::WARN_COMPLETE_PROFILE);
     }
+
+    let dl_cfg = DownloadCfg::new(cfg);
     let names = opts.toolchain;
     if !names.is_empty() {
         for name in names {
@@ -994,7 +996,7 @@ async fn update(
             }
         }
         if self_update {
-            exit_code &= self_update::self_update(cfg.process).await?;
+            exit_code &= self_update::self_update(&dl_cfg).await?;
         }
     } else if ensure_active_toolchain {
         let (toolchain, reason) = cfg.ensure_active_toolchain(force_non_host, true).await?;
@@ -1003,7 +1005,7 @@ async fn update(
     } else {
         exit_code &= common::update_all_channels(cfg, opts.force).await?;
         if self_update {
-            exit_code &= self_update::self_update(cfg.process).await?;
+            exit_code &= self_update::self_update(&dl_cfg).await?;
         }
 
         info!("cleaning up downloads & tmp directories");
@@ -1012,7 +1014,7 @@ async fn update(
     }
 
     if !cfg!(feature = "no-self-update") && self_update_mode == SelfUpdateMode::CheckOnly {
-        check_rustup_update(cfg.process).await?;
+        check_rustup_update(&dl_cfg).await?;
     }
 
     if cfg!(feature = "no-self-update") {

src/cli/self_update.rs

@@ -50,6 +50,7 @@ use serde::{Deserialize, Serialize};
 use termcolor::Color;
 use tracing::{error, info, trace, warn};
 
+use crate::dist::download::DownloadCfg;
 use crate::{
     DUP_TOOLS, TOOLS,
     cli::{
@@ -1112,7 +1113,7 @@ pub(crate) fn self_update_permitted(explicit: bool) -> Result<SelfUpdatePermissi
 }
 
 /// Performs all of a self-update: check policy, download, apply and exit.
-pub(crate) async fn self_update(process: &Process) -> Result<utils::ExitCode> {
+pub(crate) async fn self_update(dl_cfg: &DownloadCfg<'_>) -> Result<utils::ExitCode> {
     match self_update_permitted(false)? {
         SelfUpdatePermission::HardFail => {
             error!("Unable to self-update.  STOP");
@@ -1123,13 +1124,13 @@ pub(crate) async fn self_update(process: &Process) -> Result<utils::ExitCode> {
         SelfUpdatePermission::Permit => {}
     }
 
-    let setup_path = prepare_update(process).await?;
+    let setup_path = prepare_update(dl_cfg).await?;
 
     if let Some(setup_path) = &setup_path {
         return run_update(setup_path);
     } else {
         // Try again in case we emitted "tool `{}` is already installed" last time.
-        install_proxies(process)?;
+        install_proxies(dl_cfg.process)?;
     }
 
     Ok(utils::ExitCode(0))
@@ -1174,7 +1175,7 @@ pub(crate) async fn update(cfg: &Cfg<'_>) -> Result<utils::ExitCode> {
         Permit => {}
     }
 
-    match prepare_update(cfg.process).await? {
+    match prepare_update(&DownloadCfg::new(cfg)).await? {
         Some(setup_path) => {
             let Some(version) = get_and_parse_new_rustup_version(&setup_path) else {
                 error!("failed to get rustup version");
@@ -1227,8 +1228,8 @@ fn parse_new_rustup_version(version: String) -> String {
     String::from(matched_version)
 }
 
-pub(crate) async fn prepare_update(process: &Process) -> Result<Option<PathBuf>> {
-    let cargo_home = process.cargo_home()?;
+pub(crate) async fn prepare_update(dl_cfg: &DownloadCfg<'_>) -> Result<Option<PathBuf>> {
+    let cargo_home = dl_cfg.process.cargo_home()?;
     let rustup_path = cargo_home.join(format!("bin{MAIN_SEPARATOR}rustup{EXE_SUFFIX}"));
     let setup_path = cargo_home.join(format!("bin{MAIN_SEPARATOR}rustup-init{EXE_SUFFIX}"));
 
@@ -1253,19 +1254,19 @@ pub(crate) async fn prepare_update(process: &Process) -> Result<Option<PathBuf>>
     let triple = dist::TargetTriple::from_host(process).unwrap_or(triple);
 
     // Get update root.
-    let update_root = update_root(process);
+    let update_root = update_root(dl_cfg.process);
 
     // Get current version
     let current_version = env!("CARGO_PKG_VERSION");
 
     // Get available version
     info!("checking for self-update (current version: {current_version})");
-    let available_version = match process.var_opt("RUSTUP_VERSION")? {
+    let available_version = match dl_cfg.process.var_opt("RUSTUP_VERSION")? {
         Some(ver) => {
             info!("`RUSTUP_VERSION` has been set to `{ver}`");
             ver
         }
-        None => get_available_rustup_version(process).await?,
+        None => get_available_rustup_version(dl_cfg).await?,
     };
 
     // If up-to-date
@@ -1281,16 +1282,23 @@ pub(crate) async fn prepare_update(process: &Process) -> Result<Option<PathBuf>>
 
     // Download new version
     info!("downloading self-update (new version: {available_version})");
-    download_file(&download_url, &setup_path, None, &|_| (), process).await?;
+    download_file(
+        &download_url,
+        &setup_path,
+        None,
+        &dl_cfg.notifier,
+        dl_cfg.process,
+    )
+    .await?;
 
     // Mark as executable
     utils::make_executable(&setup_path)?;
 
     Ok(Some(setup_path))
 }
 
-async fn get_available_rustup_version(process: &Process) -> Result<String> {
-    let update_root = update_root(process);
+async fn get_available_rustup_version(dl_cfg: &DownloadCfg<'_>) -> Result<String> {
+    let update_root = update_root(dl_cfg.process);
     let tempdir = tempfile::Builder::new()
         .prefix("rustup-update")
         .tempdir()
@@ -1300,7 +1308,14 @@ async fn get_available_rustup_version(process: &Process) -> Result<String> {
     let release_file_url = format!("{update_root}/release-stable.toml");
     let release_file_url = utils::parse_url(&release_file_url)?;
     let release_file = tempdir.path().join("release-stable.toml");
-    download_file(&release_file_url, &release_file, None, &|_| (), process).await?;
+    download_file(
+        &release_file_url,
+        &release_file,
+        None,
+        &dl_cfg.notifier,
+        dl_cfg.process,
+    )
+    .await?;
     let release_toml_str = utils::read_file("rustup release", &release_file)?;
     let release_toml = toml::from_str::<RustupManifest>(&release_toml_str)
         .context("unable to parse rustup release file")?;
@@ -1348,13 +1363,13 @@ impl fmt::Display for SchemaVersion {
 }
 
 /// Returns whether an update was available
-pub(crate) async fn check_rustup_update(process: &Process) -> anyhow::Result<bool> {
-    let mut t = process.stdout();
+pub(crate) async fn check_rustup_update(dl_cfg: &DownloadCfg<'_>) -> anyhow::Result<bool> {
+    let mut t = dl_cfg.process.stdout();
     // Get current rustup version
     let current_version = env!("CARGO_PKG_VERSION");
 
     // Get available rustup version
-    let available_version = get_available_rustup_version(process).await?;
+    let available_version = get_available_rustup_version(dl_cfg).await?;
 
     let _ = t.attr(Attr::Bold);
     write!(t.lock(), "rustup - ")?;

src/config.rs

@@ -2,7 +2,6 @@ use std::fmt::{self, Debug, Display};
 use std::io;
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
-use std::sync::Arc;
 
 use anyhow::{Context, Result, anyhow, bail};
 use serde::Deserialize;
@@ -17,7 +16,6 @@ use crate::{
     errors::RustupError,
     fallback_settings::FallbackSettings,
     install::UpdateStatus,
-    notifications::*,
     process::Process,
     settings::{MetadataVersion, Settings, SettingsFile},
     toolchain::{
@@ -230,15 +228,15 @@ pub(crate) struct Cfg<'a> {
     pub toolchain_override: Option<ResolvableToolchainName>,
     pub env_override: Option<LocalToolchainName>,
     pub dist_root_url: String,
-    pub notify_handler: Arc<dyn Fn(Notification<'_>)>,
+    pub quiet: bool,
     pub current_dir: PathBuf,
     pub process: &'a Process,
 }
 
 impl<'a> Cfg<'a> {
     pub(crate) fn from_env(
         current_dir: PathBuf,
-        notify_handler: Arc<dyn Fn(Notification<'_>)>,
+        quiet: bool,
         process: &'a Process,
     ) -> Result<Self> {
         // Set up the rustup home directory
@@ -299,10 +297,10 @@ impl<'a> Cfg<'a> {
             update_hash_dir,
             download_dir,
             tmp_cx,
-            notify_handler,
             toolchain_override: None,
             env_override,
             dist_root_url: dist_root,
+            quiet,
             current_dir,
             process,
         };
@@ -997,7 +995,7 @@ impl Debug for Cfg<'_> {
             toolchain_override,
             env_override,
             dist_root_url,
-            notify_handler: _,
+            quiet,
             current_dir,
             process: _,
         } = self;
@@ -1014,6 +1012,7 @@ impl Debug for Cfg<'_> {
             .field("toolchain_override", toolchain_override)
             .field("env_override", env_override)
             .field("dist_root_url", dist_root_url)
+            .field("quiet", quiet)
             .field("current_dir", current_dir)
             .finish()
     }

src/diskio/mod.rs

@@ -55,6 +55,7 @@ pub(crate) mod immediate;
 #[cfg(test)]
 mod test;
 pub(crate) mod threaded;
+use threaded::PoolReference;
 
 use std::io::{self, Write};
 use std::ops::{Deref, DerefMut};
@@ -65,9 +66,8 @@ use std::{fmt::Debug, fs::OpenOptions};
 
 use anyhow::Result;
 
-use crate::notifications::Notification;
+use crate::dist::download::Notifier;
 use crate::process::Process;
-use threaded::PoolReference;
 
 /// Carries the implementation specific data for complete file transfers into the executor.
 #[derive(Debug)]
@@ -443,13 +443,13 @@ pub(crate) fn create_dir<P: AsRef<Path>>(path: P) -> io::Result<()> {
 
 /// Get the executor for disk IO.
 pub(crate) fn get_executor<'a>(
-    notify_handler: Option<&'a dyn Fn(Notification<'_>)>,
+    notifier: Option<&'a Notifier>,
     ram_budget: usize,
     process: &Process,
 ) -> anyhow::Result<Box<dyn Executor + 'a>> {
     // If this gets lots of use, consider exposing via the config file.
     Ok(match process.io_thread_count()? {
         0 | 1 => Box::new(immediate::ImmediateUnpacker::new()),
-        n => Box::new(threaded::Threaded::new(notify_handler, n, ram_budget)),
+        n => Box::new(threaded::Threaded::new(notifier, n, ram_budget)),
     })
 }

src/diskio/threaded.rs

@@ -15,6 +15,7 @@ use sharded_slab::pool::{OwnedRef, OwnedRefMut};
 use tracing::debug;
 
 use super::{CompletedIo, Executor, Item, perform};
+use crate::dist::download::Notifier;
 use crate::notifications::Notification;
 
 #[derive(Copy, Clone, Debug, Enum)]
@@ -99,7 +100,7 @@ impl fmt::Debug for Pool {
 pub(crate) struct Threaded<'a> {
     n_files: Arc<AtomicUsize>,
     pool: threadpool::ThreadPool,
-    notify_handler: Option<&'a dyn Fn(Notification<'_>)>,
+    notifier: Option<&'a Notifier>,
     rx: Receiver<Task>,
     tx: Sender<Task>,
     vec_pools: EnumMap<Bucket, Pool>,
@@ -109,7 +110,7 @@ pub(crate) struct Threaded<'a> {
 impl<'a> Threaded<'a> {
     /// Construct a new Threaded executor.
     pub(crate) fn new(
-        notify_handler: Option<&'a dyn Fn(Notification<'_>)>,
+        notify_handler: Option<&'a Notifier>,
         thread_count: usize,
         ram_budget: usize,
     ) -> Self {
@@ -168,7 +169,7 @@ impl<'a> Threaded<'a> {
         Self {
             n_files: Arc::new(AtomicUsize::new(0)),
             pool,
-            notify_handler,
+            notifier: notify_handler,
             rx,
             tx,
             vec_pools,
@@ -261,9 +262,9 @@ impl Executor for Threaded<'_> {
         // actual handling of data today, we synthesis a data buffer and
         // pretend to have bytes to deliver.
         let mut prev_files = self.n_files.load(Ordering::Relaxed);
-        if let Some(handler) = self.notify_handler {
-            handler(Notification::DownloadFinished(None));
-            handler(Notification::DownloadContentLengthReceived(
+        if let Some(notifier) = self.notifier {
+            notifier.handle(Notification::DownloadFinished(None));
+            notifier.handle(Notification::DownloadContentLengthReceived(
                 prev_files as u64,
                 None,
             ));
@@ -282,16 +283,16 @@ impl Executor for Threaded<'_> {
             prev_files = current_files;
             current_files = self.n_files.load(Ordering::Relaxed);
             let step_count = prev_files - current_files;
-            if let Some(handler) = self.notify_handler {
-                handler(Notification::DownloadDataReceived(
+            if let Some(notifier) = self.notifier {
+                notifier.handle(Notification::DownloadDataReceived(
                     &buf[0..step_count],
                     None,
                 ));
             }
         }
         self.pool.join();
-        if let Some(handler) = self.notify_handler {
-            handler(Notification::DownloadFinished(None));
+        if let Some(notifier) = self.notifier {
+            notifier.handle(Notification::DownloadFinished(None));
         }
         // close the feedback channel so that blocking reads on it can
         // complete. send is atomic, and we know the threads completed from the