diff --git a/.ai/knowledge/crates/core/AGENTS.md b/.ai/knowledge/crates/core/AGENTS.md index 7920a9121c..3c58be14f9 100644 --- a/.ai/knowledge/crates/core/AGENTS.md +++ b/.ai/knowledge/crates/core/AGENTS.md @@ -15,11 +15,11 @@ The native app and CLI depend on these crates directly through Rust APIs. ## Crate Map -- `session`: API boundary for live sessions, operations, observing, search, exports, attachments, and unbound utility commands. +- `session`: API boundary for live sessions, operations, observing, search, exports, and attachments. - `processor`: ingestion/search pipeline; `MessageProducer` coordinates `ByteSource` and `Parser` implementations. - `sources`: byte ingestion from files, TCP, UDP, serial, processes, and pcap inputs. - `parsers`: DLT, SOME/IP, text, and plugin parser integration. -- `indexer_base`: shared chunk, time, progress, and indexing primitives. +- `indexer_base`: shared indexing primitives: index sections, timed lines, and file/line utilities. - `plugins_host`: runtime support for loading and executing WASM plugins. - `merging`: multi-source chronological merge logic. - `dlt_tools`: DLT-specific utility operations. @@ -29,7 +29,7 @@ The native app and CLI depend on these crates directly through Rust APIs. ## If You Need X, Go to Y - Change app/backend coordination: `crates/core/session/src/`. -- Add or modify session operations: `crates/core/session/src/handlers/`, `crates/core/session/src/unbound/commands/`, and `crates/stypes/src/operations/`. +- Add or modify session operations: `crates/core/session/src/handlers/` and `crates/stypes/src/operations/`. - Add or modify ingestion sources: `crates/core/sources/src/` and `crates/core/processor/src/producer/`. - Add or modify parser formats: `crates/core/parsers/src/` and `crates/core/processor/src/`. - Change search/filter behavior: `crates/core/processor/src/search/` and `crates/core/session/src/state/searchers/`. @@ -47,7 +47,7 @@ The native app and CLI depend on these crates directly through Rust APIs. - `session` owns operation state and coordinates processor work for live sessions. - `processor` reads bytes from `sources`, parses them with `parsers` or plugin-backed parsers, indexes chunks, and serves search/grab/export paths. - `plugins_host` loads plugin components and adapts them into parser/source flows. -- `stypes` carries shared command, callback, progress, plugin, observe, and error types across crate boundaries. +- `stypes` carries shared callback, plugin, observe, operation, and error types across crate boundaries. ## Landmarks and Hotspots @@ -55,7 +55,7 @@ The native app and CLI depend on these crates directly through Rust APIs. - `Session` APIs in `crates/core/session/src/session.rs` for frontend/backend coordination. - Operation handlers in `crates/core/session/src/handlers/`. - State controllers under `crates/core/session/src/state/`. -- `TimedLine` and chunk/progress types in `crates/core/indexer_base/src/`. +- `TimedLine` and index-section types in `crates/core/indexer_base/src/`. - `PluginsManager` and plugin host wrappers in `crates/core/plugins_host/src/`. - Pcap handling under `crates/core/sources/src/binary/pcap/`. diff --git a/.ai/knowledge/crates/stypes/AGENTS.md b/.ai/knowledge/crates/stypes/AGENTS.md index ab0e81ca70..1d834f538f 100644 --- a/.ai/knowledge/crates/stypes/AGENTS.md +++ b/.ai/knowledge/crates/stypes/AGENTS.md @@ -9,22 +9,20 @@ Treat changes here as cross-crate API changes. 1. `crates/stypes/Cargo.toml` 2. `crates/stypes/src/lib.rs` -3. `crates/stypes/src/command/` +3. `crates/stypes/src/observe/` 4. `crates/stypes/src/callback/` 5. `crates/stypes/src/plugins/` 6. `crates/stypes/src/error/` ## If You Need X, Go to Y -- Add or modify command/config types: `crates/stypes/src/command/`. - Add or modify callback/result payloads: `crates/stypes/src/callback/`. -- Change progress reporting types: `crates/stypes/src/progress/`. - Change plugin-facing metadata/config/result types: `crates/stypes/src/plugins/`. - Change observe/session source types: `crates/stypes/src/observe/`. - Change attachment types: `crates/stypes/src/attachment/`. - Change operation identifiers or operation payloads: `crates/stypes/src/operations/`. - Change shared error types or formatting: `crates/stypes/src/error/`. -- Change folder/profile utility command types: `crates/stypes/src/command/folders/`, `crates/stypes/src/command/profiles/`. +- Change shell profile types: `crates/stypes/src/shell/`. ## Coding Guidelines @@ -37,7 +35,7 @@ Treat changes here as cross-crate API changes. ## Technology Stack - `serde` for Rust serialization. -- `uuid`, `dlt-core`, `walkdir`, and core Rust types for shared data models. +- `uuid`, `dlt-core`, and core Rust types for shared data models. ## Validation diff --git a/Cargo.lock b/Cargo.lock index a00c0406f6..cdb8603802 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2174,9 +2174,6 @@ dependencies = [ [[package]] name = "file_tools" version = "4.1.0" -dependencies = [ - "anyhow", -] [[package]] name = "filetime" @@ -3283,12 +3280,6 @@ dependencies = [ "smallvec", ] -[[package]] -name = "lazy_static" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" - [[package]] name = "leb128" version = "0.2.6" @@ -4182,7 +4173,6 @@ dependencies = [ "chrono", "chrono-tz", "dlt-core", - "lazy_static", "log", "memchr", "rand 0.9.4", @@ -4428,7 +4418,6 @@ dependencies = [ "dirs", "log", "parsers", - "processor", "rand 0.9.4", "serde", "serde_json", @@ -4579,7 +4568,6 @@ dependencies = [ "grep-searcher", "indexer_base", "itertools 0.14.0", - "lazy_static", "log", "memchr", "parsers", @@ -5362,14 +5350,10 @@ name = "session" version = "4.1.0" dependencies = [ "bincode", - "blake3", "dirs", - "dlt-core", - "file_tools", "futures", "indexer_base", "insta", - "lazy_static", "log", "merging", "mime_guess", @@ -5378,18 +5362,13 @@ dependencies = [ "processor", "rustc-hash 2.1.2", "serde", - "serde_json", - "serialport", - "shell_tools", "sources", "stypes", "tempfile", "thiserror 2.0.18", "tokio", - "tokio-stream", "tokio-util", "uuid", - "walkdir", ] [[package]] @@ -5695,7 +5674,6 @@ dependencies = [ "thiserror 2.0.18", "tokio", "uuid", - "walkdir", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index fdd68dd0ff..b29253b9e5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -64,7 +64,6 @@ log = "0.4" serde = { version = "1.0", features = ["derive", "rc"] } serde_json = "1.0" thiserror = "2.0" -lazy_static = "1.5" memchr = "2.7" bincode = "1.3" bytecount = "0.6" @@ -95,7 +94,6 @@ pcap-parser = "0.16" tempfile = "3.27" env_logger = "0.11" log4rs = "1.4" -walkdir = "2.5" anyhow = "1.0" toml = "0.8" blake3 = "1.8" diff --git a/crates/app/Cargo.toml b/crates/app/Cargo.toml index d960fba193..3754fb3160 100644 --- a/crates/app/Cargo.toml +++ b/crates/app/Cargo.toml @@ -45,7 +45,7 @@ chrono.workspace = true chrono-tz.workspace = true dirs.workspace = true serialport.workspace = true -dlt-core.workspace = true +dlt-core = { workspace = true, features = ["statistics"] } rustc-hash.workspace = true regex.workspace = true regex-syntax.workspace = true diff --git a/crates/core/session/src/unbound/cleanup.rs b/crates/app/src/host/service/cleanup.rs similarity index 60% rename from crates/core/session/src/unbound/cleanup.rs rename to crates/app/src/host/service/cleanup.rs index fb8a4c4572..09e233da16 100644 --- a/crates/core/session/src/unbound/cleanup.rs +++ b/crates/app/src/host/service/cleanup.rs @@ -1,4 +1,4 @@ -//! Function to cleanup temp files and directories created by Chipmunk and haven't +//! Cleanup of temp files and directories created by Chipmunk that haven't //! gotten cleaned up properly. use std::{ @@ -7,71 +7,72 @@ use std::{ time::{Duration, SystemTime}, }; -use crate::paths::{get_chipmunk_downloads_dir, get_streams_dir}; +use log::error; +use semver::Version; +use session_core::paths::{get_chipmunk_downloads_dir, get_streams_dir}; +use stypes::{NativeError, NativeErrorKind, Severity}; -/// Cleanup temporary files that is generated by Chipmunk on previous sessions. +use crate::{common::app_info, host::service::update::version_dir_prefix}; + +/// Cleans up temporary files that were generated by Chipmunk on previous sessions. /// -/// If provided, `download_prefix` removes matching direct child directories in -/// the downloads directory in the same scan as age-based cleanup. -pub fn cleanup_temp_files(download_prefix: Option<&str>) -> Result<(), Vec> { - let res_tmp = cleanup_temp_dir(); - let res_download = cleanup_download_dir(download_prefix); - - let errs: Vec<_> = [res_tmp, res_download] - .into_iter() - .filter_map(|res| res.err()) - .collect(); - - if errs.is_empty() { Ok(()) } else { Err(errs) } +/// Both cleanups are always attempted, failures are logged only since the caller +/// has no way to recover from them. +pub fn cleanup_temp_files() { + if let Err(err) = cleanup_temp_dir() { + error!("Error while cleaning up temporary files. Error: {err:?}"); + } + + if let Err(err) = cleanup_download_dir() { + error!("Error while cleaning up downloaded files. Error: {err:?}"); + } } /// Iterates through chipmunk temporary directory and remove the entries older /// than one months. -fn cleanup_temp_dir() -> Result<(), stypes::NativeError> { +fn cleanup_temp_dir() -> Result<(), NativeError> { + const ONE_MONTH: Duration = Duration::from_secs(30 * 24 * 60 * 60); + let tmp_dir = get_streams_dir()?; + cleanup_dir(&tmp_dir, stale_before(ONE_MONTH)?)?; + + Ok(()) +} - const ONE_MONTHS_SECONDS: u64 = 30 * 60 * 24 * 60; - let modified_limit = SystemTime::now() - .checked_sub(Duration::from_secs(ONE_MONTHS_SECONDS)) - .ok_or_else(|| stypes::NativeError { - severity: stypes::Severity::ERROR, - kind: stypes::NativeErrorKind::Io, +/// Modification time before which an entry counts as stale. +fn stale_before(max_age: Duration) -> Result { + SystemTime::now() + .checked_sub(max_age) + .ok_or_else(|| NativeError { + severity: Severity::ERROR, + kind: NativeErrorKind::Io, message: Some(String::from( "Error while calculating modification time limit", )), - })?; - - cleanup_dir(&tmp_dir, modified_limit)?; - - Ok(()) + }) } -/// Iterates through chipmunk downloads directory and remove stale entries or -/// fresh matching download directories. -fn cleanup_download_dir(prefix: Option<&str>) -> Result<(), stypes::NativeError> { +/// Iterates through chipmunk downloads directory and remove stale entries alongside +/// with the staging directories of the running version. +fn cleanup_download_dir() -> Result<(), NativeError> { + const ONE_WEEK: Duration = Duration::from_secs(7 * 24 * 60 * 60); + let download_dir = get_chipmunk_downloads_dir()?; if !download_dir.exists() { return Ok(()); } - const ONE_WEEK_SECONDS: u64 = 7 * 60 * 24 * 60; - let modified_limit = SystemTime::now() - .checked_sub(Duration::from_secs(ONE_WEEK_SECONDS)) - .ok_or_else(|| stypes::NativeError { - severity: stypes::Severity::ERROR, - kind: stypes::NativeErrorKind::Io, - message: Some(String::from( - "Error while calculating modification time limit", - )), - })?; - - cleanup_download_entries(&download_dir, modified_limit, prefix)?; + // Staging directories of the running version can't belong to an install in + // progress because the app has just started. + let date_limit = stale_before(ONE_WEEK)?; + let current_version = app_info::current_version(); + cleanup_download_entries(&download_dir, date_limit, current_version)?; Ok(()) } -// Clean files and directory within the given path that have a modified time older than -// the given modified date limit +/// Clean files and directory within the given path that have a modified time older than +/// the given modified date limit fn cleanup_dir(path: &Path, modified_date_limit: SystemTime) -> io::Result<()> { if !path.exists() { return Ok(()); @@ -84,35 +85,36 @@ fn cleanup_dir(path: &Path, modified_date_limit: SystemTime) -> io::Result<()> { .try_for_each(remove_entry) } +/// Clean stale entries within the given path, plus the staging directories of the +/// given version. fn cleanup_download_entries( path: &Path, modified_date_limit: SystemTime, - prefix: Option<&str>, + version: &Version, ) -> io::Result<()> { if !path.exists() { return Ok(()); } + let prefix = version_dir_prefix(version); + fs::read_dir(path)? .flat_map(Result::ok) .filter(|entry| { - is_stale_entry(entry, modified_date_limit) || is_prefixed_dir(entry, prefix) + is_stale_entry(entry, modified_date_limit) || is_prefixed_dir(entry, &prefix) }) .map(|entry| entry.path()) .try_for_each(remove_entry) } +/// Returns `true` when the entry was last modified before the given limit. fn is_stale_entry(entry: &fs::DirEntry, modified_date_limit: SystemTime) -> bool { entry .metadata() .is_ok_and(|meta| meta.modified().is_ok_and(|date| date < modified_date_limit)) } -fn is_prefixed_dir(entry: &fs::DirEntry, prefix: Option<&str>) -> bool { - let Some(prefix) = prefix.filter(|prefix| !prefix.is_empty()) else { - return false; - }; - +fn is_prefixed_dir(entry: &fs::DirEntry, prefix: &str) -> bool { entry.file_type().is_ok_and(|file_type| file_type.is_dir()) && entry .file_name() @@ -138,6 +140,8 @@ mod tests { time::{Duration, SystemTime}, }; + use semver::Version; + use super::{cleanup_dir, cleanup_download_entries}; #[test] @@ -159,29 +163,13 @@ mod tests { let old_limit = SystemTime::now() .checked_sub(Duration::from_secs(3600)) .unwrap(); - cleanup_download_entries(temp_path, old_limit, Some("update_4.0.1_")).unwrap(); + cleanup_download_entries(temp_path, old_limit, &Version::new(4, 0, 1)).unwrap(); assert!(!current_update.exists()); assert!(other_update.exists()); assert!(matching_file.exists()); } - #[test] - fn cleanup_download_entries_ignores_empty_prefix() { - let tempdir = tempfile::tempdir().unwrap(); - let temp_path = tempdir.path(); - - let update_dir = temp_path.join("update_4.0.1_123"); - fs::create_dir(&update_dir).unwrap(); - - let old_limit = SystemTime::now() - .checked_sub(Duration::from_secs(3600)) - .unwrap(); - cleanup_download_entries(temp_path, old_limit, Some("")).unwrap(); - - assert!(update_dir.exists()); - } - #[test] fn test_cleanup_dir() { // Create temporary directory with some entries diff --git a/crates/app/src/host/service/mod.rs b/crates/app/src/host/service/mod.rs index 8d3b060028..a45b60fa8f 100644 --- a/crates/app/src/host/service/mod.rs +++ b/crates/app/src/host/service/mod.rs @@ -18,7 +18,6 @@ use stypes::{ }; use crate::{ - common::app_info, host::{ command::{ DltStatisticsParam, ExportPresetsParam, HostCommand, ScanFavoriteFoldersParam, @@ -58,6 +57,7 @@ use plugin::{PluginEvent, PluginService}; use presets_io::{import_named_presets, serialize_named_presets}; use storage::StorageService; +mod cleanup; pub mod file; mod plugin; mod presets_io; @@ -150,7 +150,7 @@ impl HostService { plugins, }; - Self::spawn_startup_cleanup(); + tokio::task::spawn_blocking(cleanup::cleanup_temp_files); let previous_version = storage::app_version::sync_current_version(); update::spawn_check( host.communication.senders.clone(), @@ -166,17 +166,6 @@ impl HostService { .expect("Receiving startup state should never fail") } - fn spawn_startup_cleanup() { - tokio::task::spawn_blocking(move || { - let current_update = format!("update_{}_", app_info::current_version()); - if let Err(errs) = session_core::unbound::cleanup_temp_files(Some(¤t_update)) { - for err in errs { - log::error!("Error while cleaning up temporary files. Error: {err:?}"); - } - } - }); - } - async fn run(mut self) { loop { select! { diff --git a/crates/app/src/host/service/update/download.rs b/crates/app/src/host/service/update/download.rs index 3eba975ad0..c4e3a49cf2 100644 --- a/crates/app/src/host/service/update/download.rs +++ b/crates/app/src/host/service/update/download.rs @@ -8,6 +8,7 @@ use thiserror::Error; use tokio::{fs, io::AsyncWriteExt}; use uuid::Uuid; +use super::version_dir_prefix; use crate::host::ui::update::{DownloadUpdateParam, DownloadedUpdate}; const USER_AGENT: &str = "chipmunk"; @@ -60,11 +61,12 @@ fn unique_download_dir(request: &DownloadUpdateParam) -> Result String { + format!("update_{version}_") +} + /// Spawns a selected app update download request. pub fn spawn_download_update(senders: ServiceSenders, request: DownloadUpdateParam) { tokio::spawn(async move { diff --git a/crates/app/src/session/service/mod.rs b/crates/app/src/session/service/mod.rs index 1eb04decbe..9d5233ed28 100644 --- a/crates/app/src/session/service/mod.rs +++ b/crates/app/src/session/service/mod.rs @@ -916,10 +916,6 @@ impl SessionService { .send_session_msg(SessionMessage::AttachmentsUpdated { attachments, len }) .await; } - event => { - println!("************** DEBUG: Received unhandled callback: {event:?}"); - log::warn!("Unhandled callback: {event}"); - } } Ok(()) diff --git a/crates/core/parsers/Cargo.toml b/crates/core/parsers/Cargo.toml index 0c4d72ab6d..7bcff968f5 100644 --- a/crates/core/parsers/Cargo.toml +++ b/crates/core/parsers/Cargo.toml @@ -16,7 +16,6 @@ byteorder.workspace = true chrono.workspace = true chrono-tz.workspace = true dlt-core = { workspace = true, features = ["serialization", "fibex"] } -lazy_static.workspace = true log.workspace = true regex.workspace = true memchr.workspace = true diff --git a/crates/core/parsers/src/someip.rs b/crates/core/parsers/src/someip.rs index 15fbb65613..e03d31eb72 100644 --- a/crates/core/parsers/src/someip.rs +++ b/crates/core/parsers/src/someip.rs @@ -6,7 +6,7 @@ use std::{ fmt::{self, Display}, io::Write, path::PathBuf, - sync::Mutex, + sync::{LazyLock, Mutex}, }; use someip_messages::*; @@ -17,7 +17,6 @@ use someip_payload::{ }; use stypes::SomeipFilterConfig; -use lazy_static::lazy_static; use log::{debug, error}; use regex::Regex; use serde::Serialize; @@ -564,11 +563,11 @@ impl fmt::Debug for SomeipLogMessage { /// Merges the SOME/IP message columns to a single column with additional info. fn merge_columns(columns: &str) -> Cow<'_, str> { - lazy_static! { - static ref REGEX : Regex = Regex::new( + static REGEX: LazyLock = LazyLock::new(|| { + Regex::new( &format!("(SD|RPC){COLUMN_SEP}(\\d+){COLUMN_SEP}(\\d+){COLUMN_SEP}(\\d+){COLUMN_SEP}(\\d+){COLUMN_SEP}(\\d+){COLUMN_SEP}(\\d+){COLUMN_SEP}(\\d+){COLUMN_SEP}(\\d+){COLUMN_SEP}(.*)") - ).unwrap(); - } + ).expect("SOME/IP columns pattern is static and must compile") + }); REGEX.replace(columns, "${1} SERV:${2} METH:${3} LENG:${4} CLID:${5} SEID:${6} IVER:${7} MSTP:${8} RETC:${9} ${10}") } diff --git a/crates/core/plugins_host/Cargo.toml b/crates/core/plugins_host/Cargo.toml index 687bb18e51..d7b87f1cac 100644 --- a/crates/core/plugins_host/Cargo.toml +++ b/crates/core/plugins_host/Cargo.toml @@ -27,9 +27,6 @@ wasmtime.workspace = true wasmtime-wasi.workspace = true [dev-dependencies] -# Internal crates -processor.workspace = true - # External crates criterion = { workspace = true, features = ["async_tokio"] } diff --git a/crates/core/processor/Cargo.toml b/crates/core/processor/Cargo.toml index 167b48ede7..fe93966d30 100644 --- a/crates/core/processor/Cargo.toml +++ b/crates/core/processor/Cargo.toml @@ -24,7 +24,6 @@ futures.workspace = true grep-regex.workspace = true grep-searcher.workspace = true itertools.workspace = true -lazy_static.workspace = true log.workspace = true memchr.workspace = true regex.workspace = true diff --git a/crates/core/processor/src/lib.rs b/crates/core/processor/src/lib.rs index 4bd108f368..cbbb3cd125 100644 --- a/crates/core/processor/src/lib.rs +++ b/crates/core/processor/src/lib.rs @@ -11,7 +11,6 @@ // is strictly forbidden unless prior written permission is obtained // from E.S.R.Labs. extern crate indexer_base; -extern crate lazy_static; #[macro_use] extern crate log; diff --git a/crates/core/session/Cargo.toml b/crates/core/session/Cargo.toml index 18d20a5f6c..30cfdba2c5 100644 --- a/crates/core/session/Cargo.toml +++ b/crates/core/session/Cargo.toml @@ -9,8 +9,6 @@ workspace = true [dependencies] # Internal crates -file_tools.workspace = true -shell_tools.workspace = true indexer_base.workspace = true merging.workspace = true parsers.workspace = true @@ -20,27 +18,19 @@ sources.workspace = true plugins_host.workspace = true # External crates -blake3.workspace = true dirs.workspace = true -dlt-core = { workspace = true, features = ["statistics", "serialization"] } futures.workspace = true -lazy_static.workspace = true log.workspace = true mime_guess.workspace = true rustc-hash.workspace = true serde = { workspace = true , features = ["derive"] } bincode.workspace = true -serde_json.workspace = true -serialport.workspace = true thiserror.workspace = true tokio = { workspace = true , features = ["full"] } -tokio-stream.workspace = true tokio-util.workspace = true uuid = { workspace = true , features = ["serde", "v4"] } -walkdir.workspace = true [dev-dependencies] # External crates -lazy_static.workspace = true tempfile.workspace = true insta.workspace = true diff --git a/crates/core/session/src/handlers/mod.rs b/crates/core/session/src/handlers/mod.rs index f0301268e7..e1c1d22352 100644 --- a/crates/core/session/src/handlers/mod.rs +++ b/crates/core/session/src/handlers/mod.rs @@ -4,4 +4,3 @@ pub mod observe; mod observing; pub mod search; pub mod search_values; -pub mod sleep; diff --git a/crates/core/session/src/handlers/sleep.rs b/crates/core/session/src/handlers/sleep.rs deleted file mode 100644 index 9c32961d6f..0000000000 --- a/crates/core/session/src/handlers/sleep.rs +++ /dev/null @@ -1,25 +0,0 @@ -use crate::operations::{OperationAPI, OperationResult}; -use tokio::{select, time}; - -pub async fn handle( - operation_api: &OperationAPI, - ms: u64, - ignore_cancellation: bool, -) -> OperationResult { - if ignore_cancellation { - time::sleep(time::Duration::from_millis(ms)).await; - Ok(Some(stypes::ResultSleep { sleep_well: true })) - } else { - let canceler = operation_api.cancellation_token(); - select! { - _ = async move { - time::sleep(time::Duration::from_millis(ms)).await; - } => { - Ok(Some( stypes::ResultSleep { sleep_well: true })) - }, - _ = canceler.cancelled() => { - Ok(Some( stypes::ResultSleep { sleep_well: false })) - } - } - } -} diff --git a/crates/core/session/src/lib.rs b/crates/core/session/src/lib.rs index 81f6bb982d..803e73b57a 100644 --- a/crates/core/session/src/lib.rs +++ b/crates/core/session/src/lib.rs @@ -1,24 +1,7 @@ mod handlers; pub mod operations; pub mod paths; -pub mod progress; pub mod session; pub mod state; pub mod tail; pub mod tracker; -pub mod unbound; - -use std::sync::Mutex; -use tokio::sync::mpsc; - -extern crate lazy_static; - -lazy_static::lazy_static! { - pub static ref TRACKER_CHANNEL: Mutex<( - mpsc::UnboundedSender, - Option> - )> = { - let (tx, rx) = mpsc::unbounded_channel(); - Mutex::new((tx, Some(rx))) - }; -} diff --git a/crates/core/session/src/operations.rs b/crates/core/session/src/operations.rs index 39ebc04f89..4606518c47 100644 --- a/crates/core/session/src/operations.rs +++ b/crates/core/session/src/operations.rs @@ -7,11 +7,7 @@ use merging::merger::FileMergeOptions; use processor::search::filter::SearchFilter; use serde::Serialize; use sources::sde::{SdeReceiver, SdeSender}; -use std::{ - ops::RangeInclusive, - path::PathBuf, - time::{SystemTime, UNIX_EPOCH}, -}; +use std::{ops::RangeInclusive, path::PathBuf}; use tokio::{ sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, @@ -20,46 +16,6 @@ use tokio::{ use tokio_util::sync::CancellationToken; use uuid::Uuid; -#[derive(Debug, Clone, Serialize)] -pub struct OperationStat { - pub uuid: String, - pub name: String, - pub duration: u64, - pub started: u64, -} - -impl OperationStat { - pub fn new(uuid: String, name: String) -> Self { - let start = SystemTime::now(); - let timestamp = match start.duration_since(UNIX_EPOCH) { - Ok(timestamp) => timestamp.as_micros() as u64, - Err(err) => { - error!("Failed to get timestamp: {err}"); - 0 - } - }; - OperationStat { - uuid, - name, - started: timestamp, - duration: 0, - } - } - pub fn done(&mut self) { - let start = SystemTime::now(); - let timestamp = match start.duration_since(UNIX_EPOCH) { - Ok(timestamp) => timestamp.as_micros() as u64, - Err(err) => { - error!("Failed to get timestamp: {err}"); - 0 - } - }; - if timestamp > self.started { - self.duration = timestamp - self.started; - } - } -} - #[derive(Debug)] pub struct Operation { kind: OperationKind, @@ -133,7 +89,6 @@ pub enum OperationKind { Cancel { target: Uuid, }, - Sleep(u64, bool), End, } @@ -152,7 +107,6 @@ impl std::fmt::Display for OperationKind { OperationKind::Map { .. } => "Mapping", OperationKind::Values { .. } => "Values", OperationKind::Merge { .. } => "Merging", - OperationKind::Sleep(_, _) => "Sleeping", OperationKind::Cancel { .. } => "Canceling", OperationKind::End => "End", } @@ -283,7 +237,6 @@ impl OperationAPI { .tracker_api .add_operation( self.id(), - operation.kind.to_string(), tx_sde, self.cancellation_token(), self.done_token(), @@ -428,13 +381,6 @@ impl OperationAPI { } => { unimplemented!("merging not yet supported"); } - OperationKind::Sleep(ms, ignore_cancellation) => { - api.finish( - handlers::sleep::handle(&api, ms, ignore_cancellation).await, - operation_str, - ) - .await; - } OperationKind::Cancel { target } => match tracker.cancel_operation(target).await { Ok(canceled) => { if canceled { diff --git a/crates/core/session/src/progress.rs b/crates/core/session/src/progress.rs deleted file mode 100644 index 28df65693b..0000000000 --- a/crates/core/session/src/progress.rs +++ /dev/null @@ -1,187 +0,0 @@ -//! Includes functionality to keep track on the current progress of the on-going tasks. - -use crate::TRACKER_CHANNEL; -use log::{error, info}; -use std::collections::HashMap; -use tokio::{ - select, - sync::{ - mpsc::{self, UnboundedReceiver, UnboundedSender, error::SendError, unbounded_channel}, - oneshot, - }, -}; -use uuid::Uuid; - -/// Commands used to control/query the progress tracking -#[derive(Debug)] -pub enum ProgressCommand { - Content(oneshot::Sender>), - Abort(oneshot::Sender>), -} - -#[derive(Clone, Debug)] -pub struct ProgressProviderAPI { - tx: UnboundedSender, -} - -impl ProgressProviderAPI { - pub fn new() -> Result { - let tx = { - let tx_rx = TRACKER_CHANNEL.lock().map_err(|e| { - stypes::ComputationError::Communication(format!( - "Cannot init channels from mutex: {e}" - )) - })?; - tx_rx.0.clone() - // scope will release Mutex lock - }; - Ok(Self { tx }) - } - - pub fn started(&self, alias: &str, uuid: &Uuid) { - if self - .tx - .send(stypes::LifecycleTransition::started(uuid, alias)) - .is_err() - { - error!("Fail to report LifecycleTransition::Started. Channel is closed"); - } - } - - pub fn stopped(&self, uuid: &Uuid) { - if self - .tx - .send(stypes::LifecycleTransition::stopped(uuid)) - .is_err() - { - error!("Fail to report LifecycleTransition::Stopped. Channel is closed"); - } - } - - pub fn progress(&self, uuid: &Uuid, ticks: stypes::Ticks) { - if self - .tx - .send(stypes::LifecycleTransition::ticks(uuid, ticks)) - .is_err() - { - error!("Fail to report LifecycleTransition::Ticks. Channel is closed"); - } - } -} - -/// The ProgressTrackerAPI enables safe access to the state of the progress of -/// all ongoing operations (all unbound jobs and all operations in every session) -#[derive(Clone, Debug)] -pub struct ProgressTrackerAPI { - tx_api: UnboundedSender, -} - -impl ProgressTrackerAPI { - /// creates a ProgressTrackerAPI along with a channel that needs - /// to be passed to the run_tracking function - pub fn new() -> (Self, UnboundedReceiver) { - let (tx_api, rx_api) = unbounded_channel(); - (Self { tx_api }, rx_api) - } - - async fn exec_operation( - &self, - command: ProgressCommand, - rx_response: oneshot::Receiver, - ) -> Result { - let api_str = format!("{command:?}"); - self.tx_api.send(command).map_err(|e| { - stypes::ComputationError::Communication(format!( - "Failed to send to Api::{api_str}; error: {e}" - )) - })?; - rx_response.await.map_err(|_| { - stypes::ComputationError::Communication(format!( - "Failed to get response from Api::{api_str}" - )) - }) - } - - pub async fn content(&self) -> Result { - let (tx, rx) = oneshot::channel(); - self.exec_operation(ProgressCommand::Content(tx), rx) - .await? - } - - pub async fn abort(&self) -> Result<(), stypes::ComputationError> { - let (tx, rx) = oneshot::channel(); - self.exec_operation(ProgressCommand::Abort(tx), rx).await? - } -} - -fn log_if_err(res: Result<(), SendError>) { - if res.is_err() { - error!("Fail to send event into lifecycle_events_channel. Channel is closed"); - } -} - -/// Keep track of all ongoing operations and jobs -/// All jobs and operations are identified with UUIDs. Here we receive updates about the -/// progress of those long-runing operations. -/// At any time, we can then track the progress of everything that is going on -pub async fn run_tracking( - mut command_rx: UnboundedReceiver, -) -> Result, stypes::ComputationError> { - let mut ongoing_operations: HashMap = HashMap::new(); - let lifecycle_events_channel = mpsc::channel(1); - - let mut lifecycle_events = { - let mut tx_rx = TRACKER_CHANNEL.lock().map_err(|e| { - stypes::ComputationError::Communication(format!("Cannot init channels from mutex: {e}")) - })?; - tx_rx - .1 - .take() - .ok_or(stypes::ComputationError::Communication( - "ProgressTracker channel already taken".to_string(), - ))? - }; - - tokio::spawn(async move { - loop { - select! { - command = command_rx.recv() => { - match command { - Some(ProgressCommand::Content(result_channel)) => { - let res = serde_json::to_string(&ongoing_operations) - .map_err(|e| stypes::ComputationError::Process(format!("{e}"))); - let _ = result_channel.send(res); - } - Some(ProgressCommand::Abort(result_channel)) => { - let _ = result_channel.send(Ok(())); - break; - } - None => break, - } - } - lifecycle_event = lifecycle_events.recv() => { - match lifecycle_event { - Some(stypes::LifecycleTransition::Started { uuid, alias }) => { - info!("job {alias} ({uuid}) started"); - ongoing_operations.insert(uuid, stypes::Ticks::default()); - log_if_err(lifecycle_events_channel.0.send(stypes::LifecycleTransition::started(&uuid, &alias)).await); - } - Some(stypes::LifecycleTransition::Stopped(uuid)) => { - info!("job {uuid} stopped"); - ongoing_operations.remove(&uuid); - log_if_err(lifecycle_events_channel.0.send(stypes::LifecycleTransition::Stopped(uuid)).await); - } - Some(stypes::LifecycleTransition::Ticks {uuid, ticks}) => { - info!("job {uuid} reported progress: {ticks:?}"); - ongoing_operations.insert(uuid, ticks.clone()); - log_if_err(lifecycle_events_channel.0.send(stypes::LifecycleTransition::ticks(&uuid, ticks)).await); - } - None => break, - - } - } - } - } - }); - Ok(lifecycle_events_channel.1) -} diff --git a/crates/core/session/src/session.rs b/crates/core/session/src/session.rs index e8f5e51d13..d8405b43f1 100644 --- a/crates/core/session/src/session.rs +++ b/crates/core/session/src/session.rs @@ -532,35 +532,6 @@ impl Session { .map(|v| v.into()) .map_err(stypes::ComputationError::NativeError) } - - /// Used for debug goals - pub fn sleep( - &self, - operation_id: Uuid, - ms: u64, - ignore_cancellation: bool, - ) -> Result<(), stypes::ComputationError> { - self.tx_operations - .send(Operation::new( - operation_id, - operations::OperationKind::Sleep(ms, ignore_cancellation), - )) - .map_err(|e| stypes::ComputationError::Communication(e.to_string())) - } - - /// Used for debug goals - pub async fn trigger_state_error(&self) -> Result<(), stypes::ComputationError> { - self.state - .shutdown_with_error() - .map_err(stypes::ComputationError::NativeError) - } - - /// Used for debug goals - pub async fn trigger_tracker_error(&self) -> Result<(), stypes::ComputationError> { - self.tracker - .shutdown_with_error() - .map_err(stypes::ComputationError::NativeError) - } } impl Drop for Session { diff --git a/crates/core/session/src/state/api.rs b/crates/core/session/src/state/api.rs index de4e40ecbe..44fc9a26d3 100644 --- a/crates/core/session/src/state/api.rs +++ b/crates/core/session/src/state/api.rs @@ -192,13 +192,10 @@ pub enum Api { DropSearchValues(oneshot::Sender), GetIndexedRanges(oneshot::Sender>>), CloseSession(oneshot::Sender<()>), - SetDebugMode((bool, oneshot::Sender<()>)), NotifyCancelingOperation(Uuid), NotifyCanceledOperation(Uuid), AddAttachments(Vec), GetAttachments(oneshot::Sender>), - // Used for tests of error handeling - ShutdownWithError, Shutdown, } @@ -247,13 +244,11 @@ impl Display for Api { Self::DropSearchValues(_) => "DropSearchValues", Self::GetIndexedRanges(_) => "GetIndexedRanges", Self::CloseSession(_) => "CloseSession", - Self::SetDebugMode(_) => "SetDebugMode", Self::NotifyCancelingOperation(_) => "NotifyCancelingOperation", Self::NotifyCanceledOperation(_) => "NotifyCanceledOperation", Self::AddAttachments(_) => "AddAttachments", Self::GetAttachments(_) => "GetAttachments", Self::Shutdown => "Shutdown", - Self::ShutdownWithError => "ShutdownWithError", } ) } @@ -666,26 +661,12 @@ impl SessionStateAPI { self.exec_operation(Api::CloseSession(tx), rx).await } - pub async fn set_debug(&self, debug: bool) -> Result<(), stypes::NativeError> { - let (tx, rx) = oneshot::channel(); - self.exec_operation(Api::SetDebugMode((debug, tx)), rx) - .await - } - pub fn shutdown(&self) -> Result<(), stypes::NativeError> { self.tx_api.send(Api::Shutdown).map_err(|e| { stypes::NativeError::channel(&format!("fail to send to Api::Shutdown; error: {e}",)) }) } - pub fn shutdown_with_error(&self) -> Result<(), stypes::NativeError> { - self.tx_api.send(Api::ShutdownWithError).map_err(|e| { - stypes::NativeError::channel(&format!( - "fail to send to Api::ShutdownWithError; error: {e}", - )) - }) - } - /// Sends the given attachments to the session state to be stored and delivered to the /// clients within a single event. /// diff --git a/crates/core/session/src/state/mod.rs b/crates/core/session/src/state/mod.rs index 53a7e2b637..57c098bd10 100644 --- a/crates/core/session/src/state/mod.rs +++ b/crates/core/session/src/state/mod.rs @@ -83,7 +83,6 @@ pub struct SessionState { pub cancelling_operations: HashMap, pub status: Status, searcher_tx: mpsc::Sender, - pub debug: bool, } impl SessionState { @@ -101,7 +100,6 @@ impl SessionState { status: Status::Open, cancelling_operations: HashMap::new(), searcher_tx, - debug: false, } } @@ -963,14 +961,6 @@ async fn handle_api_msg( )); } } - Api::SetDebugMode((debug, tx_response)) => { - state.debug = debug; - if tx_response.send(()).is_err() { - return Err(stypes::NativeError::channel( - "fail to response to Api::SetDebugMode", - )); - } - } Api::NotifyCancelingOperation(uuid) => { state.cancelling_operations.insert(uuid, true); } @@ -991,14 +981,6 @@ async fn handle_api_msg( debug!("shutdown has been requested"); return Ok(HanldeOutpt::Break); } - Api::ShutdownWithError => { - debug!("shutdown state loop with error for testing"); - return Err(stypes::NativeError { - severity: stypes::Severity::ERROR, - kind: stypes::NativeErrorKind::Io, - message: Some(String::from("Shutdown state loop with error for testing")), - }); - } } Ok(HanldeOutpt::None) diff --git a/crates/core/session/src/tracker.rs b/crates/core/session/src/tracker.rs index 456d79b942..5e6f249364 100644 --- a/crates/core/session/src/tracker.rs +++ b/crates/core/session/src/tracker.rs @@ -1,7 +1,7 @@ //! Contains the functionality for adding and tracking all operations running in core //! including cancelling them as well. -use crate::{operations::OperationStat, progress::ProgressProviderAPI, state::SessionStateAPI}; +use crate::state::SessionStateAPI; use log::{debug, error}; use sources::sde::SdeSender; use std::collections::{HashMap, hash_map::Entry}; @@ -21,7 +21,6 @@ pub enum TrackerCommand { AddOperation( ( Uuid, - String, Option, CancellationToken, CancellationToken, @@ -30,13 +29,9 @@ pub enum TrackerCommand { ), RemoveOperation((Uuid, oneshot::Sender)), CancelOperation((Uuid, oneshot::Sender)), - SetDebugMode((bool, oneshot::Sender<()>)), - GetOperationsStat(oneshot::Sender>), GetSdeSender((Uuid, oneshot::Sender>)), CancelAll(oneshot::Sender<()>), Shutdown, - // Used for tests of error handeling - ShutdownWithError, } impl std::fmt::Display for TrackerCommand { @@ -48,22 +43,20 @@ impl std::fmt::Display for TrackerCommand { Self::AddOperation(_) => "AddOperation", Self::RemoveOperation(_) => "RemoveOperation", Self::CancelOperation(_) => "CancelOperation", - Self::SetDebugMode(_) => "SetDebugMode", - Self::GetOperationsStat(_) => "GetOperationsStat", Self::GetSdeSender(_) => "GetSdeSender", Self::CancelAll(_) => "CancelAll", Self::Shutdown => "Shutdown", - Self::ShutdownWithError => "ShutdownWithError", } ) } } +/// Cancellation and data-exchange handles of one running operation. #[derive(Debug)] -pub struct OperationTracker { - pub operations: HashMap, CancellationToken, CancellationToken)>, - pub stat: Vec, - pub debug: bool, +struct TrackedOperation { + sde_tx: Option, + canceler: CancellationToken, + done: CancellationToken, } #[derive(Clone, Debug)] @@ -94,14 +87,13 @@ impl OperationTrackerAPI { pub async fn add_operation( &self, uuid: Uuid, - name: String, tx_sde: Option, canceler: CancellationToken, done: CancellationToken, ) -> Result { let (tx, rx) = oneshot::channel(); self.exec_operation( - TrackerCommand::AddOperation((uuid, name, tx_sde, canceler, done, tx)), + TrackerCommand::AddOperation((uuid, tx_sde, canceler, done, tx)), rx, ) .await @@ -124,19 +116,6 @@ impl OperationTrackerAPI { self.exec_operation(TrackerCommand::CancelAll(tx), rx).await } - pub async fn set_debug(&self, debug: bool) -> Result<(), stypes::NativeError> { - let (tx, rx) = oneshot::channel(); - self.exec_operation(TrackerCommand::SetDebugMode((debug, tx)), rx) - .await?; - Ok(()) - } - - pub async fn get_operations_stat(&self) -> Result { - let (tx, rx) = oneshot::channel(); - self.exec_operation(TrackerCommand::GetOperationsStat(tx), rx) - .await? - } - pub async fn get_sde_sender( &self, uuid: Uuid, @@ -151,48 +130,25 @@ impl OperationTrackerAPI { stypes::NativeError::channel(&format!("fail to send to Api::Shutdown; error: {e}",)) }) } - - pub fn shutdown_with_error(&self) -> Result<(), stypes::NativeError> { - self.tx_api - .send(TrackerCommand::ShutdownWithError) - .map_err(|e| { - stypes::NativeError::channel(&format!( - "fail to send to Api::ShutdownWithError; error: {e}", - )) - }) - } } pub async fn run( state: SessionStateAPI, mut rx_api: UnboundedReceiver, ) -> Result<(), stypes::NativeError> { - let mut tracker = OperationTracker { - operations: HashMap::new(), - stat: vec![], - debug: false, - }; - let progress = ProgressProviderAPI::new()?; + let mut operations: HashMap = HashMap::new(); debug!("task is started"); while let Some(msg) = rx_api.recv().await { match msg { - TrackerCommand::AddOperation(( - uuid, - name, - tx_sde, - cancalation_token, - done_token, - tx_response, - )) => { - if tracker.debug { - tracker - .stat - .push(OperationStat::new(uuid.to_string(), name.clone())); - } + TrackerCommand::AddOperation((uuid, sde_tx, canceler, done, tx_response)) => { if tx_response - .send(match tracker.operations.entry(uuid) { + .send(match operations.entry(uuid) { Entry::Vacant(entry) => { - entry.insert((tx_sde, cancalation_token, done_token)); + entry.insert(TrackedOperation { + sde_tx, + canceler, + done, + }); true } _ => false, @@ -202,25 +158,14 @@ pub async fn run( return Err(stypes::NativeError::channel( "fail to response to Api::AddOperation", )); - } else { - progress.started(&name, &uuid); } } TrackerCommand::RemoveOperation((uuid, tx_response)) => { if let Err(err) = state.canceled_operation(uuid).await { error!("fail to notify state about canceled operation {uuid}; err: {err:?}"); } - if tracker.debug { - let str_uuid = uuid.to_string(); - if let Some(index) = tracker.stat.iter().position(|op| op.uuid == str_uuid) { - tracker.stat[index].done(); - } else { - error!("fail to find operation in stat: {str_uuid}"); - } - } - progress.stopped(&uuid); if tx_response - .send(tracker.operations.remove(&uuid).is_some()) + .send(operations.remove(&uuid).is_some()) .is_err() { return Err(stypes::NativeError::channel( @@ -235,40 +180,33 @@ pub async fn run( ); } tx_response - .send( - if let Some((_tx_sde, operation_cancalation_token, done_token)) = - tracker.operations.remove(&uuid) - { - if !done_token.is_cancelled() { - operation_cancalation_token.cancel(); - debug!("Waiting for operation {uuid} would confirm done-state"); - done_token.cancelled().await; - progress.stopped(&uuid); - } - if let Err(err) = state.canceled_operation(uuid).await { - error!( - "Failed to notify state about canceled operation {uuid}; err: {err:?}" - ); - } - true - } else { - false - }, - ) + .send(if let Some(operation) = operations.remove(&uuid) { + if !operation.done.is_cancelled() { + operation.canceler.cancel(); + debug!("Waiting for operation {uuid} would confirm done-state"); + operation.done.cancelled().await; + } + if let Err(err) = state.canceled_operation(uuid).await { + error!( + "Failed to notify state about canceled operation {uuid}; err: {err:?}" + ); + } + true + } else { + false + }) .map_err(|_| { stypes::NativeError::channel("Failed to respond to Api::CancelOperation") })?; } TrackerCommand::CancelAll(tx_response) => { - for (uuid, (_tx_sde, operation_cancalation_token, done_token)) in - &tracker.operations - { - if !done_token.is_cancelled() { - operation_cancalation_token.cancel(); + for (uuid, operation) in &operations { + if !operation.done.is_cancelled() { + operation.canceler.cancel(); debug!("waiting for operation {uuid} would confirm done-state"); if timeout( Duration::from_millis(CANCEL_OPERATION_TIMEOUT), - done_token.cancelled(), + operation.done.cancelled(), ) .await .is_err() @@ -278,50 +216,18 @@ pub async fn run( CANCEL_OPERATION_TIMEOUT / 1000 ); } - progress.stopped(uuid); } } - tracker.operations.clear(); + operations.clear(); if tx_response.send(()).is_err() { return Err(stypes::NativeError::channel( "fail to response to Api::CloseSession", )); } } - TrackerCommand::SetDebugMode((debug, tx_response)) => { - tracker.debug = debug; - if tx_response.send(()).is_err() { - return Err(stypes::NativeError::channel( - "fail to response to Api::SetDebugMode", - )); - } - } - TrackerCommand::GetOperationsStat(tx_response) => { - if tx_response - .send(match serde_json::to_string(&tracker.stat) { - Ok(serialized) => Ok(serialized), - Err(err) => Err(stypes::NativeError { - severity: stypes::Severity::ERROR, - kind: stypes::NativeErrorKind::ComputationFailed, - message: Some(format!("{err}")), - }), - }) - .is_err() - { - return Err(stypes::NativeError::channel( - "fail to response to Api::GetOperationsStat", - )); - } - } TrackerCommand::GetSdeSender((uuid, tx_response)) => { if tx_response - .send( - if let Some((tx_sde, _, _)) = tracker.operations.get(&uuid) { - tx_sde.clone() - } else { - None - }, - ) + .send(operations.get(&uuid).and_then(|op| op.sde_tx.clone())) .is_err() { return Err(stypes::NativeError::channel( @@ -333,14 +239,6 @@ pub async fn run( debug!("shutdown has been requested"); break; } - TrackerCommand::ShutdownWithError => { - debug!("shutdown tracker loop with error for testing"); - return Err(stypes::NativeError { - severity: stypes::Severity::ERROR, - kind: stypes::NativeErrorKind::Io, - message: Some(String::from("Shutdown tracker loop with error for testing")), - }); - } } } debug!("task is finished"); diff --git a/crates/core/session/src/unbound/api.rs b/crates/core/session/src/unbound/api.rs deleted file mode 100644 index ede962df42..0000000000 --- a/crates/core/session/src/unbound/api.rs +++ /dev/null @@ -1,320 +0,0 @@ -//! Definitions for Operations that forms the public API that will be called -//! from Electron front-end. - -use processor::search::filter::SearchFilter; -use serde::{Serialize, de::DeserializeOwned}; -use tokio::sync::{mpsc::UnboundedSender, oneshot}; - -use super::commands::Command; - -#[derive(Debug)] -pub enum API { - Shutdown(oneshot::Sender<()>), - CancelJob(u64), - Run(Command, u64), - /// remove finished jobs from registry - Remove(u64), -} - -#[derive(Clone, Debug)] -pub struct UnboundSessionAPI { - tx: UnboundedSender, -} - -impl UnboundSessionAPI { - pub fn new(tx: UnboundedSender) -> Self { - Self { tx } - } - - pub async fn shutdown(&self) -> Result<(), stypes::ComputationError> { - let (tx, rx): (oneshot::Sender<()>, oneshot::Receiver<()>) = oneshot::channel(); - self.tx.send(API::Shutdown(tx)).map_err(|_| { - stypes::ComputationError::Communication(String::from("Fail to send API::Shutdown")) - })?; - rx.await.map_err(|e| { - stypes::ComputationError::Communication(format!( - "Fail to get response from API::Shutdown: {e:?}" - )) - }) - } - - pub async fn cancel_job(&self, operation_id: &u64) -> Result<(), stypes::ComputationError> { - self.tx.send(API::CancelJob(*operation_id)).map_err(|_| { - stypes::ComputationError::Communication(String::from("Fail to send API::CancelJob")) - }) - } - - async fn process_command( - &self, - id: u64, - rx_results: oneshot::Receiver, stypes::ComputationError>>, - command: Command, - ) -> Result, stypes::ComputationError> { - let cmd = command.to_string(); - self.tx.send(API::Run(command, id)).map_err(|_| { - stypes::ComputationError::Communication(format!("Fail to send call {cmd}")) - })?; - rx_results - .await - .map_err(|e| stypes::ComputationError::Communication(format!("channel error: {e}")))? - } - - pub(crate) fn remove_command(&self, id: u64) -> Result<(), stypes::ComputationError> { - self.tx.send(API::Remove(id)).map_err(|_| { - stypes::ComputationError::Communication(format!("Fail to remove command id={id}")) - })?; - Ok(()) - } - - pub async fn cancel_test( - &self, - id: u64, - custom_arg_a: i64, - custom_arg_b: i64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::CancelTest(custom_arg_a, custom_arg_b, tx_results), - ) - .await - } - - pub async fn list_folder_content( - &self, - id: u64, - depth: usize, - max_len: usize, - paths: Vec, - include_files: bool, - include_folders: bool, - ) -> Result, stypes::ComputationError> - { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::FolderContent( - paths, - depth, - max_len, - include_files, - include_folders, - tx_results, - ), - ) - .await - } - - pub async fn is_file_binary( - &self, - id: u64, - file_path: String, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::IsFileBinary(file_path, tx_results)) - .await - } - - pub async fn get_file_checksum( - &self, - id: u64, - path: String, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::Checksum(path, tx_results)) - .await - } - - pub async fn get_dlt_stats( - &self, - id: u64, - files: Vec, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::GetDltStats(files, tx_results)) - .await - } - - pub async fn get_someip_statistic( - &self, - id: u64, - files: Vec, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::GetSomeipStatistic(files, tx_results), - ) - .await - } - - pub async fn get_shell_profiles( - &self, - id: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::GetShellProfiles(tx_results)) - .await - } - - pub async fn get_serial_ports_list( - &self, - id: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::SerialPortsList(tx_results)) - .await - } - - pub async fn get_regex_error( - &self, - id: u64, - filter: SearchFilter, - ) -> Result>, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::GetRegexError(filter, tx_results)) - .await - } - - pub async fn sleep( - &self, - id: u64, - ms: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::Sleep(ms, tx_results)) - .await - } - - /// Get all information of installed plugins . - pub async fn installed_plugins_list( - &self, - id: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::InstalledPluginsList(tx_results)) - .await - } - - /// Get all information of invalid plugins . - pub async fn invalid_plugins_list( - &self, - id: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::InvalidPluginsList(tx_results)) - .await - } - - /// Get the directory paths (considered ID) for installed plugins. - pub async fn installed_plugins_paths( - &self, - id: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::InstalledPluginsPaths(tx_results)) - .await - } - - /// Get the directory paths (considered ID) for invalid plugins. - pub async fn invalid_plugins_paths( - &self, - id: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::InvalidPluginsPaths(tx_results)) - .await - } - - /// Get all info for the installed plugin with provided directory path (considered ID) - pub async fn installed_plugin_info( - &self, - id: u64, - plugin_path: String, - ) -> Result>, stypes::ComputationError> - { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::InstalledPluginInfo(plugin_path, tx_results), - ) - .await - } - - /// Get all info for the invalid plugin with provided directory path (considered ID) - pub async fn invalid_plugin_info( - &self, - id: u64, - plugin_path: String, - ) -> Result>, stypes::ComputationError> - { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::InvalidPluginInfo(plugin_path, tx_results), - ) - .await - } - - /// Retrieves runtime data for a plugin located at the specified path. - pub async fn get_plugin_run_data( - &self, - id: u64, - plugin_path: String, - ) -> Result>, stypes::ComputationError> - { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::PluginRunData(plugin_path, tx_results), - ) - .await - } - - /// Reload the plugin directory. - pub async fn reload_plugins( - &self, - id: u64, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command(id, rx_results, Command::ReloadPlugins(tx_results)) - .await - } - - /// Adds a plugin with the given directory path and optional plugin type. - pub async fn add_plugin( - &self, - id: u64, - plugin_path: String, - plugin_type: Option, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::AddPlugin(plugin_path, plugin_type, tx_results), - ) - .await - } - - /// Removes the plugin with the given directory path. - pub async fn remove_plugin( - &self, - id: u64, - plugin_path: String, - ) -> Result, stypes::ComputationError> { - let (tx_results, rx_results) = oneshot::channel(); - self.process_command( - id, - rx_results, - Command::RemovePlugin(plugin_path, tx_results), - ) - .await - } -} diff --git a/crates/core/session/src/unbound/commands/cancel_test.rs b/crates/core/session/src/unbound/commands/cancel_test.rs deleted file mode 100644 index 49706d8094..0000000000 --- a/crates/core/session/src/unbound/commands/cancel_test.rs +++ /dev/null @@ -1,20 +0,0 @@ -use crate::unbound::signal::Signal; -use tokio::{ - select, - time::{Duration, sleep}, -}; - -pub async fn cancel_test( - custom_arg_a: i64, - custom_arg_b: i64, - signal: Signal, -) -> Result, stypes::ComputationError> { - Ok(select! { - _ = signal.cancelled() => { - stypes::CommandOutcome::Cancelled - } - _ = sleep(Duration::from_millis(500)) => { - stypes::CommandOutcome::Finished(custom_arg_a + custom_arg_b) - } - }) -} diff --git a/crates/core/session/src/unbound/commands/checksum.rs b/crates/core/session/src/unbound/commands/checksum.rs deleted file mode 100644 index a5d6a495fb..0000000000 --- a/crates/core/session/src/unbound/commands/checksum.rs +++ /dev/null @@ -1,29 +0,0 @@ -use crate::unbound::signal::Signal; -use blake3; -use std::{ - fs::File, - io::{self, prelude::*}, -}; - -pub fn checksum( - filename: &str, - _signal: Signal, -) -> Result, stypes::ComputationError> { - let mut file = - File::open(filename).map_err(|e| stypes::ComputationError::IoOperation(e.to_string()))?; - let mut hasher = blake3::Hasher::new(); - let mut buffer = [0; 65536]; - loop { - match file.read(&mut buffer) { - Ok(0) => break, - Ok(n) => { - hasher.update(&buffer[..n]); - } - Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue, - Err(e) => return Err(stypes::ComputationError::IoOperation(e.to_string())), - } - } - Ok(stypes::CommandOutcome::Finished( - hasher.finalize().to_string(), - )) -} diff --git a/crates/core/session/src/unbound/commands/dlt.rs b/crates/core/session/src/unbound/commands/dlt.rs deleted file mode 100644 index 192fc46d48..0000000000 --- a/crates/core/session/src/unbound/commands/dlt.rs +++ /dev/null @@ -1,44 +0,0 @@ -use crate::unbound::signal::Signal; -use dlt_core::{ - read::DltMessageReader, - statistics::{ - collect_statistics, - common::{StatisticInfo, StatisticInfoCollector}, - }, -}; -use std::fs::File; - -pub fn stats( - file_paths: Vec, - _signal: Signal, -) -> Result, stypes::ComputationError> { - let mut stat = StatisticInfo::new(); - let mut error: Option = None; - file_paths.iter().for_each(|file_path| { - if error.is_some() { - return; - } - match File::open(file_path) { - Ok(file) => { - let mut reader = DltMessageReader::new(file, true); - let mut collector = StatisticInfoCollector::default(); - - match collect_statistics(&mut reader, &mut collector) { - Ok(()) => { - stat.merge(collector.collect()); - } - Err(err) => { - error = Some(err.to_string()); - } - } - } - Err(err) => { - error = Some(err.to_string()); - } - } - }); - if let Some(err) = error { - return Err(stypes::ComputationError::IoOperation(err)); - } - Ok(stypes::CommandOutcome::Finished(stat.into())) -} diff --git a/crates/core/session/src/unbound/commands/file.rs b/crates/core/session/src/unbound/commands/file.rs deleted file mode 100644 index dddc205421..0000000000 --- a/crates/core/session/src/unbound/commands/file.rs +++ /dev/null @@ -1,9 +0,0 @@ -use file_tools::is_binary; - -pub fn is_file_binary( - file_path: String, -) -> Result, stypes::ComputationError> { - is_binary(file_path) - .map(stypes::CommandOutcome::Finished) - .map_err(|err| stypes::ComputationError::OperationNotSupported(err.to_string())) -} diff --git a/crates/core/session/src/unbound/commands/folder.rs b/crates/core/session/src/unbound/commands/folder.rs deleted file mode 100644 index 27341e86f6..0000000000 --- a/crates/core/session/src/unbound/commands/folder.rs +++ /dev/null @@ -1,72 +0,0 @@ -use crate::unbound::signal::Signal; -use walkdir::{DirEntry, WalkDir}; - -/// Find all files and/or folders in a folder -/// We first consider all elements on the same level before -/// descending into the next level. Kind of what you would get with BFS but -/// since the library we use only does DFS, we go level by level. -/// -/// paths should be a list of folders that will be searched -/// max_len is the maximum number of items after which we will stop the search -/// max_depth is the maximum folder level we will descend into to find items -/// signal used to cancel the operation -/// include_files wether to include files -/// include_folders if false folders will not be included in the result list -pub fn get_folder_content( - paths: &[String], - max_depth: usize, - max_len: usize, - include_files: bool, - include_folders: bool, - signal: Signal, -) -> Result, stypes::ComputationError> { - let mut list: Vec = vec![]; - let mut max_len_reached: bool = false; - for depth in 1..=max_depth { - if max_len_reached { - break; - } - for path in paths { - if max_len_reached { - break; - } - for dir_entry in WalkDir::new(path) - .min_depth(depth) - .max_depth(depth) - .into_iter() - .filter_map(|v| v.ok()) - .filter(|e| check_file_or_folder(e, include_files, include_folders)) - { - if signal.is_cancelling() { - return Ok(stypes::CommandOutcome::Cancelled); - } - if let Some(entity) = if let Ok(md) = dir_entry.metadata() { - stypes::FolderEntity::from(&dir_entry, &md) - } else { - None - } { - list.push(entity) - } - if list.len() >= max_len { - max_len_reached = true; - break; - } - } - } - } - Ok(stypes::CommandOutcome::Finished( - stypes::FoldersScanningResult { - list, - max_len_reached, - }, - )) -} - -fn check_file_or_folder(e: &DirEntry, include_files: bool, include_folders: bool) -> bool { - match (include_files, include_folders) { - (true, true) => true, - (true, false) => e.file_type().is_file(), - (false, true) => e.file_type().is_dir(), - _ => false, - } -} diff --git a/crates/core/session/src/unbound/commands/mod.rs b/crates/core/session/src/unbound/commands/mod.rs deleted file mode 100644 index f0b6b5875a..0000000000 --- a/crates/core/session/src/unbound/commands/mod.rs +++ /dev/null @@ -1,261 +0,0 @@ -//! Definitions of all commands related to unbound (general) session. - -mod cancel_test; -mod checksum; -mod dlt; -mod file; -mod folder; -pub mod plugins; -mod regex; -mod serial; -mod shells; -mod sleep; -mod someip; - -use crate::unbound::commands::someip::get_someip_statistic; -use plugins_host::plugins_manager::PluginsManager; -use tokio::sync::{RwLock, oneshot}; - -use super::signal::Signal; -use log::{debug, error}; -use processor::search::filter::SearchFilter; - -#[derive(Debug)] -pub enum Command { - // This command is used only for testing/debug goals - Sleep( - u64, - oneshot::Sender, stypes::ComputationError>>, - ), - FolderContent( - Vec, - usize, - usize, - bool, - bool, - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - GetRegexError( - SearchFilter, - oneshot::Sender>, stypes::ComputationError>>, - ), - Checksum( - String, - oneshot::Sender, stypes::ComputationError>>, - ), - GetDltStats( - Vec, - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - GetSomeipStatistic( - Vec, - oneshot::Sender, stypes::ComputationError>>, - ), - GetShellProfiles( - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - SerialPortsList( - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - IsFileBinary( - String, - oneshot::Sender, stypes::ComputationError>>, - ), - CancelTest( - i64, - i64, - oneshot::Sender, stypes::ComputationError>>, - ), - /// Get all information of the installed plugins . - InstalledPluginsList( - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - /// Get all information of invalid plugins . - InvalidPluginsList( - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - /// Get the directory paths (considered ID) for installed plugins. - InstalledPluginsPaths( - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - /// Get the directory paths (considered ID) for invalid plugins. - InvalidPluginsPaths( - oneshot::Sender< - Result, stypes::ComputationError>, - >, - ), - /// Get all info for the installed plugin with provided directory path (considered ID) - InstalledPluginInfo( - String, - oneshot::Sender< - Result>, stypes::ComputationError>, - >, - ), - /// Get all info for the invalid plugin with provided directory path (considered ID) - InvalidPluginInfo( - String, - oneshot::Sender< - Result< - stypes::CommandOutcome>, - stypes::ComputationError, - >, - >, - ), - /// Retrieves runtime data for a plugin located at the specified path. - PluginRunData( - String, - oneshot::Sender< - Result>, stypes::ComputationError>, - >, - ), - /// Reload all the plugins from their directory. - ReloadPlugins(oneshot::Sender, stypes::ComputationError>>), - /// Adds a plugin with the given directory path and optional plugin type. - AddPlugin( - String, - Option, - oneshot::Sender, stypes::ComputationError>>, - ), - /// Removes the plugin with the given directory path. - RemovePlugin( - String, - oneshot::Sender, stypes::ComputationError>>, - ), -} - -impl std::fmt::Display for Command { - fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { - write!( - f, - "{}", - match self { - Command::Sleep(_, _) => "Sleep", - Command::CancelTest(_, _, _) => "CancelTest", - Command::FolderContent(_, _, _, _, _, _) => "Getting folder's content", - Command::GetShellProfiles(_) => "Getting shell profiles", - Command::SerialPortsList(_) => "Getting serial ports list", - Command::Checksum(_, _) => "Calculating file's checksum", - Command::GetDltStats(_, _) => "Getting dlt stats", - Command::GetSomeipStatistic(_, _) => "Getting someip statistic", - Command::GetRegexError(_, _) => "Checking regex", - Command::IsFileBinary(_, _) => "Checking if file is binary", - Command::InstalledPluginsList(..) => "Getting installed plugins", - Command::InvalidPluginsList(..) => "Getting invalid plugins", - Command::InstalledPluginsPaths(..) => "Getting installed plugins paths", - Command::InvalidPluginsPaths(..) => "Getting invaild plugins paths", - Command::InstalledPluginInfo(..) => "Getting installed plugin info", - Command::InvalidPluginInfo(..) => "Getting invalid plugin info", - Command::PluginRunData(..) => "Getting plugin run data", - Command::ReloadPlugins(..) => "Reloading plugins' information", - Command::AddPlugin(..) => "Adding plugin", - Command::RemovePlugin(..) => "Removing plugin", - } - ) - } -} - -pub async fn process(command: Command, signal: Signal, plugins_manager: &RwLock) { - let cmd = command.to_string(); - debug!("Processing command: {cmd}"); - if match command { - Command::Sleep(ms, tx) => tx.send(sleep::sleep(ms, signal).await).is_err(), - Command::FolderContent(paths, depth, max_len, include_files, include_folders, tx) => tx - .send(folder::get_folder_content( - &paths, - depth, - max_len, - include_files, - include_folders, - signal, - )) - .is_err(), - Command::GetRegexError(filter, tx) => { - tx.send(regex::get_filter_error(filter, signal)).is_err() - } - Command::Checksum(file, tx) => tx.send(checksum::checksum(&file, signal)).is_err(), - Command::GetDltStats(files, tx) => tx.send(dlt::stats(files, signal)).is_err(), - Command::GetSomeipStatistic(files, tx) => { - tx.send(get_someip_statistic(files, signal)).is_err() - } - Command::GetShellProfiles(tx) => tx.send(shells::get_available_shells(signal)).is_err(), - Command::SerialPortsList(tx) => tx.send(serial::available_ports(signal)).is_err(), - Command::IsFileBinary(file_path, tx) => tx.send(file::is_file_binary(file_path)).is_err(), - Command::CancelTest(a, b, tx) => tx - .send(cancel_test::cancel_test(a, b, signal).await) - .is_err(), - Command::InstalledPluginsList(tx) => tx - .send(plugins::installed_plugins_list(plugins_manager, signal).await) - .is_err(), - Command::InvalidPluginsList(tx) => tx - .send(plugins::invalid_plugins_list(plugins_manager, signal).await) - .is_err(), - Command::InstalledPluginsPaths(tx) => tx - .send(plugins::installed_plugins_paths(plugins_manager, signal).await) - .is_err(), - Command::InvalidPluginsPaths(tx) => tx - .send(plugins::invalid_plugins_paths(plugins_manager, signal).await) - .is_err(), - Command::InstalledPluginInfo(path, tx) => tx - .send(plugins::installed_plugins_info(path, plugins_manager, signal).await) - .is_err(), - Command::InvalidPluginInfo(path, tx) => tx - .send(plugins::invalid_plugins_info(path, plugins_manager, signal).await) - .is_err(), - Command::PluginRunData(path, tx) => tx - .send(plugins::get_plugin_run_data(path, plugins_manager, signal).await) - .is_err(), - Command::ReloadPlugins(tx) => tx - .send(plugins::reload_plugins(plugins_manager, signal).await) - .is_err(), - Command::AddPlugin(path, typ, tx) => tx - .send(plugins::add_plugin(path, typ, plugins_manager, signal).await) - .is_err(), - Command::RemovePlugin(path, tx) => tx - .send(plugins::remove_plugin(path, plugins_manager, signal).await) - .is_err(), - } { - error!("Fail to send response for command: {cmd}"); - } -} - -pub fn err(command: Command, err: stypes::ComputationError) { - let cmd = command.to_string(); - if match command { - Command::Sleep(_, tx) => tx.send(Err(err)).is_err(), - Command::FolderContent(_path, _depth, _max_len, _, _, tx) => tx.send(Err(err)).is_err(), - Command::GetRegexError(_filter, tx) => tx.send(Err(err)).is_err(), - Command::Checksum(_file, tx) => tx.send(Err(err)).is_err(), - Command::GetDltStats(_files, tx) => tx.send(Err(err)).is_err(), - Command::GetSomeipStatistic(_files, tx) => tx.send(Err(err)).is_err(), - Command::GetShellProfiles(tx) => tx.send(Err(err)).is_err(), - Command::SerialPortsList(tx) => tx.send(Err(err)).is_err(), - Command::IsFileBinary(_filepath, tx) => tx.send(Err(err)).is_err(), - Command::CancelTest(_a, _b, tx) => tx.send(Err(err)).is_err(), - Command::InstalledPluginsList(tx) => tx.send(Err(err)).is_err(), - Command::InvalidPluginsList(tx) => tx.send(Err(err)).is_err(), - Command::InstalledPluginsPaths(tx) => tx.send(Err(err)).is_err(), - Command::InvalidPluginsPaths(tx) => tx.send(Err(err)).is_err(), - Command::InstalledPluginInfo(_, tx) => tx.send(Err(err)).is_err(), - Command::InvalidPluginInfo(_, tx) => tx.send(Err(err)).is_err(), - Command::PluginRunData(_, tx) => tx.send(Err(err)).is_err(), - Command::ReloadPlugins(tx) => tx.send(Err(err)).is_err(), - Command::AddPlugin(_, _, tx) => tx.send(Err(err)).is_err(), - Command::RemovePlugin(_, tx) => tx.send(Err(err)).is_err(), - } { - error!("Fail to send error response for command: {cmd}"); - } -} diff --git a/crates/core/session/src/unbound/commands/plugins.rs b/crates/core/session/src/unbound/commands/plugins.rs deleted file mode 100644 index 9ff8a896d5..0000000000 --- a/crates/core/session/src/unbound/commands/plugins.rs +++ /dev/null @@ -1,187 +0,0 @@ -use std::path::PathBuf; - -use crate::unbound::signal::Signal; -use plugins_host::plugins_manager::PluginsManager; -use stypes::{ - CommandOutcome, ComputationError, InvalidPluginEntity, InvalidPluginsList, PluginEntity, - PluginRunData, PluginType, PluginsList, PluginsPathsList, -}; -use tokio::sync::RwLock; - -/// Initialize the plugin manager loading all the plugins from their directory. -pub async fn load_manager() -> Result { - PluginsManager::load() - .await - .map_err(|err| ComputationError::NativeError(err.into())) -} - -/// Get all information of installed plugins . -pub async fn installed_plugins_list( - plugins_manager: &RwLock, - _signal: Signal, -) -> Result, ComputationError> { - let manager = plugins_manager.read().await; - - let installed_plugins = manager.installed_plugins().cloned().collect(); - - let plugins = PluginsList(installed_plugins); - - Ok(CommandOutcome::Finished(plugins)) -} - -/// Get all information of invalid plugins . -pub async fn invalid_plugins_list( - plugins_manager: &RwLock, - _signal: Signal, -) -> Result, ComputationError> { - let manager = plugins_manager.read().await; - - let invalid_plugins = manager.invalid_plugins().cloned().collect(); - - let plugins = InvalidPluginsList(invalid_plugins); - - Ok(CommandOutcome::Finished(plugins)) -} - -/// Get the directory paths (considered ID) of installed plugins . -pub async fn installed_plugins_paths( - plugins_manager: &RwLock, - _signal: Signal, -) -> Result, ComputationError> { - let manager = plugins_manager.read().await; - - let installed_paths: Vec<_> = manager - .installed_plugins_paths() - .map(|p| p.to_string_lossy().to_string()) - .collect(); - - let plugins = PluginsPathsList(installed_paths); - - Ok(CommandOutcome::Finished(plugins)) -} - -/// Get the directory paths (considered ID) of invalid plugins . -pub async fn invalid_plugins_paths( - plugins_manager: &RwLock, - _signal: Signal, -) -> Result, ComputationError> { - let manager = plugins_manager.read().await; - - let invalid_paths: Vec<_> = manager - .invalid_plugins_paths() - .map(|p| p.to_string_lossy().to_string()) - .collect(); - - let plugins = PluginsPathsList(invalid_paths); - - Ok(CommandOutcome::Finished(plugins)) -} - -/// Get all info for the installed plugin with provided directory path (considered ID) -pub async fn installed_plugins_info( - plugin_path: String, - plugins_manager: &RwLock, - _signal: Signal, -) -> Result>, ComputationError> { - let manager = plugins_manager.read().await; - - let plugin = manager - .get_installed_plugin(&PathBuf::from(plugin_path)) - .cloned(); - - Ok(CommandOutcome::Finished(plugin)) -} - -/// Get all info for the invalid plugin with provided directory path (considered ID) -pub async fn invalid_plugins_info( - plugin_path: String, - plugins_manager: &RwLock, - _signal: Signal, -) -> Result>, ComputationError> { - let manager = plugins_manager.read().await; - - let invalid_plug = manager - .get_invalid_plugin(&PathBuf::from(plugin_path)) - .cloned(); - - Ok(CommandOutcome::Finished(invalid_plug)) -} - -/// Retrieves runtime data for a plugin located at the specified path. -/// -/// This method searches for the plugin's runtime data (`PluginRunData`) among both -/// successfully loaded plugins and failed ones. -/// -/// # Parameters -/// - `plugin_path`: The directory path of the plugin. -/// -/// # Returns -/// - `Some(&PluginRunData)`: If the plugin's runtime data is found. -/// - `None`: If no matching plugin is found. -pub async fn get_plugin_run_data( - plugin_path: String, - plugins_manager: &RwLock, - _signal: Signal, -) -> Result>, ComputationError> { - let manager = plugins_manager.read().await; - - let invalid_plug = manager - .get_plugin_run_data(PathBuf::from(plugin_path)) - .cloned(); - - Ok(CommandOutcome::Finished(invalid_plug)) -} - -/// Reload plugins from the plugins directory. -pub async fn reload_plugins( - plugins_manager: &RwLock, - _signal: Signal, -) -> Result, ComputationError> { - let mut manager = plugins_manager.write().await; - - manager - .reload() - .await - .map_err(|err| ComputationError::NativeError(err.into()))?; - - Ok(CommandOutcome::Finished(())) -} - -/// Adds a plugin with the given directory path and the optional plugin type. -/// -/// * `plugin_path`: Path of the plugin directory to be copied into chipmunk plugins directory. -/// * `plugin_type`: Type of the plugin, when not provided plugin type will be entered from plugin -/// `WIT` signature in its binary file. -pub async fn add_plugin( - plugin_path: String, - plugin_type: Option, - plugins_manager: &RwLock, - _signal: Signal, -) -> Result, ComputationError> { - let mut manager = plugins_manager.write().await; - - manager - .add_plugin(plugin_path.into(), plugin_type) - .await - .map_err(|err| ComputationError::NativeError(err.into()))?; - - Ok(CommandOutcome::Finished(())) -} - -/// Removes the plugin with the given directory path. -/// -/// * `plugin_path`: Path of the plugin Chipmunk plugins directory. -pub async fn remove_plugin( - plugin_path: String, - plugins_manager: &RwLock, - _signal: Signal, -) -> Result, ComputationError> { - let mut manager = plugins_manager.write().await; - - manager - .remove_plugin(PathBuf::from(plugin_path).as_path()) - .await - .map_err(|err| ComputationError::NativeError(err.into()))?; - - Ok(CommandOutcome::Finished(())) -} diff --git a/crates/core/session/src/unbound/commands/regex.rs b/crates/core/session/src/unbound/commands/regex.rs deleted file mode 100644 index bbefd3dc02..0000000000 --- a/crates/core/session/src/unbound/commands/regex.rs +++ /dev/null @@ -1,9 +0,0 @@ -use crate::unbound::signal::Signal; -use processor::search::filter::{SearchFilter, get_filter_error as validator}; - -pub fn get_filter_error( - filter: SearchFilter, - _signal: Signal, -) -> Result>, stypes::ComputationError> { - Ok(stypes::CommandOutcome::Finished(validator(&filter))) -} diff --git a/crates/core/session/src/unbound/commands/serial.rs b/crates/core/session/src/unbound/commands/serial.rs deleted file mode 100644 index ac55909d90..0000000000 --- a/crates/core/session/src/unbound/commands/serial.rs +++ /dev/null @@ -1,16 +0,0 @@ -use crate::unbound::signal::Signal; - -pub fn available_ports( - _signal: Signal, -) -> Result, stypes::ComputationError> { - serialport::available_ports() - .map_err(|e| stypes::ComputationError::IoOperation(e.to_string())) - .map(|ports| { - stypes::CommandOutcome::Finished(stypes::SerialPortsList( - ports - .into_iter() - .map(|p| p.port_name) - .collect::>(), - )) - }) -} diff --git a/crates/core/session/src/unbound/commands/shells.rs b/crates/core/session/src/unbound/commands/shells.rs deleted file mode 100644 index a1eb94aaf4..0000000000 --- a/crates/core/session/src/unbound/commands/shells.rs +++ /dev/null @@ -1,10 +0,0 @@ -use crate::unbound::signal::Signal; - -pub fn get_available_shells( - _signal: Signal, -) -> Result, stypes::ComputationError> { - let shells = shell_tools::get_available_shells(); - Ok(stypes::CommandOutcome::Finished(stypes::ProfileList( - shells.to_vec(), - ))) -} diff --git a/crates/core/session/src/unbound/commands/sleep.rs b/crates/core/session/src/unbound/commands/sleep.rs deleted file mode 100644 index 1002648696..0000000000 --- a/crates/core/session/src/unbound/commands/sleep.rs +++ /dev/null @@ -1,12 +0,0 @@ -use crate::unbound::signal::Signal; -use tokio::time; - -// This command is used for testing/debug goals only. It should ignore signal to ignore -// cancellation. -pub async fn sleep( - ms: u64, - _signal: Signal, -) -> Result, stypes::ComputationError> { - let _ = time::sleep(time::Duration::from_millis(ms)).await; - Ok(stypes::CommandOutcome::Finished(())) -} diff --git a/crates/core/session/src/unbound/commands/someip.rs b/crates/core/session/src/unbound/commands/someip.rs deleted file mode 100644 index bffc3f7c6c..0000000000 --- a/crates/core/session/src/unbound/commands/someip.rs +++ /dev/null @@ -1,45 +0,0 @@ -use crate::unbound::signal::Signal; - -pub fn get_someip_statistic( - _files: Vec, - _signal: Signal, -) -> Result, stypes::ComputationError> { - Err(stypes::ComputationError::OperationNotSupported( - "NYI".into(), - )) - // use parsers::someip::{read_someip_statistic_from_pcapng, SomeipStatistic}; - // use log::{error, warn}; - // use std::path::Path; - - // let mut statistic = SomeipStatistic::new(); - // let mut error: Option = None; - // warn!("Getting statistic for: {files:?}"); - // files.iter().for_each(|file| { - // if error.is_some() { - // return; - // } - // if signal.is_cancelling() { - // return; - // } - // match read_someip_statistic_from_pcapng(Path::new(&file), &signal.token()) { - // Ok(result) => { - // statistic.merge(result); - // } - // Err(err) => { - // error = Some(err.to_string()); - // } - // } - // }); - // if let Some(err) = error { - // error!("Fail to get statistic for: {files:?}"); - // return Err(stypes::ComputationError::IoOperation(err)); - // } - // if signal.is_cancelling() { - // warn!("Operation of geting statistic for: {files:?} has been cancelled"); - // return Ok(stypes::CommandOutcome::Cancelled); - // } - // Ok(stypes::CommandOutcome::Finished( - // serde_json::to_string(&statistic) - // .map_err(|e| stypes::ComputationError::IoOperation(e.to_string()))?, - // )) -} diff --git a/crates/core/session/src/unbound/mod.rs b/crates/core/session/src/unbound/mod.rs deleted file mode 100644 index 4eec6bc3ca..0000000000 --- a/crates/core/session/src/unbound/mod.rs +++ /dev/null @@ -1,177 +0,0 @@ -//! Contains implementation for general functionalities in Chipmunk that are not related to logs -//! processing and analyzing sessions. -//! -//! It includes starting up of chipmunk with its needed components, alongside with the definitions -//! of the available parsers and sources, and plugins management. - -pub mod api; -mod cleanup; -pub mod commands; -mod signal; - -use crate::{ - progress::ProgressProviderAPI, - unbound::{ - api::{API, UnboundSessionAPI}, - signal::Signal, - }, -}; -pub use cleanup::cleanup_temp_files; -use log::{debug, error, warn}; -use std::{collections::HashMap, sync::Arc}; -use tokio::{ - sync::{ - RwLock, - mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, - }, - time::{Duration, timeout}, -}; -use tokio_util::sync::CancellationToken; -use uuid::Uuid; - -pub const CANCEL_OPERATIONS_TIMEOUT: u64 = 2000; - -pub struct UnboundSession { - rx: Option>, - pub finished: CancellationToken, - session_api: UnboundSessionAPI, -} - -impl UnboundSession { - pub fn new() -> (Self, UnboundSessionAPI) { - let (tx, rx): (UnboundedSender, UnboundedReceiver) = unbounded_channel(); - let session_api = UnboundSessionAPI::new(tx); - ( - Self { - rx: Some(rx), - finished: CancellationToken::new(), - session_api: session_api.clone(), - }, - session_api, - ) - } - - pub async fn init(&mut self) -> Result<(), stypes::ComputationError> { - // TODO: Plugins manager is used temporally here in initial phase and we should consider - // moving it to its own module. Reasons: - // * It doesn't need parallelism for most of task. - // * It'll need different state and locking management for downloading plugins, Updating - // caches etc... - let plugins_manager = commands::plugins::load_manager().await?; - let plugins_manager = Arc::new(RwLock::new(plugins_manager)); - let finished = self.finished.clone(); - let mut rx = self - .rx - .take() - .ok_or(stypes::ComputationError::SessionUnavailable)?; // Error: session already running - let progress = ProgressProviderAPI::new()?; - let session_api = self.session_api.clone(); - tokio::spawn(async move { - let mut jobs: HashMap = HashMap::new(); - let mut uuids: HashMap = HashMap::new(); - while let Some(api) = rx.recv().await { - jobs.retain(|id, signal| { - let cancelled = signal.is_cancelled(); - if cancelled { - UnboundSession::stopped(&progress, &uuids, id); - } - !cancelled - }); - match api { - API::Run(job, id) => { - let signal = Signal::new(job.to_string()); - if jobs.contains_key(&id) { - commands::err( - job, - stypes::ComputationError::InvalidArgs(String::from( - "Job has invalid id. Id already exists.", - )), - ); - continue; - } - jobs.insert(id, signal.clone()); - UnboundSession::started(&progress, job.to_string(), &mut uuids, &id); - let api = session_api.clone(); - let plugs_ref_clone = Arc::clone(&plugins_manager); - tokio::spawn(async move { - debug!("Job {job} has been called"); - commands::process(job, signal.clone(), plugs_ref_clone.as_ref()).await; - signal.confirm(); - let _ = api.remove_command(id); - }); - } - API::CancelJob(id) => { - if let Some(signal) = jobs.get(&id) { - signal.invoke(); - debug!("Cancel signal has been sent to job {} ({id})", signal.alias); - } else { - warn!("Fail to cancel job; id {id} doesn't exist."); - } - } - API::Shutdown(tx) => { - jobs.iter().for_each(|(_uuid, signal)| { - signal.invoke(); - }); - match timeout(Duration::from_millis(CANCEL_OPERATIONS_TIMEOUT), async { - for (id, signal) in jobs.iter() { - signal.confirmed().await; - UnboundSession::stopped(&progress, &uuids, id); - } - }) - .await - { - Ok(_) => debug!("All jobs of unbound session are down"), - Err(_) => warn!( - "Unbound session wasn't shutdown normaly. Force shutdown because timeout {CANCEL_OPERATIONS_TIMEOUT}" - ), - } - jobs.clear(); - if tx.send(()).is_err() { - error!("Fail to send shutdown confirmation"); - } - break; - } - API::Remove(id) => { - if jobs.remove(&id).is_some() { - UnboundSession::stopped(&progress, &uuids, &id); - } - } - } - } - finished.cancel(); - debug!("Unbound session is down"); - }); - - // Call cleanup here because this function should be called once when chipmunk starts. - // Run cleaning up on a separate thread to avoid latency in startup in case temporary - // files are too large. - tokio::task::spawn_blocking(|| { - if let Err(errs) = cleanup_temp_files(None) { - errs.iter().for_each(|err| { - log::error!("Error while cleaning up temporary files. Error: {err:?}") - }); - } - }); - - Ok(()) - } - - fn started( - progress: &ProgressProviderAPI, - alias: String, - uuids: &mut HashMap, - id: &u64, - ) { - let uuid = Uuid::new_v4(); - uuids.insert(*id, uuid); - progress.started(&alias, &uuid); - } - - fn stopped(progress: &ProgressProviderAPI, uuids: &HashMap, id: &u64) { - if let Some(uuid) = uuids.get(id) { - progress.stopped(uuid); - } else { - error!("Fail to find UUID for operation id={id}"); - } - } -} diff --git a/crates/core/session/src/unbound/signal.rs b/crates/core/session/src/unbound/signal.rs deleted file mode 100644 index 56592cc694..0000000000 --- a/crates/core/session/src/unbound/signal.rs +++ /dev/null @@ -1,54 +0,0 @@ -use tokio_util::sync::CancellationToken; - -#[derive(Clone, Debug)] -pub struct Signal { - pub alias: String, - signal: CancellationToken, - confirmation: CancellationToken, -} - -impl Signal { - pub fn new(alias: String) -> Signal { - Signal { - alias, - signal: CancellationToken::new(), - confirmation: CancellationToken::new(), - } - } - - pub fn token(&self) -> CancellationToken { - self.signal.clone() - } - - pub fn invoke(&self) { - if self.confirmation.is_cancelled() { - return; - } - if !self.signal.is_cancelled() { - self.signal.cancel(); - } - } - - pub async fn cancelled(&self) { - self.signal.cancelled().await - } - - pub async fn confirmed(&self) { - self.confirmation.cancelled().await - } - - pub fn is_cancelled(&self) -> bool { - self.confirmation.is_cancelled() - } - - pub fn is_cancelling(&self) -> bool { - self.signal.is_cancelled() - } - - pub fn confirm(&self) { - if self.confirmation.is_cancelled() { - return; - } - self.confirmation.cancel(); - } -} diff --git a/crates/core/session/tests/snapshot_tests/utls.rs b/crates/core/session/tests/snapshot_tests/utls.rs index ed0539f8be..10ab9be543 100644 --- a/crates/core/session/tests/snapshot_tests/utls.rs +++ b/crates/core/session/tests/snapshot_tests/utls.rs @@ -113,8 +113,7 @@ pub async fn run_observe_session>( while let Some(feedback) = receiver.recv().await { match feedback { - stypes::CallbackEvent::FileRead | stypes::CallbackEvent::SessionDestroyed => break, - stypes::CallbackEvent::SessionError(err) => panic!("Received session error: {err:#?}"), + stypes::CallbackEvent::FileRead => break, stypes::CallbackEvent::OperationError { error, .. } => { panic!("Received operation error: {error:#?}") } diff --git a/crates/file_tools/Cargo.toml b/crates/file_tools/Cargo.toml index a155699dc2..914e5ea405 100644 --- a/crates/file_tools/Cargo.toml +++ b/crates/file_tools/Cargo.toml @@ -7,7 +7,3 @@ edition.workspace = true workspace = true # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html - -[dependencies] -# External crates -anyhow.workspace = true diff --git a/crates/stypes/Cargo.toml b/crates/stypes/Cargo.toml index 4707ac2266..8980090139 100644 --- a/crates/stypes/Cargo.toml +++ b/crates/stypes/Cargo.toml @@ -10,9 +10,8 @@ workspace = true [dependencies] # External crates serde = { workspace = true, features = ["derive"] } -dlt-core = { workspace = true, features = ["fibex", "statistics", "serialization"] } +dlt-core = { workspace = true, features = ["fibex", "serialization"] } uuid = { workspace = true, features = ["serde", "v4"] } tokio.workspace = true thiserror.workspace = true -walkdir.workspace = true anyhow.workspace = true diff --git a/crates/stypes/src/callback/formating.rs b/crates/stypes/src/callback/formating.rs index c9140eef19..a910f295a6 100644 --- a/crates/stypes/src/callback/formating.rs +++ b/crates/stypes/src/callback/formating.rs @@ -14,13 +14,10 @@ impl std::fmt::Display for CallbackEvent { /// - `SearchMapUpdated` - Indicates that the search map has been updated. /// - `SearchValuesUpdated` - Indicates that search values have been updated. /// - `AttachmentsUpdated: {len}` - Displays the total count of the session attachments. - /// - `Progress` - Indicates progress for an operation. - /// - `SessionError: {err}` - Displays details of a session error. /// - `OperationError: {uuid}: {error}` - Displays the UUID of the operation and the error details. /// - `OperationStarted: {uuid}` - Displays the UUID of a started operation. /// - `OperationProcessing: {uuid}` - Displays the UUID of an operation in progress. /// - `OperationDone: {info.uuid}` - Displays the UUID of a completed operation. - /// - `SessionDestroyed` - Indicates that the session has been destroyed. fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { match self { Self::StreamUpdated(len) => write!(f, "StreamUpdated({len})"), @@ -35,18 +32,12 @@ impl std::fmt::Display for CallbackEvent { } => { write!(f, "AttachmentsUpdated: {len}") } - Self::Progress { - uuid: _, - progress: _, - } => write!(f, "Progress"), - Self::SessionError(err) => write!(f, "SessionError: {err}"), Self::OperationError { uuid, error } => { write!(f, "OperationError: {uuid}: {error}") } Self::OperationStarted(uuid) => write!(f, "OperationStarted: {uuid}"), Self::OperationProcessing(uuid) => write!(f, "OperationProcessing: {uuid}"), Self::OperationDone(info) => write!(f, "OperationDone: {}", info.uuid), - Self::SessionDestroyed => write!(f, "SessionDestroyed"), } } } diff --git a/crates/stypes/src/callback/mod.rs b/crates/stypes/src/callback/mod.rs index c78f817aff..820032d8ca 100644 --- a/crates/stypes/src/callback/mod.rs +++ b/crates/stypes/src/callback/mod.rs @@ -61,17 +61,6 @@ pub enum CallbackEvent { attachments: Vec, }, - /// Triggered when progress is made during an operation. - Progress { - /// The unique identifier of the operation. - uuid: Uuid, - /// Information about the progress. - progress: Progress, - }, - - /// Triggered in the event of an undefined session error. - SessionError(NativeError), - /// Triggered when an operation ends with an error. /// This event may follow `OperationStarted` since that event only indicates /// that the operation began successfully. It may also follow `OperationProcessing`. @@ -98,9 +87,4 @@ pub enum CallbackEvent { /// Triggered upon the successful completion of an operation. /// - `OperationDone`: The results of the completed operation. OperationDone(OperationDone), - - /// Triggered when the current session is fully closed, and all necessary cleanup - /// procedures are completed. This event guarantees that all possible read/write - /// operations are stopped, and all previously created loops are terminated. - SessionDestroyed, } diff --git a/crates/stypes/src/command/dltstat/converting.rs b/crates/stypes/src/command/dltstat/converting.rs deleted file mode 100644 index 382885f837..0000000000 --- a/crates/stypes/src/command/dltstat/converting.rs +++ /dev/null @@ -1,38 +0,0 @@ -use crate::*; -use dlt_core::statistics::common::{LevelDistribution, StatisticInfo}; - -impl From for DltLevelDistribution { - fn from(v: LevelDistribution) -> Self { - DltLevelDistribution { - non_log: v.non_log, - log_fatal: v.log_fatal, - log_error: v.log_error, - log_warning: v.log_warning, - log_info: v.log_info, - log_debug: v.log_debug, - log_verbose: v.log_verbose, - log_invalid: v.log_invalid, - } - } -} - -trait InnerInto { - fn inner_into(self) -> T; -} - -impl InnerInto> for Vec<(String, LevelDistribution)> { - fn inner_into(self) -> Vec<(String, DltLevelDistribution)> { - self.into_iter().map(|(k, l)| (k, l.into())).collect() - } -} - -impl From for DltStatisticInfo { - fn from(v: StatisticInfo) -> Self { - DltStatisticInfo { - app_ids: v.app_ids.inner_into(), - context_ids: v.context_ids.inner_into(), - ecu_ids: v.ecu_ids.inner_into(), - contained_non_verbose: v.contained_non_verbose, - } - } -} diff --git a/crates/stypes/src/command/dltstat/mod.rs b/crates/stypes/src/command/dltstat/mod.rs deleted file mode 100644 index 6e693d8941..0000000000 --- a/crates/stypes/src/command/dltstat/mod.rs +++ /dev/null @@ -1,23 +0,0 @@ -mod converting; - -use crate::*; - -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct DltLevelDistribution { - pub non_log: usize, - pub log_fatal: usize, - pub log_error: usize, - pub log_warning: usize, - pub log_info: usize, - pub log_debug: usize, - pub log_verbose: usize, - pub log_invalid: usize, -} - -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct DltStatisticInfo { - pub app_ids: Vec<(String, DltLevelDistribution)>, - pub context_ids: Vec<(String, DltLevelDistribution)>, - pub ecu_ids: Vec<(String, DltLevelDistribution)>, - pub contained_non_verbose: bool, -} diff --git a/crates/stypes/src/command/extending.rs b/crates/stypes/src/command/extending.rs deleted file mode 100644 index d0b4682259..0000000000 --- a/crates/stypes/src/command/extending.rs +++ /dev/null @@ -1,19 +0,0 @@ -use crate::*; - -impl CommandOutcome { - /// Converts a `CommandOutcome` into a `UuidCommandOutcome`, associating it with a given `Uuid`. - /// - /// # Parameters - /// - `self`: The `CommandOutcome` instance to be converted. - /// - `uuid`: The `Uuid` to associate with the resulting `UuidCommandOutcome`. - /// - /// # Returns - /// - `UuidCommandOutcome::Cancelled` if the `CommandOutcome` is `Cancelled`. - /// - `UuidCommandOutcome::Finished` if the `CommandOutcome` is `Finished`, pairing the given `Uuid` with the result. - pub fn as_command_result(self, uuid: Uuid) -> UuidCommandOutcome { - match self { - CommandOutcome::Cancelled => UuidCommandOutcome::Cancelled(uuid), - CommandOutcome::Finished(c) => UuidCommandOutcome::Finished((uuid, c)), - } - } -} diff --git a/crates/stypes/src/command/folders/extending.rs b/crates/stypes/src/command/folders/extending.rs deleted file mode 100644 index 89a977b4ad..0000000000 --- a/crates/stypes/src/command/folders/extending.rs +++ /dev/null @@ -1,100 +0,0 @@ -use crate::*; -use std::fs::Metadata; -use walkdir::DirEntry; - -impl FolderEntityDetails { - /// Creates a `FolderEntityDetails` instance from a directory entry. - /// - /// # Parameters - /// - `entity`: The `DirEntry` representing a file or folder. - /// - /// # Returns - /// - `Some(FolderEntityDetails)` if the parent directory can be determined. - /// - `None` otherwise. - pub fn from(entity: &DirEntry) -> Option { - entity.path().parent().map(|parent| FolderEntityDetails { - full: entity.path().to_string_lossy().to_string(), - filename: entity.file_name().to_string_lossy().to_string(), - path: parent.to_string_lossy().to_string(), - basename: entity.file_name().to_string_lossy().to_string(), - ext: entity - .path() - .extension() - .unwrap_or_default() - .to_string_lossy() - .to_string(), - }) - } -} - -impl FolderEntity { - /// Creates a `FolderEntity` instance from a directory entry and its metadata. - /// - /// # Parameters - /// - `entity`: The `DirEntry` representing a file or folder. - /// - `md`: The `Metadata` of the directory entry. - /// - /// # Returns - /// - `Some(FolderEntity)` if the entry is a directory, file, or symbolic link. - /// - `None` otherwise. - pub fn from(entity: &DirEntry, md: &Metadata) -> Option { - if md.is_dir() { - FolderEntity::dir(entity) - } else if md.is_symlink() { - FolderEntity::symlink(entity) - } else { - FolderEntity::file(entity) - } - } - - /// Creates a `FolderEntity` instance for a directory. - /// - /// # Parameters - /// - `entity`: The `DirEntry` representing the directory. - /// - /// # Returns - /// - `Some(FolderEntity)` if the directory has a valid file name. - /// - `None` otherwise. - fn dir(entity: &DirEntry) -> Option { - entity.path().file_name().map(|filename| FolderEntity { - name: filename.to_string_lossy().to_string(), - fullname: entity.path().to_string_lossy().to_string(), - kind: FolderEntityType::Directory, - details: None, - }) - } - - /// Creates a `FolderEntity` instance for a file. - /// - /// # Parameters - /// - `entity`: The `DirEntry` representing the file. - /// - /// # Returns - /// - `Some(FolderEntity)` if the file has a valid file name. - /// - `None` otherwise. - fn file(entity: &DirEntry) -> Option { - entity.path().file_name().map(|filename| FolderEntity { - name: filename.to_string_lossy().to_string(), - fullname: entity.path().to_string_lossy().to_string(), - kind: FolderEntityType::File, - details: FolderEntityDetails::from(entity), - }) - } - - /// Creates a `FolderEntity` instance for a symbolic link. - /// - /// # Parameters - /// - `entity`: The `DirEntry` representing the symbolic link. - /// - /// # Returns - /// - `Some(FolderEntity)` if the symbolic link has a valid file name. - /// - `None` otherwise. - fn symlink(entity: &DirEntry) -> Option { - entity.path().file_name().map(|filename| FolderEntity { - name: filename.to_string_lossy().to_string(), - fullname: entity.path().to_string_lossy().to_string(), - kind: FolderEntityType::SymbolicLink, - details: FolderEntityDetails::from(entity), - }) - } -} diff --git a/crates/stypes/src/command/folders/mod.rs b/crates/stypes/src/command/folders/mod.rs deleted file mode 100644 index 50c2b27dfd..0000000000 --- a/crates/stypes/src/command/folders/mod.rs +++ /dev/null @@ -1,60 +0,0 @@ -mod extending; - -use crate::*; - -/// Represents the type of a folder entity in the file system. -#[allow(clippy::upper_case_acronyms)] -#[derive(Clone, Serialize, Deserialize, Debug)] -pub enum FolderEntityType { - /// A block device (e.g., a disk or partition). - BlockDevice, - /// A character device (e.g., a terminal or serial port). - CharacterDevice, - /// A directory. - Directory, - /// A named pipe (FIFO). - FIFO, - /// A regular file. - File, - /// A socket. - Socket, - /// A symbolic link. - SymbolicLink, -} - -/// Contains detailed information about a folder entity. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct FolderEntityDetails { - /// The name of the file or folder. - filename: String, - /// The full path to the file or folder. - full: String, - /// The directory path containing the file or folder. - path: String, - /// The base name of the file or folder. - basename: String, - /// The file extension, if applicable. - ext: String, -} - -/// Represents the result of scanning a folder. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct FoldersScanningResult { - /// A list of folder entities found during the scan. - pub list: Vec, - /// Indicates whether the maximum length of results was reached. - pub max_len_reached: bool, -} - -/// Represents a folder entity in the file system. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct FolderEntity { - /// The name of the entity (file or folder). - name: String, - /// The full path of the entity. - fullname: String, - /// The type of the entity (e.g., file, directory, symbolic link). - kind: FolderEntityType, - /// Optional detailed information about the entity. - details: Option, -} diff --git a/crates/stypes/src/command/mod.rs b/crates/stypes/src/command/mod.rs deleted file mode 100644 index eb058a5993..0000000000 --- a/crates/stypes/src/command/mod.rs +++ /dev/null @@ -1,34 +0,0 @@ -mod extending; - -mod dltstat; -mod folders; -mod profiles; -mod serial; - -pub use dltstat::*; -pub use folders::*; -pub use profiles::*; -pub use serial::*; - -use crate::*; - -/// Represents the result of a command execution. -/// At the core level, this type is used for all commands invoked within an `UnboundSession`. -/// It is only used to indicate the successful completion or interruption of a command. -#[derive(Clone, Serialize, Deserialize, Debug)] -#[serde(bound(deserialize = "T: DeserializeOwned"))] -pub enum CommandOutcome { - /// Indicates that the command was successfully completed. - Finished(T), - /// Indicates that the command execution was interrupted. - Cancelled, -} - -/// Similar to `CommandOutcome`, but additionally contains the identifier of the executed command. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub enum UuidCommandOutcome { - /// Indicates that the command was successfully completed. - Finished((Uuid, T)), - /// Indicates that the command execution was interrupted. - Cancelled(Uuid), -} diff --git a/crates/stypes/src/command/serial/mod.rs b/crates/stypes/src/command/serial/mod.rs deleted file mode 100644 index 1e98012697..0000000000 --- a/crates/stypes/src/command/serial/mod.rs +++ /dev/null @@ -1,8 +0,0 @@ -use crate::*; - -/// Represents a list of serial ports. -/// -/// This structure contains a vector of strings, where each string represents the name -/// or identifier of a serial port available on the system. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct SerialPortsList(pub Vec); diff --git a/crates/stypes/src/lf_transition/extending.rs b/crates/stypes/src/lf_transition/extending.rs deleted file mode 100644 index b3616f9a10..0000000000 --- a/crates/stypes/src/lf_transition/extending.rs +++ /dev/null @@ -1,53 +0,0 @@ -use crate::*; - -impl LifecycleTransition { - /// Retrieves the `Uuid` associated with the lifecycle transition. - /// - /// # Returns - /// - The `Uuid` of the operation, regardless of its state. - pub fn uuid(&self) -> Uuid { - match self { - Self::Started { uuid, alias: _ } => *uuid, - Self::Ticks { uuid, ticks: _ } => *uuid, - Self::Stopped(uuid) => *uuid, - } - } - - /// Creates a new `LifecycleTransition::Started` instance. - /// - /// # Parameters - /// - `uuid`: The unique identifier of the operation. - /// - `alias`: A user-friendly name for the operation. - /// - /// # Returns - /// - A new `LifecycleTransition::Started` instance. - pub fn started(uuid: &Uuid, alias: &str) -> Self { - LifecycleTransition::Started { - uuid: *uuid, - alias: alias.to_owned(), - } - } - - /// Creates a new `LifecycleTransition::Stopped` instance. - /// - /// # Parameters - /// - `uuid`: The unique identifier of the operation. - /// - /// # Returns - /// - A new `LifecycleTransition::Stopped` instance. - pub fn stopped(uuid: &Uuid) -> Self { - LifecycleTransition::Stopped(*uuid) - } - - /// Creates a new `LifecycleTransition::Ticks` instance. - /// - /// # Parameters - /// - `uuid`: The unique identifier of the operation. - /// - `ticks`: Progress information associated with the operation. - /// - /// # Returns - /// - A new `LifecycleTransition::Ticks` instance. - pub fn ticks(uuid: &Uuid, ticks: Ticks) -> Self { - LifecycleTransition::Ticks { uuid: *uuid, ticks } - } -} diff --git a/crates/stypes/src/lf_transition/mod.rs b/crates/stypes/src/lf_transition/mod.rs deleted file mode 100644 index 8d198d895d..0000000000 --- a/crates/stypes/src/lf_transition/mod.rs +++ /dev/null @@ -1,25 +0,0 @@ -mod extending; - -use crate::*; - -/// Describes the progress of an operation. -#[derive(Debug, Serialize, Deserialize, Clone)] -pub enum LifecycleTransition { - /// The operation has started. - Started { - /// The unique identifier of the operation. - uuid: Uuid, - /// A user-friendly name of the operation for display purposes. - alias: String, - }, - /// The progress of the operation. - Ticks { - /// The unique identifier of the operation. - uuid: Uuid, - /// The progress data associated with the operation. - ticks: Ticks, - }, - /// The operation has completed or was interrupted. - /// - `Uuid`: The unique identifier of the operation. - Stopped(Uuid), -} diff --git a/crates/stypes/src/lib.rs b/crates/stypes/src/lib.rs index fc2c4f1498..6f18b13478 100644 --- a/crates/stypes/src/lib.rs +++ b/crates/stypes/src/lib.rs @@ -2,26 +2,22 @@ mod attachment; mod callback; -mod command; mod error; -mod lf_transition; mod miscellaneous; mod observe; mod operations; mod plugins; -mod progress; +mod shell; pub use attachment::*; pub use callback::*; -pub use command::*; pub use error::*; -pub use lf_transition::*; pub use miscellaneous::*; pub use observe::*; pub use operations::*; pub use plugins::*; -pub use progress::*; +pub use shell::*; -pub(crate) use serde::{Deserialize, Serialize, de::DeserializeOwned}; +pub(crate) use serde::{Deserialize, Serialize}; pub(crate) use std::{collections::HashMap, path::PathBuf}; pub(crate) use uuid::Uuid; diff --git a/crates/stypes/src/miscellaneous/converting.rs b/crates/stypes/src/miscellaneous/converting.rs index f242e3b405..4461662120 100644 --- a/crates/stypes/src/miscellaneous/converting.rs +++ b/crates/stypes/src/miscellaneous/converting.rs @@ -1,12 +1,6 @@ use crate::*; use std::ops::RangeInclusive; -impl From> for MapKeyValue { - fn from(map: HashMap) -> Self { - MapKeyValue(map) - } -} - impl From> for GrabbedElementList { /// Converts a `Vec` into a `GrabbedElementList`. /// diff --git a/crates/stypes/src/miscellaneous/mod.rs b/crates/stypes/src/miscellaneous/mod.rs index c0ddf56056..06ebaf835a 100644 --- a/crates/stypes/src/miscellaneous/mod.rs +++ b/crates/stypes/src/miscellaneous/mod.rs @@ -3,9 +3,6 @@ mod extending; use crate::*; -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct MapKeyValue(pub HashMap); - /// Representation of ranges. We cannot use std ranges as soon as no way /// to derive Serialize, Deserialize #[derive(Clone, Serialize, Deserialize, Debug)] diff --git a/crates/stypes/src/observe/mod.rs b/crates/stypes/src/observe/mod.rs index ecbf2a91d3..22fd54f061 100644 --- a/crates/stypes/src/observe/mod.rs +++ b/crates/stypes/src/observe/mod.rs @@ -15,13 +15,6 @@ pub struct MulticastInfo { pub interface: Option, } -/// Configuration for UDP connections. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct UdpConnectionInfo { - /// A list of multicast addresses to listen on. - pub multicast_addr: Vec, -} - /// Specifies the parser to be used for processing session data. #[allow(clippy::large_enum_variant)] #[derive(Debug, Serialize, Deserialize, Clone)] diff --git a/crates/stypes/src/operations/mod.rs b/crates/stypes/src/operations/mod.rs index 8b01a6197c..8ea2f07a63 100644 --- a/crates/stypes/src/operations/mod.rs +++ b/crates/stypes/src/operations/mod.rs @@ -38,9 +38,3 @@ pub struct ResultU64(pub u64); #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ResultBool(pub bool); - -/// Used only for debug session lifecycle -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ResultSleep { - pub sleep_well: bool, -} diff --git a/crates/stypes/src/plugins/mod.rs b/crates/stypes/src/plugins/mod.rs index 43e415a4cd..273dc74c50 100644 --- a/crates/stypes/src/plugins/mod.rs +++ b/crates/stypes/src/plugins/mod.rs @@ -217,15 +217,3 @@ pub struct ColumnInfo { /// Width of column (-1) for unlimited. pub width: i16, } - -/// Represents a list of [`PluginEntity`]. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct PluginsList(pub Vec); - -/// Represents a list of [`InvalidPluginEntity`]. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct InvalidPluginsList(pub Vec); - -/// Represents a list of [`InvalidPluginEntity`]. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct PluginsPathsList(pub Vec); diff --git a/crates/stypes/src/progress/extending.rs b/crates/stypes/src/progress/extending.rs deleted file mode 100644 index c40b14cce4..0000000000 --- a/crates/stypes/src/progress/extending.rs +++ /dev/null @@ -1,15 +0,0 @@ -use crate::*; - -impl Ticks { - /// Checks if the operation associated with the `Ticks` instance is complete. - /// - /// # Returns - /// - `true` if the `count` equals `total` and `total` is not `None`. - /// - `false` otherwise. - /// - /// # Details - /// - If `total` is `None`, the operation is considered incomplete. - pub fn done(&self) -> bool { - self.total.is_some_and(|total| self.count == total) - } -} diff --git a/crates/stypes/src/progress/mod.rs b/crates/stypes/src/progress/mod.rs deleted file mode 100644 index 87dcf8608c..0000000000 --- a/crates/stypes/src/progress/mod.rs +++ /dev/null @@ -1,38 +0,0 @@ -mod extending; - -use crate::*; - -/// Represents a notification about an event (including potential errors) -/// related to processing a specific log entry, if such data is available. -#[derive(Debug, Serialize, Deserialize, Clone)] -pub struct Notification { - /// The severity level of the event. - pub severity: Severity, - /// The content or message describing the event. - pub content: String, - /// The log entry number that triggered the event, if applicable. - pub line: Option, -} - -/// Describes the progress of an operation. -#[derive(Debug, Serialize, Deserialize, Clone)] -pub enum Progress { - /// Represents the current progress status. - Ticks(Ticks), - /// A notification related to the progress of the operation. - Notification(Notification), - /// Indicates that the operation has been stopped. - Stopped, -} - -/// Provides detailed information about the progress of an operation. -#[derive(Debug, Serialize, Deserialize, Clone, Default)] -pub struct Ticks { - /// The current progress count, typically representing `n` out of `100%`. - pub count: u64, - /// The name of the current progress stage, for user display purposes. - pub state: Option, - /// The total progress counter. Usually `100`, but for file operations, - /// it might represent the file size, where `count` indicates the number of bytes read. - pub total: Option, -} diff --git a/crates/stypes/src/command/profiles/extending.rs b/crates/stypes/src/shell/extending.rs similarity index 100% rename from crates/stypes/src/command/profiles/extending.rs rename to crates/stypes/src/shell/extending.rs diff --git a/crates/stypes/src/command/profiles/mod.rs b/crates/stypes/src/shell/mod.rs similarity index 62% rename from crates/stypes/src/command/profiles/mod.rs rename to crates/stypes/src/shell/mod.rs index 4eb77f0b84..6544b8803a 100644 --- a/crates/stypes/src/command/profiles/mod.rs +++ b/crates/stypes/src/shell/mod.rs @@ -1,3 +1,5 @@ +//! Shell types used to run process sources through a user shell. + mod extending; use crate::*; @@ -20,10 +22,3 @@ pub struct ShellProfile { /// Path to executable file of shell pub path: PathBuf, } - -/// Represents a list of serial ports. -/// -/// This structure contains a vector of strings, where each string represents the name -/// or identifier of a serial port available on the system. -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct ProfileList(pub Vec); diff --git a/docs/development/core/plugins.md b/docs/development/core/plugins.md index 2cf0392688..cc9ab8ce15 100644 --- a/docs/development/core/plugins.md +++ b/docs/development/core/plugins.md @@ -18,7 +18,7 @@ On startup, the `PluginsManager` scans the designated plugins directory (` To optimize startup performance, the `PluginsManager` employs a caching mechanism. After the initial scan, extracted plugin metadata, configurations, and a hash of the binary are saved to a cache file (located within the plugins directory). On subsequent runs, unchanged plugins are loaded directly from this cache, avoiding the need to recompile and re-extract their information from the binary. The cache can typically be invalidated or reloaded via a UI action. -The `PluginsManager` keeps track of all loaded plugins, their current state, and configurations, making them available to the rest of the Chipmunk application, such as providing loaded plugins to the `UnboundSession` view in the UI. +The `PluginsManager` keeps track of all loaded plugins, their current state, and configurations, making them available to the rest of the Chipmunk application. The host plugin service (`crates/app/src/host/service/plugin/`) owns the manager and publishes its data to the UI. ## Plugin Hosts (Parser and Byte-Source)