CLI: Collector changes for producer & Increase version

* Apply the recent changes in producer with records collector in
  chipmunk CLI tool.
* Change the final report since we have access to bytes information with
  the new changes.
* Increase app version & Update changelog

Author: AmmarAbouZor

application/apps/indexer/processor/src/producer/mod.rs

@@ -306,6 +306,24 @@ impl<T: LogMessage, P: Parser<T>, D: ByteSource> MessageProducer<T, P, D> {
         }
     }
 
+    /// Total loaded bytes form byte source in this session.
+    #[inline]
+    pub fn total_loaded_bytes(&self) -> usize {
+        self.total_loaded
+    }
+    
+    /// Total skipped bytes by source and parser in this session.
+    #[inline]
+    pub fn total_skipped_bytes(&self) -> usize {
+        self.total_skipped
+    }
+
+    /// Total amount of parsed items produced in this session.
+    #[inline]
+    pub fn total_produced_items(&self) -> usize {
+        self.total_messages
+    }
+
     /// Append incoming (SDE) Source-Data-Exchange to the underline byte source data.
     pub async fn sde_income(
         &mut self,

cli/chipmunk-cli/CHANGELOG.md

@@ -1,3 +1,10 @@
+# 0.2.2
+
+## Changes:
+
+* Include bytes informations in session summary.
+* Internal performance improvements.
+
 # 0.2.1
 
 ## Changes:

cli/chipmunk-cli/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "chipmunk-cli"
-version = "0.2.1"
+version = "0.2.2"
 authors = ["Ammar Abou Zor <ammar.abou.zor@accenture.com>"]
 edition = "2024"
 description = "CLI Tool for parsing bytes form different source supporting multiple data formats"

cli/chipmunk-cli/Cargo.toml

@@ -4,8 +4,8 @@ use anyhow::Context;
 use std::{io::Write as _, path::PathBuf};
 use tokio_util::sync::CancellationToken;
 
-use parsers::{LogMessage, Parser};
-use processor::producer::MessageProducer;
+use parsers::{LogMessage, ParseYield, Parser};
+use processor::producer::{GeneralLogCollector, MessageProducer, ProduceSummary};
 use sources::ByteSource;
 
 use crate::session::create_append_file_writer;
@@ -37,33 +37,38 @@ where
     W: MessageFormatter,
 {
     let mut producer = MessageProducer::new(parser, bytesource);
+    let mut collector = GeneralLogCollector::default();
 
     let mut file_writer = create_append_file_writer(&output_path)?;
 
+    let write_sum = |p: &mut MessageProducer<_, _, _>| {
+        super::write_summary(
+            p.total_produced_items(),
+            p.total_loaded_bytes(),
+            p.total_skipped_bytes(),
+        );
+    };
+
     let mut msg_count = 0;
-    let mut skipped_count = 0;
-    let mut empty_count = 0;
-    let mut incomplete_count = 0;
 
     loop {
+        collector.get_records().clear();
         tokio::select! {
             _ = cancel_token.cancelled() => {
                 file_writer.flush().context("Error writing data to file.")?;
-                super::write_summary(msg_count, skipped_count, empty_count, incomplete_count);
+                write_sum(&mut producer);
 
                 return Ok(());
             },
-            Some(items) = producer.read_next_segment() => {
-                for (_, item) in items {
-                    match item {
-                        parsers::MessageStreamItem::Item(parse_yield) => {
-                            let msg = match parse_yield {
-                                parsers::ParseYield::Message(msg) => msg,
-                                parsers::ParseYield::Attachment(_attachment) => {
-                                    // attachment are postponed for now.
-                                    continue;
-                                }
-                                parsers::ParseYield::MessageAndAttachment((msg, _attachment)) => msg,
+            res = producer.produce_next(&mut collector) => {
+                let summary = res?;
+                match summary {
+                    ProduceSummary::Processed {..} => {
+                        for record in collector.get_records() {
+                           let msg =  match record {
+                                ParseYield::Message(msg) => msg,
+                                ParseYield::Attachment(..) => continue,
+                                ParseYield::MessageAndAttachment((msg, _att)) => msg,
                             };
                             msg_formatter.write_msg(&mut file_writer, msg)?;
 
@@ -72,16 +77,12 @@ where
                                 println!("Processing... {msg_count} messages have been written to file.");
                             }
                         }
-                        parsers::MessageStreamItem::Skipped => skipped_count += 1,
-                        parsers::MessageStreamItem::Incomplete => incomplete_count += 1,
-                        parsers::MessageStreamItem::Empty => empty_count += 1,
-                        parsers::MessageStreamItem::Done => {
-                            println!("Parsing Done");
-                            super::write_summary(msg_count, skipped_count, empty_count, incomplete_count);
-
-                            return Ok(());
-                        }
-                    }
+                    },
+                    // We don't support file tailing in the CLI tool.
+                    ProduceSummary::NoBytesAvailable {..} | ProduceSummary::Done {..} => {
+                        write_sum(&mut producer);
+                        return Ok(());
+                    },
                 }
             }
 

cli/chipmunk-cli/src/session/file.rs

@@ -123,27 +123,15 @@ where
 }
 
 /// Writes summary of the process session.
-fn write_summary(
-    msg_count: usize,
-    skipped_count: usize,
-    empty_count: usize,
-    incomplete_count: usize,
-) {
+fn write_summary(msg_count: usize, loaded_bytes: usize, skipped_bytes: usize) {
     const UNDERLINE_ANSI: &str = "\x1b[4m";
     const RESET_ANSI: &str = "\x1b[0m";
 
     println!("{UNDERLINE_ANSI}Process Summary{RESET_ANSI}:");
 
     println!("* {msg_count} messages has been written to file.");
-    if skipped_count > 0 {
-        println!("* {skipped_count} messages skipped");
-    }
-    if empty_count > 0 {
-        println!("* {empty_count} messages were empty");
-    }
-    if incomplete_count > 0 {
-        println!("* {incomplete_count} messages were incomplete");
-    }
+    println!("* {loaded_bytes} bytes has been loaded from source.");
+    println!("* {skipped_bytes} bytes has been skipped.");
 }
 
 /// Creates or append a file with the provided [`file_path`] returning its buffer writer.

cli/chipmunk-cli/src/session/mod.rs

@@ -5,8 +5,8 @@ use std::{io::Write as _, ops::Deref, path::PathBuf, time::Duration};
 use tokio::sync::watch;
 use tokio_util::sync::CancellationToken;
 
-use parsers::{LogMessage, Parser};
-use processor::producer::MessageProducer;
+use parsers::{LogMessage, ParseYield, Parser};
+use processor::producer::{GeneralLogCollector, MessageProducer, ProduceSummary};
 use sources::{ByteSource, socket::tcp::reconnect::ReconnectStateMsg};
 
 use crate::session::create_append_file_writer;
@@ -44,25 +44,31 @@ where
 
     let mut file_writer = create_append_file_writer(&output_path)?;
 
+    let mut collector = GeneralLogCollector::default();
+
     // Flush the file writer every 500 milliseconds for users tailing the output
     // file when messages are receive in relative slow frequency.
     let mut flush_interval = tokio::time::interval(Duration::from_millis(500));
 
     // Counters to keep track on the status of the session.
-    let mut msg_count = 0;
     let mut reconnecting = false;
-    let mut skipped_count = 0;
-    let mut empty_count = 0;
-    let mut incomplete_count = 0;
 
     // Keep track how many message has been received since the last flush.
     let mut msg_since_last_flush = 0;
 
+    let write_sum = |p: &mut MessageProducer<_, _, _>| {
+        super::write_summary(
+            p.total_produced_items(),
+            p.total_loaded_bytes(),
+            p.total_skipped_bytes(),
+        );
+    };
+
     loop {
         tokio::select! {
             _ = cancel_token.cancelled() => {
                 file_writer.flush().context("Error writing data to file.")?;
-                super::write_summary(msg_count, skipped_count, empty_count, incomplete_count);
+                write_sum(&mut producer);
 
                 return Ok(());
             }
@@ -98,36 +104,30 @@ where
             }
             _ = update_interval.tick() => {
                 if !reconnecting {
+                    let msg_count = producer.total_produced_items();
                     println!("Processing... {msg_count} messages have been written to file.");
                 }
             }
-            Some(items) = producer.read_next_segment() => {
-                for (_, item) in items {
-                    match item {
-                        parsers::MessageStreamItem::Item(parse_yield) => {
-                            let msg = match parse_yield {
-                                parsers::ParseYield::Message(msg) => msg,
-                                parsers::ParseYield::Attachment(_attachment) => {
-                                    // attachment are postponed for now.
-                                    continue;
-                                }
-                                parsers::ParseYield::MessageAndAttachment((msg, _attachment)) => msg,
+            res = producer.produce_next(&mut collector) => {
+                let summary = res?;
+
+                match summary {
+                    ProduceSummary::Processed {..} => {
+                        for record in collector.get_records() {
+                           let msg =  match record {
+                                ParseYield::Message(msg) => msg,
+                                ParseYield::Attachment(..) => continue,
+                                ParseYield::MessageAndAttachment((msg, _att)) => msg,
                             };
                             msg_formatter.write_msg(&mut file_writer, msg)?;
                             msg_since_last_flush += 1;
-
-                            msg_count += 1;
                         }
-                        parsers::MessageStreamItem::Skipped => skipped_count += 1,
-                        parsers::MessageStreamItem::Incomplete => incomplete_count += 1,
-                        parsers::MessageStreamItem::Empty => empty_count += 1,
-                        parsers::MessageStreamItem::Done => {
-                            println!("Parsing Done");
-                            super::write_summary(msg_count, skipped_count, empty_count, incomplete_count);
-
-                            return Ok(());
-                        }
-                    }
+                    },
+                    // No tailing support for streams.
+                    ProduceSummary::NoBytesAvailable {..} | ProduceSummary::Done {..} => {
+                        write_sum(&mut producer);
+                        return Ok(());
+                    },
                 }
             }
         };