Add support for stderr and exit code (#4647)

crates/nu-cli/src/util.rs

@@ -18,7 +18,7 @@ pub fn print_pipeline_data(
 
     let stdout = std::io::stdout();
 
-    if let PipelineData::RawStream(stream, _, _) = input {
+    if let PipelineData::ExternalStream { stdout: stream, .. } = input {
         for s in stream {
             let _ = stdout.lock().write_all(s?.as_binary()?);
         }

crates/nu-command/src/conversions/into/binary.rs

@@ -99,7 +99,7 @@ fn into_binary(
     let column_paths: Vec<CellPath> = call.rest(engine_state, stack, 0)?;
 
     match input {
-        PipelineData::RawStream(stream, ..) => {
+        PipelineData::ExternalStream { stdout: stream, .. } => {
             // TODO: in the future, we may want this to stream out, converting each to bytes
             let output = stream.into_bytes()?;
             Ok(Value::Binary {

crates/nu-command/src/conversions/into/string.rs

@@ -150,7 +150,7 @@ fn string_helper(
     }
 
     match input {
-        PipelineData::RawStream(stream, ..) => {
+        PipelineData::ExternalStream { stdout: stream, .. } => {
             // TODO: in the future, we may want this to stream out, converting each to bytes
             let output = stream.into_string()?;
             Ok(Value::String {

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

@@ -28,7 +28,7 @@ impl Command for Describe {
         input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
         let head = call.head;
-        if matches!(input, PipelineData::RawStream(..)) {
+        if matches!(input, PipelineData::ExternalStream { .. }) {
             Ok(PipelineData::Value(
                 Value::string("raw input", call.head),
                 None,

crates/nu-command/src/default_context.rs

@@ -132,6 +132,7 @@ pub fn create_default_context(cwd: impl AsRef<Path>) -> EngineState {
         // System
         bind_command! {
             Benchmark,
+            Complete,
             Exec,
             External,
             Ps,

crates/nu-command/src/filesystem/open.rs

@@ -120,15 +120,17 @@ impl Command for Open {
 
             let buf_reader = BufReader::new(file);
 
-            let output = PipelineData::RawStream(
+            let output = PipelineData::ExternalStream {
-                RawStream::new(
+                stdout: RawStream::new(
                     Box::new(BufferedReader { input: buf_reader }),
                     ctrlc,
                     call_span,
                 ),
-                call_span,
+                stderr: None,
-                None,
+                exit_code: None,
-            );
+                span: call_span,
+                metadata: None,
+            };
 
             let ext = if raw {
                 None

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

@@ -86,7 +86,7 @@ fn getcol(
             .into_iter()
             .map(move |x| Value::String { val: x, span })
             .into_pipeline_data(engine_state.ctrlc.clone())),
-        PipelineData::Value(..) | PipelineData::RawStream(..) => {
+        PipelineData::Value(..) | PipelineData::ExternalStream { .. } => {
             let cols = vec![];
             let vals = vec![];
             Ok(Value::Record { cols, vals, span }.into_pipeline_data())

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

@@ -158,7 +158,7 @@ impl Command for Each {
                     }
                 })
                 .into_pipeline_data(ctrlc)),
-            PipelineData::RawStream(stream, ..) => Ok(stream
+            PipelineData::ExternalStream { stdout: stream, .. } => Ok(stream
                 .into_iter()
                 .enumerate()
                 .map(move |(idx, x)| {

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

@@ -111,7 +111,7 @@ fn getcol(
                 .map(move |x| Value::String { val: x, span })
                 .into_pipeline_data(engine_state.ctrlc.clone()))
         }
-        PipelineData::Value(..) | PipelineData::RawStream(..) => {
+        PipelineData::Value(..) | PipelineData::ExternalStream { .. } => {
             let cols = vec![];
             let vals = vec![];
             Ok(Value::Record { cols, vals, span }.into_pipeline_data())

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

@@ -110,7 +110,7 @@ impl Command for Lines {
                 format!("Not supported input: {}", val.as_string()?),
                 call.head,
             )),
-            PipelineData::RawStream(..) => {
+            PipelineData::ExternalStream { .. } => {
                 let config = stack.get_config()?;
 
                 //FIXME: Make sure this can fail in the future to let the user

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

@@ -200,7 +200,7 @@ impl Command for ParEach {
                 .into_iter()
                 .flatten()
                 .into_pipeline_data(ctrlc)),
-            PipelineData::RawStream(stream, ..) => Ok(stream
+            PipelineData::ExternalStream { stdout: stream, .. } => Ok(stream
                 .enumerate()
                 .par_bridge()
                 .map(move |(idx, x)| {

crates/nu-command/src/filters/skip/skip_.rs

@@ -76,7 +76,12 @@ impl Command for Skip {
         let ctrlc = engine_state.ctrlc.clone();
 
         match input {
-            PipelineData::RawStream(stream, bytes_span, metadata) => {
+            PipelineData::ExternalStream {
+                stdout: stream,
+                span: bytes_span,
+                metadata,
+                ..
+            } => {
                 let mut remaining = n;
                 let mut output = vec![];
 

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

@@ -50,7 +50,7 @@ impl Command for Wrap {
                     span,
                 })
                 .into_pipeline_data(engine_state.ctrlc.clone())),
-            PipelineData::RawStream(..) => Ok(Value::Record {
+            PipelineData::ExternalStream { .. } => Ok(Value::Record {
                 cols: vec![name],
                 vals: vec![input.into_value(call.head)],
                 span,

crates/nu-command/src/network/fetch.rs

@@ -356,17 +356,19 @@ fn response_to_buffer(
 ) -> nu_protocol::PipelineData {
     let buffered_input = BufReader::new(response);
 
-    PipelineData::RawStream(
+    PipelineData::ExternalStream {
-        RawStream::new(
+        stdout: RawStream::new(
             Box::new(BufferedReader {
                 input: buffered_input,
             }),
             engine_state.ctrlc.clone(),
             span,
         ),
+        stderr: None,
+        exit_code: None,
         span,
-        None,
+        metadata: None,
-    )
+    }
 }
 
 // Only panics if the user agent is invalid but we define it statically so either

crates/nu-command/src/network/post.rs

@@ -378,17 +378,19 @@ fn response_to_buffer(
 ) -> nu_protocol::PipelineData {
     let buffered_input = BufReader::new(response);
 
-    PipelineData::RawStream(
+    PipelineData::ExternalStream {
-        RawStream::new(
+        stdout: RawStream::new(
             Box::new(BufferedReader {
                 input: buffered_input,
             }),
             engine_state.ctrlc.clone(),
             span,
         ),
+        stderr: None,
+        exit_code: None,
         span,
-        None,
+        metadata: None,
-    )
+    }
 }
 // Only panics if the user agent is invalid but we define it statically so either
 // it always or never fails

crates/nu-command/src/strings/decode.rs

@@ -44,7 +44,7 @@ impl Command for Decode {
         let encoding: Spanned<String> = call.req(engine_state, stack, 0)?;
 
         match input {
-            PipelineData::RawStream(stream, ..) => {
+            PipelineData::ExternalStream { stdout: stream, .. } => {
                 let bytes: Vec<u8> = stream.into_bytes()?.item;
 
                 let encoding = match Encoding::for_label(encoding.item.as_bytes()) {

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

@@ -0,0 +1,100 @@
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, Example, IntoPipelineData, PipelineData, ShellError, Signature, Value,
+};
+
+#[derive(Clone)]
+pub struct Complete;
+
+impl Command for Complete {
+    fn name(&self) -> &str {
+        "complete"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("complete").category(Category::System)
+    }
+
+    fn usage(&self) -> &str {
+        "Complete the external piped in, collecting outputs and exit code"
+    }
+
+    fn run(
+        &self,
+        _engine_state: &EngineState,
+        _stack: &mut Stack,
+        call: &Call,
+        input: PipelineData,
+    ) -> Result<nu_protocol::PipelineData, nu_protocol::ShellError> {
+        match input {
+            PipelineData::ExternalStream {
+                stdout,
+                stderr,
+                exit_code,
+                ..
+            } => {
+                let mut cols = vec!["stdout".to_string()];
+                let mut vals = vec![];
+
+                let stdout = stdout.into_bytes()?;
+                if let Ok(st) = String::from_utf8(stdout.item.clone()) {
+                    vals.push(Value::String {
+                        val: st,
+                        span: stdout.span,
+                    })
+                } else {
+                    vals.push(Value::Binary {
+                        val: stdout.item,
+                        span: stdout.span,
+                    })
+                };
+
+                if let Some(stderr) = stderr {
+                    cols.push("stderr".to_string());
+                    let stderr = stderr.into_bytes()?;
+                    if let Ok(st) = String::from_utf8(stderr.item.clone()) {
+                        vals.push(Value::String {
+                            val: st,
+                            span: stderr.span,
+                        })
+                    } else {
+                        vals.push(Value::Binary {
+                            val: stderr.item,
+                            span: stderr.span,
+                        })
+                    };
+                }
+
+                if let Some(exit_code) = exit_code {
+                    let mut v: Vec<_> = exit_code.collect();
+
+                    if let Some(v) = v.pop() {
+                        cols.push("exit_code".to_string());
+                        vals.push(v);
+                    }
+                }
+
+                Ok(Value::Record {
+                    cols,
+                    vals,
+                    span: call.head,
+                }
+                .into_pipeline_data())
+            }
+            _ => Err(ShellError::SpannedLabeledError(
+                "Complete only works with external streams".to_string(),
+                "complete only works on external streams".to_string(),
+                call.head,
+            )),
+        }
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![Example {
+            description: "Run the external completion",
+            example: "^external arg1 | complete",
+            result: None,
+        }]
+    }
+}

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

@@ -1,4 +1,5 @@
 mod benchmark;
+mod complete;
 mod exec;
 mod ps;
 mod run_external;
@@ -6,6 +7,7 @@ mod sys;
 mod which_;
 
 pub use benchmark::Benchmark;
+pub use complete::Complete;
 pub use exec::Exec;
 pub use ps::Ps;
 pub use run_external::{External, ExternalCommand};

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

@@ -8,7 +8,7 @@ use std::sync::mpsc;
 use nu_engine::env_to_strings;
 use nu_protocol::engine::{EngineState, Stack};
 use nu_protocol::{ast::Call, engine::Command, ShellError, Signature, SyntaxShape, Value};
-use nu_protocol::{Category, Example, PipelineData, RawStream, Span, Spanned};
+use nu_protocol::{Category, Example, ListStream, PipelineData, RawStream, Span, Spanned};
 
 use itertools::Itertools;
 
@@ -192,14 +192,51 @@ impl ExternalCommand {
                 let redirect_stderr = self.redirect_stderr;
                 let span = self.name.span;
                 let output_ctrlc = ctrlc.clone();
-                let (tx, rx) = mpsc::channel();
+                let (stdout_tx, stdout_rx) = mpsc::channel();
+                let (stderr_tx, stderr_rx) = mpsc::channel();
+                let (exit_code_tx, exit_code_rx) = mpsc::channel();
 
                 std::thread::spawn(move || {
                     // If this external is not the last expression, then its output is piped to a channel
                     // and we create a ListStream that can be consumed
 
                     if redirect_stderr {
-                        let _ = child.stderr.take();
+                        let stderr = child.stderr.take().ok_or_else(|| {
+                            ShellError::ExternalCommand(
+                                "Error taking stderr from external".to_string(),
+                                "Redirects need access to stderr of an external command"
+                                    .to_string(),
+                                span,
+                            )
+                        })?;
+
+                        // Stderr is read using the Buffer reader. It will do so until there is an
+                        // error or there are no more bytes to read
+                        let mut buf_read = BufReader::with_capacity(OUTPUT_BUFFER_SIZE, stderr);
+                        while let Ok(bytes) = buf_read.fill_buf() {
+                            if bytes.is_empty() {
+                                break;
+                            }
+
+                            // The Cow generated from the function represents the conversion
+                            // from bytes to String. If no replacements are required, then the
+                            // borrowed value is a proper UTF-8 string. The Owned option represents
+                            // a string where the values had to be replaced, thus marking it as bytes
+                            let bytes = bytes.to_vec();
+                            let length = bytes.len();
+                            buf_read.consume(length);
+
+                            if let Some(ctrlc) = &ctrlc {
+                                if ctrlc.load(Ordering::SeqCst) {
+                                    break;
+                                }
+                            }
+
+                            match stderr_tx.send(bytes) {
+                                Ok(_) => continue,
+                                Err(_) => break,
+                            }
+                        }
                     }
 
                     if redirect_stdout {
@@ -234,7 +271,7 @@ impl ExternalCommand {
                                 }
                             }
 
-                            match tx.send(bytes) {
+                            match stdout_tx.send(bytes) {
                                 Ok(_) => continue,
                                 Err(_) => break,
                             }
@@ -247,16 +284,42 @@ impl ExternalCommand {
                             err.to_string(),
                             span,
                         )),
-                        Ok(_) => Ok(()),
+                        Ok(x) => {
+                            if let Some(code) = x.code() {
+                                let _ = exit_code_tx.send(Value::Int {
+                                    val: code as i64,
+                                    span: head,
+                                });
+                            } else if x.success() {
+                                let _ = exit_code_tx.send(Value::Int { val: 0, span: head });
+                            } else {
+                                let _ = exit_code_tx.send(Value::Int {
+                                    val: -1,
+                                    span: head,
+                                });
+                            }
+                            Ok(())
+                        }
                     }
                 });
-                let receiver = ChannelReceiver::new(rx);
+                let stdout_receiver = ChannelReceiver::new(stdout_rx);
+                let stderr_receiver = ChannelReceiver::new(stderr_rx);
+                let exit_code_receiver = ValueReceiver::new(exit_code_rx);
 
-                Ok(PipelineData::RawStream(
+                Ok(PipelineData::ExternalStream {
-                    RawStream::new(Box::new(receiver), output_ctrlc, head),
+                    stdout: RawStream::new(Box::new(stdout_receiver), output_ctrlc.clone(), head),
-                    head,
+                    stderr: Some(RawStream::new(
-                    None,
+                        Box::new(stderr_receiver),
-                ))
+                        output_ctrlc.clone(),
+                        head,
+                    )),
+                    exit_code: Some(ListStream::from_stream(
+                        Box::new(exit_code_receiver),
+                        output_ctrlc,
+                    )),
+                    span: head,
+                    metadata: None,
+                })
             }
         }
     }
@@ -452,8 +515,8 @@ fn trim_enclosing_quotes(input: &str) -> String {
     }
 }
 
-// Receiver used for the ListStream
+// Receiver used for the RawStream
-// It implements iterator so it can be used as a ListStream
+// It implements iterator so it can be used as a RawStream
 struct ChannelReceiver {
     rx: mpsc::Receiver<Vec<u8>>,
 }
@@ -474,3 +537,26 @@ impl Iterator for ChannelReceiver {
         }
     }
 }
+
+// Receiver used for the ListStream
+// It implements iterator so it can be used as a ListStream
+struct ValueReceiver {
+    rx: mpsc::Receiver<Value>,
+}
+
+impl ValueReceiver {
+    pub fn new(rx: mpsc::Receiver<Value>) -> Self {
+        Self { rx }
+    }
+}
+
+impl Iterator for ValueReceiver {
+    type Item = Value;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.rx.recv() {
+            Ok(v) => Some(v),
+            Err(_) => None,
+        }
+    }
+}

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

@@ -62,21 +62,25 @@ impl Command for Table {
         };
 
         match input {
-            PipelineData::RawStream(..) => Ok(input),
+            PipelineData::ExternalStream { .. } => Ok(input),
-            PipelineData::Value(Value::Binary { val, .. }, ..) => Ok(PipelineData::RawStream(
+            PipelineData::Value(Value::Binary { val, .. }, ..) => {
-                RawStream::new(
+                Ok(PipelineData::ExternalStream {
-                    Box::new(
+                    stdout: RawStream::new(
-                        vec![Ok(format!("{}\n", nu_pretty_hex::pretty_hex(&val))
+                        Box::new(
-                            .as_bytes()
+                            vec![Ok(format!("{}\n", nu_pretty_hex::pretty_hex(&val))
-                            .to_vec())]
+                                .as_bytes()
-                        .into_iter(),
+                                .to_vec())]
+                            .into_iter(),
+                        ),
+                        ctrlc,
+                        head,
                     ),
-                    ctrlc,
+                    stderr: None,
-                    head,
+                    exit_code: None,
-                ),
+                    span: head,
-                head,
+                    metadata: None,
-                None,
+                })
-            )),
+            }
             PipelineData::Value(Value::List { vals, .. }, metadata) => handle_row_stream(
                 engine_state,
                 stack,
@@ -255,8 +259,8 @@ fn handle_row_stream(
 
     let head = call.head;
 
-    Ok(PipelineData::RawStream(
+    Ok(PipelineData::ExternalStream {
-        RawStream::new(
+        stdout: RawStream::new(
             Box::new(PagingTableCreator {
                 row_offset,
                 config,
@@ -267,9 +271,11 @@ fn handle_row_stream(
             ctrlc,
             head,
         ),
-        head,
+        stderr: None,
-        None,
+        exit_code: None,
-    ))
+        span: head,
+        metadata: None,
+    })
 }
 
 fn convert_to_table(

crates/nu-parser/src/known_external.rs

@@ -39,6 +39,9 @@ impl Command for KnownExternal {
         let call_span = call.span();
         let contents = engine_state.get_span_contents(&call_span);
 
+        let redirect_stdout = call.redirect_stdout;
+        let redirect_stderr = call.redirect_stderr;
+
         let (lexed, _) = crate::lex(contents, call_span.start, &[], &[], true);
 
         let spans: Vec<_> = lexed.into_iter().map(|x| x.span).collect();
@@ -61,7 +64,7 @@ impl Command for KnownExternal {
                     call.positional.push(arg.clone())
                 }
 
-                if call.redirect_stdout {
+                if redirect_stdout {
                     call.named.push((
                         Spanned {
                             item: "redirect-stdout".into(),
@@ -71,7 +74,7 @@ impl Command for KnownExternal {
                     ))
                 }
 
-                if call.redirect_stderr {
+                if redirect_stderr {
                     call.named.push((
                         Spanned {
                             item: "redirect-stderr".into(),

crates/nu-protocol/src/pipeline_data.rs

@@ -35,7 +35,13 @@ use crate::{ast::PathMember, Config, ListStream, RawStream, ShellError, Span, Va
 pub enum PipelineData {
     Value(Value, Option<PipelineMetadata>),
     ListStream(ListStream, Option<PipelineMetadata>),
-    RawStream(RawStream, Span, Option<PipelineMetadata>),
+    ExternalStream {
+        stdout: RawStream,
+        stderr: Option<RawStream>,
+        exit_code: Option<ListStream>,
+        span: Span,
+        metadata: Option<PipelineMetadata>,
+    },
 }
 
 #[derive(Debug, Clone)]
@@ -60,7 +66,7 @@ impl PipelineData {
     pub fn metadata(&self) -> Option<PipelineMetadata> {
         match self {
             PipelineData::ListStream(_, x) => x.clone(),
-            PipelineData::RawStream(_, _, x) => x.clone(),
+            PipelineData::ExternalStream { metadata: x, .. } => x.clone(),
             PipelineData::Value(_, x) => x.clone(),
         }
     }
@@ -68,7 +74,7 @@ impl PipelineData {
     pub fn set_metadata(mut self, metadata: Option<PipelineMetadata>) -> Self {
         match &mut self {
             PipelineData::ListStream(_, x) => *x = metadata,
-            PipelineData::RawStream(_, _, x) => *x = metadata,
+            PipelineData::ExternalStream { metadata: x, .. } => *x = metadata,
             PipelineData::Value(_, x) => *x = metadata,
         }
 
@@ -87,7 +93,7 @@ impl PipelineData {
                 vals: s.collect(),
                 span, // FIXME?
             },
-            PipelineData::RawStream(mut s, ..) => {
+            PipelineData::ExternalStream { stdout: mut s, .. } => {
                 let mut items = vec![];
 
                 for val in &mut s {
@@ -151,7 +157,7 @@ impl PipelineData {
         match self {
             PipelineData::Value(v, ..) => Ok(v.into_string(separator, config)),
             PipelineData::ListStream(s, ..) => Ok(s.into_string(separator, config)),
-            PipelineData::RawStream(s, ..) => {
+            PipelineData::ExternalStream { stdout: s, .. } => {
                 let mut items = vec![];
 
                 for val in s {
@@ -230,7 +236,7 @@ impl PipelineData {
                 Ok(vals.into_iter().map(f).into_pipeline_data(ctrlc))
             }
             PipelineData::ListStream(stream, ..) => Ok(stream.map(f).into_pipeline_data(ctrlc)),
-            PipelineData::RawStream(stream, ..) => {
+            PipelineData::ExternalStream { stdout: stream, .. } => {
                 let collected = stream.into_bytes()?;
 
                 if let Ok(st) = String::from_utf8(collected.clone().item) {
@@ -277,7 +283,7 @@ impl PipelineData {
             PipelineData::ListStream(stream, ..) => {
                 Ok(stream.flat_map(f).into_pipeline_data(ctrlc))
             }
-            PipelineData::RawStream(stream, ..) => {
+            PipelineData::ExternalStream { stdout: stream, .. } => {
                 let collected = stream.into_bytes()?;
 
                 if let Ok(st) = String::from_utf8(collected.clone().item) {
@@ -318,7 +324,7 @@ impl PipelineData {
                 Ok(vals.into_iter().filter(f).into_pipeline_data(ctrlc))
             }
             PipelineData::ListStream(stream, ..) => Ok(stream.filter(f).into_pipeline_data(ctrlc)),
-            PipelineData::RawStream(stream, ..) => {
+            PipelineData::ExternalStream { stdout: stream, .. } => {
                 let collected = stream.into_bytes()?;
 
                 if let Ok(st) = String::from_utf8(collected.clone().item) {
@@ -408,7 +414,7 @@ impl Iterator for PipelineIterator {
             PipelineData::Value(Value::Nothing { .. }, ..) => None,
             PipelineData::Value(v, ..) => Some(std::mem::take(v)),
             PipelineData::ListStream(stream, ..) => stream.next(),
-            PipelineData::RawStream(stream, ..) => stream.next().map(|x| match x {
+            PipelineData::ExternalStream { stdout: stream, .. } => stream.next().map(|x| match x {
                 Ok(x) => x,
                 Err(err) => Value::Error { error: err },
             }),

src/main.rs

@@ -171,15 +171,17 @@ fn main() -> Result<()> {
                 let stdin = std::io::stdin();
                 let buf_reader = BufReader::new(stdin);
 
-                PipelineData::RawStream(
+                PipelineData::ExternalStream {
-                    RawStream::new(
+                    stdout: RawStream::new(
                         Box::new(BufferedReader::new(buf_reader)),
                         Some(ctrlc),
                         redirect_stdin.span,
                     ),
-                    redirect_stdin.span,
+                    stderr: None,
-                    None,
+                    exit_code: None,
-                )
+                    span: redirect_stdin.span,
+                    metadata: None,
+                }
             } else {
                 PipelineData::new(Span::new(0, 0))
             };