Feat/cs/processor refactor

Cargo.toml

@@ -12,10 +12,9 @@ repository = "https://github.com/ReagentX/Logria"
 version = "0.0.0"
 
 [dependencies]
-clap = { version = "=4.5.39", features = ["cargo"] }
+clap = { version = "=4.5.40", features = ["cargo"] }
 crossterm = "=0.29.0"
 dirs = "=6.0.0"
-format_num = "=0.1.0"
 is_executable = "=1.0.4"
 regex = "=1.11.1"
 serde = { version = "=1.0.219", features = ["derive"] }

src/communication/handlers/highlight.rs

@@ -6,13 +6,15 @@ use regex::bytes::Regex;
 use super::{handler::Handler, processor::ProcessorMethods};
 use crate::{
     communication::{
-        handlers::user_input::UserInputHandler, input::InputType::Normal, reader::MainWindow,
+        handlers::{processor::update_progress, user_input::UserInputHandler},
+        input::InputType::Normal,
+        reader::MainWindow,
     },
     constants::cli::{
         cli_chars::{COMMAND_CHAR, HIGHLIGHT_CHAR, NORMAL_STR, TOGGLE_HIGHLIGHT_CHAR},
         patterns::ANSI_COLOR_PATTERN,
     },
-    ui::scroll::{self, update_current_match_index, ScrollState},
+    ui::scroll::{self, ScrollState, update_current_match_index},
 };
 
 pub struct HighlightHandler {
@@ -73,34 +75,34 @@ impl HighlightHandler {
 impl ProcessorMethods for HighlightHandler {
     /// Process matches, loading the buffer of indexes to matched messages in the main buffer
     fn process_matches(&mut self, window: &mut MainWindow) -> Result<()> {
-        // TODO: Possibly async? Possibly loading indicator for large jobs?
+        let mut wrote_progress = false;
         if self.current_pattern.is_some() {
             // Start from where we left off to the most recent message
-            let buf_range = (window.config.last_index_regexed, window.messages().len());
+            let start = window.config.last_index_regexed;
+            let end = window.messages().len();
 
-            // Iterate "forever", skipping to the start and taking up till end-start
-            // TODO: Something to indicate progress
-            for index in (0..).skip(buf_range.0).take(buf_range.1 - buf_range.0) {
+            for index in start..end {
                 if self.test(&window.messages()[index]) {
                     window.config.matched_rows.push(index);
                 }
 
+                // Update the user interface with the current state
+                wrote_progress = update_progress(window, start, end, index)?;
+
                 // Update the last spot so we know where to start next time
                 window.config.last_index_regexed = index + 1;
             }
+            if wrote_progress {
+                window.write_status()?;
+            }
         }
         Ok(())
     }
 
     /// Return the app to a normal input state
     fn return_to_normal(&mut self, window: &mut MainWindow) -> Result<()> {
         self.clear_matches(window)?;
-        // Handle reset of scroll state
         window.config.current_matched_row = 0;
-        if matches!(window.config.scroll_state, ScrollState::Centered) {
-            window.config.scroll_state = ScrollState::Free;
-        }
-
         window.config.current_status = None;
         window.update_input_type(Normal)?;
         window.set_cli_cursor(None)?;
@@ -116,7 +118,9 @@ impl ProcessorMethods for HighlightHandler {
         window.config.matched_rows.clear();
         window.config.last_index_regexed = 0;
         window.config.highlight_match = false;
-        window.config.scroll_state = ScrollState::Free;
+        if matches!(window.config.scroll_state, ScrollState::Centered) {
+            window.config.scroll_state = ScrollState::Free;
+        }
         window.reset_command_line()?;
         Ok(())
     }

src/communication/handlers/parser.rs

@@ -6,7 +6,9 @@ use regex::Regex;
 use crate::{
     communication::{
         handlers::{
-            handler::Handler, multiple_choice::MultipleChoiceHandler, processor::ProcessorMethods,
+            handler::Handler,
+            multiple_choice::MultipleChoiceHandler,
+            processor::{ProcessorMethods, update_progress},
         },
         input::{InputType::Normal, StreamType},
         reader::MainWindow,
@@ -219,21 +221,14 @@ impl ProcessorMethods for ParserHandler {
     fn process_matches(&mut self, window: &mut MainWindow) -> Result<()> {
         // Only process if the parser is set up properly
         if let ParserState::Full = window.config.parser_state {
-            // TODO: Possibly async? Possibly loading indicator for large jobs?
             if self.parser.is_some() {
+                let mut wrote_progress = false;
                 // Start from where we left off to the most recent message
-                let buf_range = (
-                    window.config.last_index_processed,
-                    window.previous_messages().len(),
-                );
-
-                // Iterate "forever", skipping to the start and taking up till end-start
-                // TODO: Something to indicate progress
-                let last = buf_range.1.checked_sub(1).unwrap_or(buf_range.0);
-                for index in (0..)
-                    .skip(buf_range.0)
-                    .take(buf_range.1.checked_sub(buf_range.0).unwrap_or(buf_range.0))
-                {
+                let start = window.config.last_index_processed;
+                let end = window.previous_messages().len();
+
+                let last = end.checked_sub(1).unwrap_or(end);
+                for index in start..end {
                     if window.config.aggregation_enabled {
                         match self.aggregate_handle(
                             &window.previous_messages()[index],
@@ -260,9 +255,16 @@ impl ProcessorMethods for ParserHandler {
                     ) {
                         window.config.auxiliary_messages.push(message);
                     }
+
+                    // Update the user interface with the current state
+                    wrote_progress = update_progress(window, start, end, index)?;
+
                     // Update the last spot so we know where to start next time
                     window.config.last_index_processed = index + 1;
                 }
+                if wrote_progress {
+                    window.write_status()?;
+                }
             }
         }
         Ok(())
@@ -394,6 +396,11 @@ impl Handler for ParserHandler {
                         if window.config.aggregation_enabled {
                             window.config.current_status = Some(self.status.clone());
                             window.config.aggregation_enabled = false;
+                            if let Some(parser) = &mut self.parser {
+                                parser.aggregator_map.values_mut().for_each(|agg| {
+                                    agg.reset();
+                                });
+                            }
                         } else {
                             let new_status = self.status.clone();
                             window.config.current_status = Some(new_status.replace(
@@ -531,13 +538,13 @@ mod parse_tests {
                 "Sum",
                 "    Total: 5,850",
                 "Count",
-                "    10\u{1b}[0m: 1 (1%)",
-                "    100\u{1b}[0m: 1 (1%)",
-                "    101\u{1b}[0m: 1 (1%)",
-                "    102\u{1b}[0m: 1 (1%)",
-                "    103\u{1b}[0m: 1 (1%)",
+                "    95\u{1b}[0m: 1 (1%)",
+                "    96\u{1b}[0m: 1 (1%)",
+                "    97\u{1b}[0m: 1 (1%)",
+                "    98\u{1b}[0m: 1 (1%)",
+                "    99\u{1b}[0m: 1 (1%)",
                 "Mode",
-                "    10\u{1b}[0m: 1 (1%)",
+                "    99\u{1b}[0m: 1 (1%)",
             ]
         );
     }

src/communication/handlers/processor.rs

@@ -7,3 +7,33 @@ pub trait ProcessorMethods {
     fn clear_matches(&mut self, window: &mut MainWindow) -> Result<()>;
     fn process_matches(&mut self, window: &mut MainWindow) -> Result<()>;
 }
+
+/// The step size for progress indicator updates.
+const STEP: usize = 999;
+/// Threshold for when to print progress updates in the aggregator.
+const THRESHOLD: usize = 25_000;
+
+#[inline(always)]
+pub fn update_progress(
+    window: &mut MainWindow,
+    start: usize,
+    end: usize,
+    index: usize,
+) -> Result<bool> {
+    // Update the user interface with the current state
+    if end - start > THRESHOLD && (index % STEP == 0 || index == end - 1) {
+        let word = if index == end - 1 {
+            "Processed"
+        } else {
+            "Processing"
+        };
+        window.write_to_command_line(&format!(
+            "{word} messages: {}/{} ({}%)",
+            (index + 1) - start,
+            end - start,
+            ((index + 1 - start) * 100) / (end - start)
+        ))?;
+        return Ok(true);
+    }
+    Ok(false)
+}

src/communication/handlers/regex.rs

@@ -6,7 +6,9 @@ use regex::bytes::Regex;
 use super::{handler::Handler, processor::ProcessorMethods};
 use crate::{
     communication::{
-        handlers::user_input::UserInputHandler, input::InputType::Normal, reader::MainWindow,
+        handlers::{processor::update_progress, user_input::UserInputHandler},
+        input::InputType::Normal,
+        reader::MainWindow,
     },
     constants::cli::{
         cli_chars::{COMMAND_CHAR, NORMAL_STR, REGEX_CHAR, TOGGLE_HIGHLIGHT_CHAR},
@@ -64,21 +66,27 @@ impl RegexHandler {
 impl ProcessorMethods for RegexHandler {
     /// Process matches, loading the buffer of indexes to matched messages in the main buffer
     fn process_matches(&mut self, window: &mut MainWindow) -> Result<()> {
-        // TODO: Possibly async? Possibly loading indicator for large jobs?
+        let mut wrote_progress = false;
         if self.current_pattern.is_some() {
             // Start from where we left off to the most recent message
-            let buf_range = (window.config.last_index_regexed, window.messages().len());
+            // Start from where we left off to the most recent message
+            let start = window.config.last_index_regexed;
+            let end = window.messages().len();
 
-            // Iterate "forever", skipping to the start and taking up till end-start
-            // TODO: Something to indicate progress
-            for index in (0..).skip(buf_range.0).take(buf_range.1 - buf_range.0) {
+            for index in start..end {
                 if self.test(&window.messages()[index]) {
                     window.config.matched_rows.push(index);
                 }
 
+                // Update the user interface with the current state
+                wrote_progress = update_progress(window, start, end, index)?;
+
                 // Update the last spot so we know where to start next time
                 window.config.last_index_regexed = index + 1;
             }
+            if wrote_progress {
+                window.write_status()?;
+            }
         }
         Ok(())
     }

src/extensions/parser.rs

@@ -163,15 +163,15 @@ impl Parser {
                     }
                     AggregationMethod::Mode => {
                         self.aggregator_map
-                            .insert(method_name.to_string(), Box::new(Counter::new(Some(1))));
+                            .insert(method_name.to_string(), Box::new(Counter::mean()));
                     }
                     AggregationMethod::Sum => {
                         self.aggregator_map
                             .insert(method_name.to_string(), Box::new(Sum::new()));
                     }
                     AggregationMethod::Count => {
                         self.aggregator_map
-                            .insert(method_name.to_string(), Box::new(Counter::new(None)));
+                            .insert(method_name.to_string(), Box::new(Counter::new()));
                     }
                     AggregationMethod::Date(format) => {
                         self.aggregator_map.insert(