Add spinner to consume command (#1975)

Closes #1881

Add spinner to consume commmand, the spinner will not be rendered if we are in a non-tty environment


https://user-images.githubusercontent.com/22335041/144466096-1eb8df62-7601-475e-82e6-8504f5b1e3a2.mp4

CHANGELOG.md

@@ -4,6 +4,7 @@
 * Add support for tuple structs in fluvio-protocol derived macros. ([#1828](https://github.com/infinyon/fluvio/issues/1828))
 * Expose fluvio completions in the top-level subcommand. ([#1850](https://github.com/infinyon/fluvio/issues/1850))
 * Make installation more reliable ([#1961](https://github.com/infinyon/fluvio/pull/1961))
+* Add Spinner to `fluvio consume` command. ([#1881](https://github.com/infinyon/fluvio/issues/1881))
 
 ## Platform Version 0.9.13 - 2021-11-19
 * Fix connector create with `create_topic` option to succeed if topic already exists. ([#1823](https://github.com/infinyon/fluvio/pull/1823))

crates/fluvio-cli/Cargo.toml

@@ -84,6 +84,7 @@ k8-types = { version = "0.5.0", features = ["core"] }
 fluvio-types = { version = "0.2.0", path = "../fluvio-types", optional = true }
 fluvio-future = { version = "0.3.0", features = ["fs", "io", "subscriber", "native2_tls"], optional = true }
 fluvio-sc-schema = { version = "0.11.0", path = "../fluvio-sc-schema", features = ["use_serde"], optional = true }
+indicatif = "0.16.2"
 
 [dev-dependencies]
 fluvio-future = { version = "0.3.0", features = ["fixture"] }

crates/fluvio-cli/src/consume/mod.rs

@@ -35,6 +35,7 @@ use crossterm::{
     terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
 };
 
+use crate::render::ProgressRenderer;
 use crate::{CliError, Result};
 use crate::common::FluvioExtensionMetadata;
 use self::record_format::{
@@ -414,6 +415,10 @@ impl ConsumeOpt {
         if io::stdout().is_tty() {
             // This needs to know if it is a tty before opening this
             let mut user_input_reader = EventStream::new();
+            let pb = indicatif::ProgressBar::new(1);
+            pb.set_style(indicatif::ProgressStyle::default_bar().template("{spinner}"));
+            pb.enable_steady_tick(100);
+            let pb: ProgressRenderer = pb.into();
 
             loop {
                 select! {
@@ -437,6 +442,7 @@ impl ConsumeOpt {
                                 &mut header_print,
                                 &mut maybe_terminal_stdout,
                                 &mut maybe_table_model,
+                                &pb,
                             );
                         },
                         None => break,
@@ -461,6 +467,7 @@ impl ConsumeOpt {
                 }
             }
         } else {
+            let pb = ProgressRenderer::default();
             // We do not support `--output=full_table` when we don't have a TTY (i.e., CI environment)
             while let Some(result) = stream.next().await {
                 let result: std::result::Result<Record, _> = result;
@@ -481,6 +488,7 @@ impl ConsumeOpt {
                     &mut header_print,
                     &mut None,
                     &mut None,
+                    &pb,
                 );
             }
         }
@@ -512,6 +520,7 @@ impl ConsumeOpt {
         header_print: &mut bool,
         terminal: &mut Option<Terminal<CrosstermBackend<Stdout>>>,
         table_model: &mut Option<TableModel>,
+        pb: &ProgressRenderer,
     ) {
         let formatted_key = record
             .key()
@@ -563,10 +572,11 @@ impl ConsumeOpt {
         if self.output != Some(ConsumeOutputType::full_table) {
             match formatted_value {
                 Some(value) if self.key_value => {
-                    println!("[{}] {}", formatted_key, value);
+                    let output = format!("[{}] {}", formatted_key, value);
+                    pb.println(&output);
                 }
                 Some(value) => {
-                    println!("{}", value);
+                    pb.println(&value);
                 }
                 // (Some(_), None) only if JSON cannot be printed, so skip.
                 _ => debug!("Skipping record that cannot be formatted"),

crates/fluvio-cli/src/lib.rs

@@ -16,6 +16,7 @@ mod connector;
 mod tableformat;
 mod smartmodule;
 mod derivedstream;
+mod render;
 
 pub(crate) use error::{Result, CliError};
 

crates/fluvio-cli/src/render.rs

@@ -0,0 +1,30 @@
+use indicatif::ProgressBar;
+
+#[derive(Debug)]
+pub enum ProgressRenderer {
+    /// Render the progress using eprintln macro
+    Std,
+    /// Render the progress using Indicatiff
+    Indicatiff(ProgressBar),
+}
+
+impl ProgressRenderer {
+    pub fn println(&self, msg: &str) {
+        match self {
+            ProgressRenderer::Std => println!("{}", msg),
+            ProgressRenderer::Indicatiff(pb) => pb.println(msg),
+        }
+    }
+}
+
+impl From<ProgressBar> for ProgressRenderer {
+    fn from(pb: ProgressBar) -> Self {
+        Self::Indicatiff(pb)
+    }
+}
+
+impl Default for ProgressRenderer {
+    fn default() -> Self {
+        Self::Std
+    }
+}