diff --git a/packages/core_new/src/api/ingress.rs b/packages/core_new/src/api/ingress.rs index 93b8f335..0200b9b4 100644 --- a/packages/core_new/src/api/ingress.rs +++ b/packages/core_new/src/api/ingress.rs @@ -1,5 +1,6 @@ use crate::{store::{ingress::{IngressId, IngressType}, State}, LauncherResult}; +#[tracing::instrument] pub async fn init_ingress( ingress_type: IngressType, message: &str, @@ -13,25 +14,43 @@ pub async fn init_ingress( Ok(id) } -pub async fn send_ingress(id: IngressId) -> LauncherResult<()> { +#[tracing::instrument] +pub async fn send_ingress(id: &IngressId, increment: f64) -> LauncherResult<()> { let state = State::get().await?; let processor = &state.ingress_processor; - processor.send(id.0).await?; + processor.send(id.0, increment).await?; Ok(()) } -pub async fn update_ingress(id: IngressId, current: f64) -> LauncherResult<()> { - let state = State::get().await?; - let processor = &state.ingress_processor; +#[cfg(test)] +pub mod tests { + use std::time::Duration; - processor.update(id.0, current).await -} + use crate::{api::{ingress::{init_ingress, send_ingress}, proxy::proxy_cli::ProxyCli}, initialize_core, store::ingress::IngressType, LauncherResult}; -pub async fn finish_ingress(id: IngressId) -> LauncherResult<()> { - let state = State::get().await?; - let processor = &state.ingress_processor; - processor.finish(id.0).await?; - Ok(()) + #[tokio::test] + pub async fn create_and_update_ingress() -> LauncherResult<()> { + assert!(cfg!(feature = "cli")); + + initialize_core(ProxyCli::new()).await?; + + let id = init_ingress(IngressType::Download { file_name: "Some-Mod-1.8.9.jar".into() }, "This is a test message", 100.0).await?; + tokio::time::sleep(Duration::from_millis(2350)).await; + + send_ingress(&id, 10.0).await?; + tokio::time::sleep(Duration::from_millis(2500)).await; + + send_ingress(&id, 20.0).await?; + tokio::time::sleep(Duration::from_millis(2500)).await; + + send_ingress(&id, 70.0).await?; + tokio::time::sleep(Duration::from_millis(2500)).await; + + drop(id); + + Ok(()) + } + } \ No newline at end of file diff --git a/packages/core_new/src/api/proxy/proxy_cli.rs b/packages/core_new/src/api/proxy/proxy_cli.rs index b4ee7eb4..28a3e2f7 100644 --- a/packages/core_new/src/api/proxy/proxy_cli.rs +++ b/packages/core_new/src/api/proxy/proxy_cli.rs @@ -4,7 +4,7 @@ use indicatif::ProgressBar; use tokio::sync::RwLock; use uuid::Uuid; -use crate::{store::{ingress::IngressPayload, State}, LauncherResult}; +use crate::{constants::CLI_TOTAL_INGRESS, store::ingress::IngressPayload, LauncherResult}; use super::LauncherProxy; @@ -24,21 +24,33 @@ impl ProxyCli { impl LauncherProxy for ProxyCli { async fn send_ingress(&self, ingress: IngressPayload) -> LauncherResult<()> { - let feeds = self.ingress_feeds.write().await; + let mut feeds = self.ingress_feeds.write().await; + + let completed = ingress.percent.is_none(); if let Some(progress) = feeds.get(&ingress.id) { - // progress.set_position(ingress.percent as u64); - // TODO: customize progress bar - } else { - let progress = ProgressBar::new(ingress.total as u64); + if completed { + progress.finish(); + } else { + #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)] + let new_pos = (ingress.percent.unwrap() * CLI_TOTAL_INGRESS as f64).round() as u64; + progress.set_position(new_pos); + } + } else if !completed { + let progress = ProgressBar::new(CLI_TOTAL_INGRESS); progress.set_message(ingress.message.clone()); progress.set_position(0); + progress.set_style(indicatif::ProgressStyle::default_bar().template( + "{spinner:.green}, [{elapsed_precise}] [{bar:.lime/green}] {pos}/{len} {msg}" + ).unwrap().progress_chars("#>-")); + + feeds.insert(ingress.id, progress); } Ok(()) } async fn send_message(&self, message: crate::api::proxy::message::MessagePayload) -> LauncherResult<()> { - println!("[{:?}] {:?}", message.level, message.message); + println!("[{:?}] {}", message.level, message.message); Ok(()) } diff --git a/packages/core_new/src/lib.rs b/packages/core_new/src/lib.rs index a0f90c92..c8712c58 100644 --- a/packages/core_new/src/lib.rs +++ b/packages/core_new/src/lib.rs @@ -6,5 +6,17 @@ pub mod api; mod error; mod logger; +use api::proxy::LauncherProxy; pub use error::{LauncherResult, LauncherError}; -pub use logger::start_logger; \ No newline at end of file +pub use logger::start_logger; +use store::{proxy::ProxyState, Dirs, State}; + +pub async fn initialize_core(proxy_backend: impl LauncherProxy + 'static) -> LauncherResult<()> { + Dirs::get().await?; + start_logger().await; + + ProxyState::initialize(proxy_backend).await?; + State::get().await?; + + Ok(()) +} \ No newline at end of file diff --git a/packages/core_new/src/logger.rs b/packages/core_new/src/logger.rs index 6103b6b4..8a626188 100644 --- a/packages/core_new/src/logger.rs +++ b/packages/core_new/src/logger.rs @@ -2,12 +2,12 @@ // This will log to the console, and will not log to a file #[cfg(debug_assertions)] pub async fn start_logger() { - use tracing_subscriber::prelude::*; + use tracing_subscriber::{prelude::*, util::SubscriberInitExt}; let filter = tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(format!("{}=debug", env!("CARGO_PKG_NAME")))); + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(format!("{}=info", env!("CARGO_PKG_NAME")))); - let subscriber = tracing_subscriber::registry() + tracing_subscriber::registry() .with( tracing_subscriber::fmt::layer() .with_ansi(true) @@ -18,10 +18,8 @@ pub async fn start_logger() { .pretty() ) .with(filter) - .with(tracing_error::ErrorLayer::default()); - - tracing::subscriber::set_global_default(subscriber) - .expect("setting default subscriber failed"); + .with(tracing_error::ErrorLayer::default()) + .init(); } // Handling for the live production logging @@ -60,7 +58,7 @@ pub async fn start_logger() { let filter = tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(format!("{}=info", env!("CARGO_PKG_NAME")))); - let subscriber = tracing_subscriber::registry() + tracing_subscriber::registry() .with( tracing_subscriber::fmt::layer() .compact() @@ -69,8 +67,6 @@ pub async fn start_logger() { .with_timer(ChronoLocal::rfc_3339()), ) .with(filter) - .with(tracing_error::ErrorLayer::default()); - - tracing::subscriber::set_global_default(subscriber) - .expect("Setting default subscriber failed"); + .with(tracing_error::ErrorLayer::default()) + .init(); } diff --git a/packages/core_new/src/store/db.rs b/packages/core_new/src/store/db.rs index 7e72c977..5ec565d3 100644 --- a/packages/core_new/src/store/db.rs +++ b/packages/core_new/src/store/db.rs @@ -26,6 +26,7 @@ pub async fn create_pool_from_path(path: PathBuf) -> LauncherResult Ok(pool) } +#[tracing::instrument] pub async fn create_pool() -> LauncherResult { let path = Dirs::get().await?.db_file(); create_pool_from_path(path).await diff --git a/packages/core_new/src/store/ingress.rs b/packages/core_new/src/store/ingress.rs index 64df0f3b..2ddc9202 100644 --- a/packages/core_new/src/store/ingress.rs +++ b/packages/core_new/src/store/ingress.rs @@ -21,10 +21,15 @@ pub enum IngressError { impl IngressProcessor { #[must_use] + #[tracing::instrument] pub fn new() -> Self { - Self { + let processor = Self { ingress_feeds: RwLock::new(HashMap::new()), - } + }; + + tracing::debug!("created ingress processor"); + + processor } pub async fn create(&self, ingress_type: IngressType, message: String, total: f64) -> LauncherResult { @@ -42,24 +47,21 @@ impl IngressProcessor { feeds.insert(uuid, ingress); - self.send(uuid).await?; + // Drop the write lock to prevent deadlock + drop(feeds); + + self.send(uuid, 0.0).await?; Ok(IngressId(uuid)) } - pub async fn update(&self, id: Uuid, current: f64) -> LauncherResult<()> { + pub async fn send(&self, id: Uuid, increment: f64) -> LauncherResult<()> { let mut feeds = self.ingress_feeds.write().await; - let ingress = feeds.get_mut(&id).ok_or_else(|| IngressError::NotFound)?; - ingress.current = current; - - Ok(()) - } - - pub async fn send(&self, id: Uuid) -> LauncherResult<()> { - let feeds = self.ingress_feeds.read().await; - let ingress = feeds.get(&id).ok_or_else(|| IngressError::NotFound)?; + let ingress = feeds.get_mut(&id).ok_or(IngressError::NotFound)?; let proxy = ProxyState::get()?; + + ingress.current += increment; proxy.send_ingress(IngressPayload { id: ingress.id.0, message: ingress.message.clone(), @@ -71,9 +73,9 @@ impl IngressProcessor { Ok(()) } - pub async fn finish(&self, uuid: Uuid) -> LauncherResult { + pub async fn remove(&self, uuid: Uuid) -> LauncherResult { let mut feeds = self.ingress_feeds.write().await; - Ok(feeds.remove(&uuid).ok_or_else(|| IngressError::NotFound)?) + Ok(feeds.remove(&uuid).ok_or(IngressError::NotFound)?) } } @@ -119,7 +121,7 @@ impl Drop for IngressId { let state = State::get().await?; let processor = &state.ingress_processor; - let ingress = processor.finish(ingress_uuid).await?; + let ingress = processor.remove(ingress_uuid).await?; let proxy = ProxyState::get()?; diff --git a/packages/core_new/src/store/proxy.rs b/packages/core_new/src/store/proxy.rs index cfad055b..6a787c80 100644 --- a/packages/core_new/src/store/proxy.rs +++ b/packages/core_new/src/store/proxy.rs @@ -20,11 +20,11 @@ impl Deref for ProxyState { } impl ProxyState { - pub async fn initialize(proxy: Box) -> LauncherResult> { + pub async fn initialize(proxy: impl LauncherProxy + 'static) -> LauncherResult> { PROXY_STATE .get_or_try_init(|| async { Ok(Arc::new(Self { - inner: proxy, + inner: Box::new(proxy), })) }) .await diff --git a/packages/core_new/src/store/settings.rs b/packages/core_new/src/store/settings.rs index 3c167366..8da723a6 100644 --- a/packages/core_new/src/store/settings.rs +++ b/packages/core_new/src/store/settings.rs @@ -32,6 +32,7 @@ impl Default for Settings { } impl Settings { + #[tracing::instrument] pub async fn new() -> Self { match Self::read().await { Ok(settings) => settings, diff --git a/packages/core_new/src/store/state.rs b/packages/core_new/src/store/state.rs index 29f27a4f..12a99c86 100644 --- a/packages/core_new/src/store/state.rs +++ b/packages/core_new/src/store/state.rs @@ -41,6 +41,7 @@ impl State { LAUNCHER_STATE.initialized() } + #[tracing::instrument] async fn initialize() -> LauncherResult> { let ingress_processor = IngressProcessor::new(); let settings = Settings::new().await;