From 0c3ea636fbad037690b6768dede130cab4ec5445 Mon Sep 17 00:00:00 2001 From: JT <547158+jntrnr@users.noreply.github.com> Date: Fri, 25 Feb 2022 14:51:31 -0500 Subject: [PATCH] Add support for stderr and exit code (#4647) --- crates/nu-cli/src/util.rs | 2 +- .../nu-command/src/conversions/into/binary.rs | 2 +- .../nu-command/src/conversions/into/string.rs | 2 +- .../nu-command/src/core_commands/describe.rs | 2 +- crates/nu-command/src/default_context.rs | 1 + crates/nu-command/src/filesystem/open.rs | 12 +- crates/nu-command/src/filters/columns.rs | 2 +- crates/nu-command/src/filters/each.rs | 2 +- crates/nu-command/src/filters/length.rs | 2 +- crates/nu-command/src/filters/lines.rs | 2 +- crates/nu-command/src/filters/par_each.rs | 2 +- crates/nu-command/src/filters/skip/skip_.rs | 7 +- crates/nu-command/src/filters/wrap.rs | 2 +- crates/nu-command/src/network/fetch.rs | 10 +- crates/nu-command/src/network/post.rs | 10 +- crates/nu-command/src/strings/decode.rs | 2 +- crates/nu-command/src/system/complete.rs | 100 ++++++++++++++++ crates/nu-command/src/system/mod.rs | 2 + crates/nu-command/src/system/run_external.rs | 112 ++++++++++++++++-- crates/nu-command/src/viewers/table.rs | 44 ++++--- crates/nu-parser/src/known_external.rs | 7 +- crates/nu-protocol/src/pipeline_data.rs | 24 ++-- src/main.rs | 12 +- 23 files changed, 290 insertions(+), 73 deletions(-) create mode 100644 crates/nu-command/src/system/complete.rs diff --git a/crates/nu-cli/src/util.rs b/crates/nu-cli/src/util.rs index ee8cf8f14d..0674c872b3 100644 --- a/crates/nu-cli/src/util.rs +++ b/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()?); } diff --git a/crates/nu-command/src/conversions/into/binary.rs b/crates/nu-command/src/conversions/into/binary.rs index 6c635f49da..6f3094fb77 100644 --- a/crates/nu-command/src/conversions/into/binary.rs +++ b/crates/nu-command/src/conversions/into/binary.rs @@ -99,7 +99,7 @@ fn into_binary( let column_paths: Vec = 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 { diff --git a/crates/nu-command/src/conversions/into/string.rs b/crates/nu-command/src/conversions/into/string.rs index e57b114726..76863fb3d1 100644 --- a/crates/nu-command/src/conversions/into/string.rs +++ b/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 { diff --git a/crates/nu-command/src/core_commands/describe.rs b/crates/nu-command/src/core_commands/describe.rs index 419b1a6d82..7f18c2bfa1 100644 --- a/crates/nu-command/src/core_commands/describe.rs +++ b/crates/nu-command/src/core_commands/describe.rs @@ -28,7 +28,7 @@ impl Command for Describe { input: PipelineData, ) -> Result { let head = call.head; - if matches!(input, PipelineData::RawStream(..)) { + if matches!(input, PipelineData::ExternalStream { .. }) { Ok(PipelineData::Value( Value::string("raw input", call.head), None, diff --git a/crates/nu-command/src/default_context.rs b/crates/nu-command/src/default_context.rs index 50c410a220..a71b8247c1 100644 --- a/crates/nu-command/src/default_context.rs +++ b/crates/nu-command/src/default_context.rs @@ -132,6 +132,7 @@ pub fn create_default_context(cwd: impl AsRef) -> EngineState { // System bind_command! { Benchmark, + Complete, Exec, External, Ps, diff --git a/crates/nu-command/src/filesystem/open.rs b/crates/nu-command/src/filesystem/open.rs index 8521ddcecd..c4c00f23ea 100644 --- a/crates/nu-command/src/filesystem/open.rs +++ b/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( - RawStream::new( + let output = PipelineData::ExternalStream { + stdout: RawStream::new( Box::new(BufferedReader { input: buf_reader }), ctrlc, call_span, ), - call_span, - None, - ); + stderr: None, + exit_code: None, + span: call_span, + metadata: None, + }; let ext = if raw { None diff --git a/crates/nu-command/src/filters/columns.rs b/crates/nu-command/src/filters/columns.rs index f6ece34815..6d3e45f780 100644 --- a/crates/nu-command/src/filters/columns.rs +++ b/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()) diff --git a/crates/nu-command/src/filters/each.rs b/crates/nu-command/src/filters/each.rs index 3d3112325b..495e6c56cc 100644 --- a/crates/nu-command/src/filters/each.rs +++ b/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)| { diff --git a/crates/nu-command/src/filters/length.rs b/crates/nu-command/src/filters/length.rs index 05c321166d..687459232c 100644 --- a/crates/nu-command/src/filters/length.rs +++ b/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()) diff --git a/crates/nu-command/src/filters/lines.rs b/crates/nu-command/src/filters/lines.rs index 24c8acc86d..90c32c7187 100644 --- a/crates/nu-command/src/filters/lines.rs +++ b/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 diff --git a/crates/nu-command/src/filters/par_each.rs b/crates/nu-command/src/filters/par_each.rs index 602cd2907a..6f1542a978 100644 --- a/crates/nu-command/src/filters/par_each.rs +++ b/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)| { diff --git a/crates/nu-command/src/filters/skip/skip_.rs b/crates/nu-command/src/filters/skip/skip_.rs index 59ca04a9c2..bd0d1722a3 100644 --- a/crates/nu-command/src/filters/skip/skip_.rs +++ b/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![]; diff --git a/crates/nu-command/src/filters/wrap.rs b/crates/nu-command/src/filters/wrap.rs index dae73e239f..947a09077a 100644 --- a/crates/nu-command/src/filters/wrap.rs +++ b/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, diff --git a/crates/nu-command/src/network/fetch.rs b/crates/nu-command/src/network/fetch.rs index 84a72614a9..d21de60693 100644 --- a/crates/nu-command/src/network/fetch.rs +++ b/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( - RawStream::new( + PipelineData::ExternalStream { + 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 diff --git a/crates/nu-command/src/network/post.rs b/crates/nu-command/src/network/post.rs index 848c80e2b2..4cd32aef4a 100644 --- a/crates/nu-command/src/network/post.rs +++ b/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( - RawStream::new( + PipelineData::ExternalStream { + 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 diff --git a/crates/nu-command/src/strings/decode.rs b/crates/nu-command/src/strings/decode.rs index 7b54791b5f..71e5f9d24a 100644 --- a/crates/nu-command/src/strings/decode.rs +++ b/crates/nu-command/src/strings/decode.rs @@ -44,7 +44,7 @@ impl Command for Decode { let encoding: Spanned = call.req(engine_state, stack, 0)?; match input { - PipelineData::RawStream(stream, ..) => { + PipelineData::ExternalStream { stdout: stream, .. } => { let bytes: Vec = stream.into_bytes()?.item; let encoding = match Encoding::for_label(encoding.item.as_bytes()) { diff --git a/crates/nu-command/src/system/complete.rs b/crates/nu-command/src/system/complete.rs new file mode 100644 index 0000000000..5ae77360d3 --- /dev/null +++ b/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 { + 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 { + vec![Example { + description: "Run the external completion", + example: "^external arg1 | complete", + result: None, + }] + } +} diff --git a/crates/nu-command/src/system/mod.rs b/crates/nu-command/src/system/mod.rs index 76572089fb..3b49075525 100644 --- a/crates/nu-command/src/system/mod.rs +++ b/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}; diff --git a/crates/nu-command/src/system/run_external.rs b/crates/nu-command/src/system/run_external.rs index 48f47a4149..46daf6ab3f 100644 --- a/crates/nu-command/src/system/run_external.rs +++ b/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( - RawStream::new(Box::new(receiver), output_ctrlc, head), - head, - None, - )) + Ok(PipelineData::ExternalStream { + stdout: RawStream::new(Box::new(stdout_receiver), output_ctrlc.clone(), head), + stderr: Some(RawStream::new( + 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 -// It implements iterator so it can be used as a ListStream +// Receiver used for the RawStream +// It implements iterator so it can be used as a RawStream struct ChannelReceiver { rx: mpsc::Receiver>, } @@ -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, +} + +impl ValueReceiver { + pub fn new(rx: mpsc::Receiver) -> Self { + Self { rx } + } +} + +impl Iterator for ValueReceiver { + type Item = Value; + + fn next(&mut self) -> Option { + match self.rx.recv() { + Ok(v) => Some(v), + Err(_) => None, + } + } +} diff --git a/crates/nu-command/src/viewers/table.rs b/crates/nu-command/src/viewers/table.rs index 4c2f8c0702..c4c6984d36 100644 --- a/crates/nu-command/src/viewers/table.rs +++ b/crates/nu-command/src/viewers/table.rs @@ -62,21 +62,25 @@ impl Command for Table { }; match input { - PipelineData::RawStream(..) => Ok(input), - PipelineData::Value(Value::Binary { val, .. }, ..) => Ok(PipelineData::RawStream( - RawStream::new( - Box::new( - vec![Ok(format!("{}\n", nu_pretty_hex::pretty_hex(&val)) - .as_bytes() - .to_vec())] - .into_iter(), + PipelineData::ExternalStream { .. } => Ok(input), + PipelineData::Value(Value::Binary { val, .. }, ..) => { + Ok(PipelineData::ExternalStream { + stdout: RawStream::new( + Box::new( + vec![Ok(format!("{}\n", nu_pretty_hex::pretty_hex(&val)) + .as_bytes() + .to_vec())] + .into_iter(), + ), + ctrlc, + head, ), - ctrlc, - head, - ), - head, - None, - )), + stderr: None, + exit_code: None, + span: head, + metadata: 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( - RawStream::new( + Ok(PipelineData::ExternalStream { + stdout: RawStream::new( Box::new(PagingTableCreator { row_offset, config, @@ -267,9 +271,11 @@ fn handle_row_stream( ctrlc, head, ), - head, - None, - )) + stderr: None, + exit_code: None, + span: head, + metadata: None, + }) } fn convert_to_table( diff --git a/crates/nu-parser/src/known_external.rs b/crates/nu-parser/src/known_external.rs index 391d311f3a..358a295836 100644 --- a/crates/nu-parser/src/known_external.rs +++ b/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(), diff --git a/crates/nu-protocol/src/pipeline_data.rs b/crates/nu-protocol/src/pipeline_data.rs index 534825af6b..b881646522 100644 --- a/crates/nu-protocol/src/pipeline_data.rs +++ b/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), ListStream(ListStream, Option), - RawStream(RawStream, Span, Option), + ExternalStream { + stdout: RawStream, + stderr: Option, + exit_code: Option, + span: Span, + metadata: Option, + }, } #[derive(Debug, Clone)] @@ -60,7 +66,7 @@ impl PipelineData { pub fn metadata(&self) -> Option { 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) -> 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 }, }), diff --git a/src/main.rs b/src/main.rs index 259589594c..962634e308 100644 --- a/src/main.rs +++ b/src/main.rs @@ -171,15 +171,17 @@ fn main() -> Result<()> { let stdin = std::io::stdin(); let buf_reader = BufReader::new(stdin); - PipelineData::RawStream( - RawStream::new( + PipelineData::ExternalStream { + stdout: RawStream::new( Box::new(BufferedReader::new(buf_reader)), Some(ctrlc), redirect_stdin.span, ), - redirect_stdin.span, - None, - ) + stderr: None, + exit_code: None, + span: redirect_stdin.span, + metadata: None, + } } else { PipelineData::new(Span::new(0, 0)) };