Add support for optional list stream output formatting (#6325)

* add support for optional list stream output formatting

* cargo fmt

* table: add ValueFormatter test

crates/nu-cli/src/eval_file.rs

@@ -109,7 +109,7 @@ pub fn print_table_or_error(
 
     // Make sure everything has finished
     if let Some(exit_code) = exit_code {
-        let mut exit_code: Vec<_> = exit_code.into_iter().collect();
+        let mut exit_code: Vec<_> = exit_code.into_iter().map(|(value, _)| value).collect();
         exit_code
             .pop()
             .and_then(|last_exit_code| match last_exit_code {

crates/nu-cli/src/util.rs

@@ -233,7 +233,7 @@ pub fn eval_source(
     match eval_block(engine_state, stack, &block, input, false, false) {
         Ok(mut pipeline_data) => {
             if let PipelineData::ExternalStream { exit_code, .. } = &mut pipeline_data {
-                if let Some(exit_code) = exit_code.take().and_then(|it| it.last()) {
+                if let Some((exit_code, _)) = exit_code.take().and_then(|it| it.last()) {
                     stack.add_env_var("LAST_EXIT_CODE".to_string(), exit_code);
                 } else {
                     set_last_exit_code(stack, 0);

crates/nu-command/src/core_commands/echo.rs

@@ -34,28 +34,32 @@ impl Command for Echo {
         call: &Call,
         _input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
-        call.rest(engine_state, stack, 0).map(|to_be_echoed| {
+        call.rest(engine_state, stack, 0)
-            let n = to_be_echoed.len();
+            .map(|to_be_echoed: Vec<Value>| {
-            match n.cmp(&1usize) {
+                let n = to_be_echoed.len();
-                //  More than one value is converted in a stream of values
+                match n.cmp(&1usize) {
-                std::cmp::Ordering::Greater => PipelineData::ListStream(
+                    //  More than one value is converted in a stream of values
-                    ListStream::from_stream(to_be_echoed.into_iter(), engine_state.ctrlc.clone()),
+                    std::cmp::Ordering::Greater => PipelineData::ListStream(
-                    None,
+                        ListStream::from_stream(
-                ),
+                            to_be_echoed.into_iter(),
+                            engine_state.ctrlc.clone(),
+                        ),
+                        None,
+                    ),
 
-                //  But a single value can be forwarded as it is
+                    //  But a single value can be forwarded as it is
-                std::cmp::Ordering::Equal => PipelineData::Value(to_be_echoed[0].clone(), None),
+                    std::cmp::Ordering::Equal => PipelineData::Value(to_be_echoed[0].clone(), None),
 
-                //  When there are no elements, we echo the empty string
+                    //  When there are no elements, we echo the empty string
-                std::cmp::Ordering::Less => PipelineData::Value(
+                    std::cmp::Ordering::Less => PipelineData::Value(
-                    Value::String {
+                        Value::String {
-                        val: "".to_string(),
+                            val: "".to_string(),
-                        span: call.head,
+                            span: call.head,
-                    },
+                        },
-                    None,
+                        None,
-                ),
+                    ),
-            }
+                }
-        })
+            })
     }
 
     fn examples(&self) -> Vec<Example> {

crates/nu-command/src/core_commands/for_.rs

@@ -91,7 +91,7 @@ impl Command for For {
             Value::List { vals, .. } => {
                 Ok(ListStream::from_stream(vals.into_iter(), ctrlc.clone())
                     .enumerate()
-                    .map(move |(idx, x)| {
+                    .map(move |(idx, (x, _))| {
                         stack.with_env(&orig_env_vars, &orig_env_hidden);
 
                         stack.add_var(

crates/nu-command/src/filters/columns.rs

@@ -91,7 +91,7 @@ fn getcol(
                 .into_pipeline_data(engine_state.ctrlc.clone()))
         }
         PipelineData::ListStream(stream, ..) => {
-            let v: Vec<_> = stream.into_iter().collect();
+            let v: Vec<_> = stream.into_iter().map(|(v, _)| v).collect();
             let input_cols = get_columns(&v);
 
             Ok(input_cols

crates/nu-command/src/filters/drop/column.rs

@@ -115,7 +115,7 @@ fn dropcol(
         PipelineData::ListStream(stream, ..) => {
             let mut output = vec![];
 
-            let v: Vec<_> = stream.into_iter().collect();
+            let v: Vec<_> = stream.into_iter().map(|(v, _)| v).collect();
             let input_cols = get_input_cols(v.clone());
             let kc = get_keep_columns(input_cols, columns);
             keep_columns = get_cellpath_columns(kc, span);

crates/nu-command/src/filters/find.rs

@@ -364,7 +364,7 @@ fn find_with_rest_and_highlight(
         PipelineData::ListStream(stream, meta) => {
             Ok(ListStream::from_stream(
                 stream
-                    .map(move |mut x| match &mut x {
+                    .map(move |(mut x, _)| match &mut x {
                         Value::Record { cols, vals, span } => {
                             let mut output = vec![];
                             for val in vals {

crates/nu-command/src/filters/length.rs

@@ -107,7 +107,7 @@ fn getcol(
                 .into_pipeline_data(engine_state.ctrlc.clone()))
         }
         PipelineData::ListStream(stream, ..) => {
-            let v: Vec<_> = stream.into_iter().collect();
+            let v: Vec<_> = stream.into_iter().map(|(v, _)| v).collect();
             let input_cols = get_columns(&v);
 
             Ok(input_cols

crates/nu-command/src/filters/lines.rs

@@ -69,7 +69,7 @@ impl Command for Lines {
 
                 let iter = stream
                     .into_iter()
-                    .filter_map(move |value| {
+                    .filter_map(move |(value, _)| {
                         if let Value::String { val, span } = value {
                             if split_char != "\r\n" && val.contains("\r\n") {
                                 split_char = "\r\n";

crates/nu-command/src/filters/par_each.rs

@@ -170,7 +170,7 @@ impl Command for ParEach {
             PipelineData::ListStream(stream, ..) => Ok(stream
                 .enumerate()
                 .par_bridge()
-                .map(move |(idx, x)| {
+                .map(move |(idx, (x, _))| {
                     let block = engine_state.get_block(block_id);
 
                     let mut stack = stack.clone();

crates/nu-command/src/filters/reject.rs

@@ -125,7 +125,7 @@ fn reject(
         PipelineData::ListStream(stream, ..) => {
             let mut output = vec![];
 
-            let v: Vec<_> = stream.into_iter().collect();
+            let v: Vec<_> = stream.into_iter().map(|(v, _)| v).collect();
             let input_cols = get_input_cols(v.clone());
             let kc = get_keep_columns(input_cols, columns);
             keep_columns = get_cellpath_columns(kc, span);

crates/nu-command/src/filters/select.rs

@@ -146,7 +146,7 @@ fn select(
                 .set_metadata(metadata))
         }
         PipelineData::ListStream(stream, metadata, ..) => Ok(stream
-            .map(move |x| {
+            .map(move |(x, _)| {
                 if !columns.is_empty() {
                     let mut cols = vec![];
                     let mut vals = vec![];

crates/nu-command/src/filters/wrap.rs

@@ -44,7 +44,7 @@ impl Command for Wrap {
                 })
                 .into_pipeline_data(engine_state.ctrlc.clone())),
             PipelineData::ListStream(stream, ..) => Ok(stream
-                .map(move |x| Value::Record {
+                .map(move |(x, _)| Value::Record {
                     cols: vec![name.clone()],
                     vals: vec![x],
                     span,

crates/nu-command/src/math/utils.rs

@@ -62,7 +62,9 @@ pub fn calculate(
     mf: impl Fn(&[Value], &Span) -> Result<Value, ShellError>,
 ) -> Result<Value, ShellError> {
     match values {
-        PipelineData::ListStream(s, ..) => helper_for_tables(&s.collect::<Vec<Value>>(), name, mf),
+        PipelineData::ListStream(s, ..) => {
+            helper_for_tables(&s.map(|(v, _)| v).collect::<Vec<Value>>(), name, mf)
+        }
         PipelineData::Value(Value::List { ref vals, .. }, ..) => match &vals[..] {
             [Value::Record { .. }, _end @ ..] => helper_for_tables(vals, name, mf),
             _ => mf(vals, &name),

crates/nu-command/src/system/complete.rs

@@ -70,7 +70,7 @@ impl Command for Complete {
                 }
 
                 if let Some(exit_code) = exit_code {
-                    let mut v: Vec<_> = exit_code.collect();
+                    let mut v: Vec<_> = exit_code.map(|(v, _)| v).collect();
 
                     if let Some(v) = v.pop() {
                         cols.push("exit_code".to_string());

crates/nu-command/src/viewers/griddle.rs

@@ -90,7 +90,7 @@ prints out the list properly."#
             }
             PipelineData::ListStream(stream, ..) => {
                 // dbg!("value::stream");
-                let data = convert_to_list(stream, config, call.head);
+                let data = convert_to_list(stream.map(|(v, _)| v), config, call.head);
                 if let Some(items) = data {
                     Ok(create_grid_output(
                         items,

crates/nu-command/src/viewers/table.rs

@@ -260,7 +260,7 @@ fn handle_row_stream(
             let ls_colors = get_ls_colors(ls_colors_env_str);
 
             ListStream::from_stream(
-                stream.map(move |mut x| match &mut x {
+                stream.map(move |(mut x, _)| match &mut x {
                     Value::Record { cols, vals, .. } => {
                         let mut idx = 0;
 
@@ -512,7 +512,11 @@ impl Iterator for PagingTableCreator {
         let mut idx = 0;
 
         // Pull from stream until time runs out or we have enough items
-        for item in self.stream.by_ref() {
+        for (mut item, formatter) in self.stream.by_ref() {
+            if let Some(formatter) = formatter {
+                item = formatter.format(item);
+            }
+
             batch.push(item);
             idx += 1;
 
@@ -577,3 +581,70 @@ fn load_theme_from_config(config: &Config) -> TableTheme {
         _ => nu_table::TableTheme::rounded(),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use nu_protocol::ValueFormatter;
+
+    use super::*;
+
+    #[test]
+    fn list_stream_value_formatters() {
+        let span = Span::test_data();
+        let ctrlc = None;
+        let config = Config {
+            use_ansi_coloring: false,
+            ..<_>::default()
+        };
+
+        let hex_formatter = ValueFormatter::from_fn(|value| {
+            let (value, span) = match value {
+                Value::Int { val, span } => (val, span),
+                _ => return value,
+            };
+            let value = format!("0x{:016x}", value);
+
+            Value::string(value, span)
+        });
+
+        let stream = ListStream::from_stream(
+            [
+                (Value::int(42, span), Some(hex_formatter.clone())),
+                (Value::int(777, span), None),
+                (Value::int(-1, span), Some(hex_formatter)),
+            ]
+            .into_iter(),
+            ctrlc.clone(),
+        );
+
+        let paging_table_creator = PagingTableCreator {
+            head: span,
+            stream,
+            ctrlc,
+            config,
+            row_offset: 0,
+            width_param: Some(80),
+        };
+
+        let mut output = Vec::new();
+
+        for chunk in paging_table_creator {
+            let chunk = chunk.unwrap();
+
+            output.extend(chunk);
+        }
+
+        let output = String::from_utf8(output).unwrap();
+
+        assert_eq!(
+            output,
+            concat!(
+                "╭───┬────────────────────╮\n",
+                "│ 0 │ 0x000000000000002a │\n",
+                "│ 1 │                777 │\n",
+                "│ 2 │ 0xffffffffffffffff │\n",
+                "╰───┴────────────────────╯"
+            )
+        )
+    }
+}

crates/nu-engine/src/eval.rs

@@ -248,7 +248,10 @@ fn eval_external(
         match exit_code {
             Some(exit_code_stream) => {
                 let ctrlc = exit_code_stream.ctrlc.clone();
-                let exit_code: Vec<Value> = exit_code_stream.into_iter().collect();
+                let exit_code: Vec<Value> = exit_code_stream
+                    .into_iter()
+                    .map(|(value, _)| value)
+                    .collect();
                 if let Some(Value::Int { val: code, .. }) = exit_code.last() {
                     // if exit_code is not 0, it indicates error occured, return back Err.
                     if *code != 0 {
@@ -775,7 +778,7 @@ pub fn eval_block(
                     };
 
                     if let Some(exit_code) = exit_code {
-                        let mut v: Vec<_> = exit_code.collect();
+                        let mut v: Vec<_> = exit_code.map(|(value, _)| value).collect();
 
                         if let Some(v) = v.pop() {
                             stack.add_env_var("LAST_EXIT_CODE".into(), v);

crates/nu-protocol/src/pipeline_data.rs

@@ -4,6 +4,7 @@ use crate::{
     format_error, Config, ListStream, RawStream, ShellError, Span, Value,
 };
 use nu_utils::{stderr_write_all_and_flush, stdout_write_all_and_flush};
+use std::fmt;
 use std::sync::{atomic::AtomicBool, Arc};
 
 /// The foundational abstraction for input and output to commands
@@ -94,7 +95,7 @@ impl PipelineData {
             PipelineData::Value(Value::Nothing { .. }, ..) => Value::nothing(span),
             PipelineData::Value(v, ..) => v,
             PipelineData::ListStream(s, ..) => Value::List {
-                vals: s.collect(),
+                vals: s.map(|(value, _)| value).collect(),
                 span, // FIXME?
             },
             PipelineData::ExternalStream {
@@ -222,7 +223,7 @@ impl PipelineData {
         match self {
             // FIXME: there are probably better ways of doing this
             PipelineData::ListStream(stream, ..) => Value::List {
-                vals: stream.collect(),
+                vals: stream.map(|(value, _)| value).collect(),
                 span: head,
             }
             .follow_cell_path(cell_path, insensitive),
@@ -240,7 +241,7 @@ impl PipelineData {
         match self {
             // FIXME: there are probably better ways of doing this
             PipelineData::ListStream(stream, ..) => Value::List {
-                vals: stream.collect(),
+                vals: stream.map(|(value, _)| value).collect(),
                 span: head,
             }
             .upsert_cell_path(cell_path, callback),
@@ -263,7 +264,9 @@ impl PipelineData {
             PipelineData::Value(Value::List { vals, .. }, ..) => {
                 Ok(vals.into_iter().map(f).into_pipeline_data(ctrlc))
             }
-            PipelineData::ListStream(stream, ..) => Ok(stream.map(f).into_pipeline_data(ctrlc)),
+            PipelineData::ListStream(stream, ..) => Ok(stream
+                .map(move |(value, _)| f(value))
+                .into_pipeline_data(ctrlc)),
             PipelineData::ExternalStream { stdout: None, .. } => {
                 Ok(PipelineData::new(Span { start: 0, end: 0 }))
             }
@@ -315,9 +318,9 @@ impl PipelineData {
             PipelineData::Value(Value::List { vals, .. }, ..) => {
                 Ok(vals.into_iter().flat_map(f).into_pipeline_data(ctrlc))
             }
-            PipelineData::ListStream(stream, ..) => {
+            PipelineData::ListStream(stream, ..) => Ok(stream
-                Ok(stream.flat_map(f).into_pipeline_data(ctrlc))
+                .flat_map(move |(value, _)| f(value))
-            }
+                .into_pipeline_data(ctrlc)),
             PipelineData::ExternalStream { stdout: None, .. } => {
                 Ok(PipelineData::new(Span { start: 0, end: 0 }))
             }
@@ -366,7 +369,10 @@ impl PipelineData {
             PipelineData::Value(Value::List { vals, .. }, ..) => {
                 Ok(vals.into_iter().filter(f).into_pipeline_data(ctrlc))
             }
-            PipelineData::ListStream(stream, ..) => Ok(stream.filter(f).into_pipeline_data(ctrlc)),
+            PipelineData::ListStream(stream, ..) => Ok(stream
+                .filter(move |(value, _)| f(value))
+                .map(|(value, _)| value)
+                .into_pipeline_data(ctrlc)),
             PipelineData::ExternalStream { stdout: None, .. } => {
                 Ok(PipelineData::new(Span { start: 0, end: 0 }))
             }
@@ -510,6 +516,31 @@ impl PipelineData {
     }
 }
 
+#[derive(Clone)]
+pub struct ValueFormatter(Arc<dyn Fn(Value) -> Value + Send + Sync>);
+
+impl ValueFormatter {
+    pub fn from_fn<F>(f: F) -> Self
+    where
+        F: Fn(Value) -> Value,
+        F: Send + Sync + 'static,
+    {
+        Self(Arc::new(f))
+    }
+
+    pub fn format(&self, value: Value) -> Value {
+        self.0(value)
+    }
+}
+
+impl fmt::Debug for ValueFormatter {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_tuple("PipelineDataFormatter")
+            .field(&"<formatter>")
+            .finish()
+    }
+}
+
 pub struct PipelineIterator(PipelineData);
 
 impl IntoIterator for PipelineData {
@@ -522,7 +553,7 @@ impl IntoIterator for PipelineData {
             PipelineData::Value(Value::List { vals, .. }, metadata) => {
                 PipelineIterator(PipelineData::ListStream(
                     ListStream {
-                        stream: Box::new(vals.into_iter()),
+                        stream: Box::new(vals.into_iter().map(|v| (v, None))),
                         ctrlc: None,
                     },
                     metadata,
@@ -532,14 +563,14 @@ impl IntoIterator for PipelineData {
                 match val.into_range_iter(None) {
                     Ok(iter) => PipelineIterator(PipelineData::ListStream(
                         ListStream {
-                            stream: Box::new(iter),
+                            stream: Box::new(iter.map(|v| (v, None))),
                             ctrlc: None,
                         },
                         metadata,
                     )),
                     Err(error) => PipelineIterator(PipelineData::ListStream(
                         ListStream {
-                            stream: Box::new(std::iter::once(Value::Error { error })),
+                            stream: Box::new(std::iter::once((Value::Error { error }, None))),
                             ctrlc: None,
                         },
                         metadata,
@@ -558,7 +589,7 @@ impl Iterator for PipelineIterator {
         match &mut self.0 {
             PipelineData::Value(Value::Nothing { .. }, ..) => None,
             PipelineData::Value(v, ..) => Some(std::mem::take(v)),
-            PipelineData::ListStream(stream, ..) => stream.next(),
+            PipelineData::ListStream(stream, ..) => stream.next().map(|(value, _)| value),
             PipelineData::ExternalStream { stdout: None, .. } => None,
             PipelineData::ExternalStream {
                 stdout: Some(stream),
@@ -577,10 +608,12 @@ pub trait IntoPipelineData {
 
 impl<V> IntoPipelineData for V
 where
-    V: Into<Value>,
+    V: Into<(Value, Option<ValueFormatter>)>,
 {
     fn into_pipeline_data(self) -> PipelineData {
-        PipelineData::Value(self.into(), None)
+        let (value, _formatter) = self.into();
+
+        PipelineData::Value(value, None)
     }
 }
 
@@ -597,7 +630,7 @@ impl<I> IntoInterruptiblePipelineData for I
 where
     I: IntoIterator + Send + 'static,
     I::IntoIter: Send + 'static,
-    <I::IntoIter as Iterator>::Item: Into<Value>,
+    <I::IntoIter as Iterator>::Item: Into<(Value, Option<ValueFormatter>)>,
 {
     fn into_pipeline_data(self, ctrlc: Option<Arc<AtomicBool>>) -> PipelineData {
         PipelineData::ListStream(

crates/nu-protocol/src/value/mod.rs

@@ -7,8 +7,8 @@ mod unit;
 
 use crate::ast::Operator;
 use crate::ast::{CellPath, PathMember};
-use crate::ShellError;
 use crate::{did_you_mean, BlockId, Config, Span, Spanned, Type, VarId};
+use crate::{ShellError, ValueFormatter};
 use byte_unit::ByteUnit;
 use chrono::{DateTime, Duration, FixedOffset};
 use chrono_humanize::HumanTime;
@@ -1127,6 +1127,12 @@ impl Default for Value {
     }
 }
 
+impl From<Value> for (Value, Option<ValueFormatter>) {
+    fn from(val: Value) -> Self {
+        (val, None)
+    }
+}
+
 impl PartialOrd for Value {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         // Compare two floating point numbers. The decision interval for equality is dynamically

crates/nu-protocol/src/value/stream.rs

@@ -171,23 +171,25 @@ impl Iterator for RawStream {
 /// Like other iterators in Rust, observing values from this stream will drain the items as you view them
 /// and the stream cannot be replayed.
 pub struct ListStream {
-    pub stream: Box<dyn Iterator<Item = Value> + Send + 'static>,
+    pub stream: Box<dyn Iterator<Item = (Value, Option<ValueFormatter>)> + Send + 'static>,
     pub ctrlc: Option<Arc<AtomicBool>>,
 }
 
 impl ListStream {
     pub fn into_string(self, separator: &str, config: &Config) -> String {
-        self.map(|x: Value| x.into_string(", ", config))
+        self.map(|(x, _): (Value, _)| x.into_string(", ", config))
             .collect::<Vec<String>>()
             .join(separator)
     }
 
     pub fn from_stream(
-        input: impl Iterator<Item = Value> + Send + 'static,
+        input: impl Iterator<Item = impl Into<(Value, Option<ValueFormatter>)> + 'static>
+            + Send
+            + 'static,
         ctrlc: Option<Arc<AtomicBool>>,
     ) -> ListStream {
         ListStream {
-            stream: Box::new(input),
+            stream: Box::new(input.map(Into::into)),
             ctrlc,
         }
     }
@@ -200,7 +202,7 @@ impl Debug for ListStream {
 }
 
 impl Iterator for ListStream {
-    type Item = Value;
+    type Item = (Value, Option<ValueFormatter>);
 
     fn next(&mut self) -> Option<Self::Item> {
         if let Some(ctrlc) = &self.ctrlc {