Replace ExternalStream with new ByteStream type (#12774)

# Description
This PR introduces a `ByteStream` type which is a `Read`-able stream of
bytes. Internally, it has an enum over three different byte stream
sources:
```rust
pub enum ByteStreamSource {
    Read(Box<dyn Read + Send + 'static>),
    File(File),
    Child(ChildProcess),
}
```

This is in comparison to the current `RawStream` type, which is an
`Iterator<Item = Vec<u8>>` and has to allocate for each read chunk.

Currently, `PipelineData::ExternalStream` serves a weird dual role where
it is either external command output or a wrapper around `RawStream`.
`ByteStream` makes this distinction more clear (via `ByteStreamSource`)
and replaces `PipelineData::ExternalStream` in this PR:
```rust
pub enum PipelineData {
    Empty,
    Value(Value, Option<PipelineMetadata>),
    ListStream(ListStream, Option<PipelineMetadata>),
    ByteStream(ByteStream, Option<PipelineMetadata>),
}
```

The PR is relatively large, but a decent amount of it is just repetitive
changes.

This PR fixes #7017, fixes #10763, and fixes #12369.

This PR also improves performance when piping external commands. Nushell
should, in most cases, have competitive pipeline throughput compared to,
e.g., bash.
| Command | Before (MB/s) | After (MB/s) | Bash (MB/s) |
| -------------------------------------------------- | -------------:|
------------:| -----------:|
| `throughput \| rg 'x'` | 3059 | 3744 | 3739 |
| `throughput \| nu --testbin relay o> /dev/null` | 3508 | 8087 | 8136 |

# User-Facing Changes
- This is a breaking change for the plugin communication protocol,
because the `ExternalStreamInfo` was replaced with `ByteStreamInfo`.
Plugins now only have to deal with a single input stream, as opposed to
the previous three streams: stdout, stderr, and exit code.
- The output of `describe` has been changed for external/byte streams.
- Temporary breaking change: `bytes starts-with` no longer works with
byte streams. This is to keep the PR smaller, and `bytes ends-with`
already does not work on byte streams.
- If a process core dumped, then instead of having a `Value::Error` in
the `exit_code` column of the output returned from `complete`, it now is
a `Value::Int` with the negation of the signal number.

# After Submitting
- Update docs and book as necessary
- Release notes (e.g., plugin protocol changes)
- Adapt/convert commands to work with byte streams (high priority is
`str length`, `bytes starts-with`, and maybe `bytes ends-with`).
- Refactor the `tee` code, Devyn has already done some work on this.

---------

Co-authored-by: Devyn Cairns <devyn.cairns@gmail.com>

crates/nu-protocol/Cargo.toml

@@ -31,6 +31,10 @@ serde = { workspace = true, default-features = false }
 serde_json = { workspace = true, optional = true }
 thiserror = "1.0"
 typetag = "0.2"
+os_pipe = { workspace = true, features = ["io_safety"] }
+
+[target.'cfg(unix)'.dependencies]
+nix = { workspace = true, default-features = false, features = ["signal"] }
 
 [features]
 plugin = [

crates/nu-protocol/src/debugger/debugger_trait.rs

@@ -44,7 +44,7 @@ pub trait DebugContext: Clone + Copy + Debug {
     fn leave_element(
         engine_state: &EngineState,
         element: &PipelineElement,
-        result: &Result<(PipelineData, bool), ShellError>,
+        result: &Result<PipelineData, ShellError>,
     ) {
     }
 }
@@ -77,7 +77,7 @@ impl DebugContext for WithDebug {
     fn leave_element(
         engine_state: &EngineState,
         element: &PipelineElement,
-        result: &Result<(PipelineData, bool), ShellError>,
+        result: &Result<PipelineData, ShellError>,
     ) {
         if let Ok(mut debugger) = engine_state.debugger.lock() {
             debugger
@@ -128,7 +128,7 @@ pub trait Debugger: Send + Debug {
         &mut self,
         engine_state: &EngineState,
         element: &PipelineElement,
-        result: &Result<(PipelineData, bool), ShellError>,
+        result: &Result<PipelineData, ShellError>,
     ) {
     }
 

crates/nu-protocol/src/debugger/profiler.rs

@@ -158,7 +158,7 @@ impl Debugger for Profiler {
         &mut self,
         _engine_state: &EngineState,
         element: &PipelineElement,
-        result: &Result<(PipelineData, bool), ShellError>,
+        result: &Result<PipelineData, ShellError>,
     ) {
         if self.depth > self.max_depth {
             return;
@@ -167,12 +167,10 @@ impl Debugger for Profiler {
         let element_span = element.expr.span;
 
         let out_opt = self.collect_values.then(|| match result {
-            Ok((pipeline_data, _not_sure_what_this_is)) => match pipeline_data {
+            Ok(pipeline_data) => match pipeline_data {
                 PipelineData::Value(val, ..) => val.clone(),
                 PipelineData::ListStream(..) => Value::string("list stream", element_span),
-                PipelineData::ExternalStream { .. } => {
-                    Value::string("external stream", element_span)
-                }
+                PipelineData::ByteStream(..) => Value::string("byte stream", element_span),
                 _ => Value::nothing(element_span),
             },
             Err(e) => Value::error(e.clone(), element_span),

crates/nu-protocol/src/errors/shell_error.rs

@@ -1,5 +1,6 @@
 use miette::Diagnostic;
 use serde::{Deserialize, Serialize};
+use std::io;
 use thiserror::Error;
 
 use crate::{
@@ -1374,42 +1375,79 @@ impl ShellError {
     }
 }
 
-impl From<std::io::Error> for ShellError {
-    fn from(input: std::io::Error) -> ShellError {
-        ShellError::IOError {
-            msg: format!("{input:?}"),
+impl From<io::Error> for ShellError {
+    fn from(error: io::Error) -> ShellError {
+        if error.kind() == io::ErrorKind::Other {
+            match error.into_inner() {
+                Some(err) => match err.downcast() {
+                    Ok(err) => *err,
+                    Err(err) => Self::IOError {
+                        msg: err.to_string(),
+                    },
+                },
+                None => Self::IOError {
+                    msg: "unknown error".into(),
+                },
+            }
+        } else {
+            Self::IOError {
+                msg: error.to_string(),
+            }
         }
     }
 }
 
-impl From<Spanned<std::io::Error>> for ShellError {
-    fn from(error: Spanned<std::io::Error>) -> Self {
-        ShellError::IOErrorSpanned {
-            msg: error.item.to_string(),
-            span: error.span,
+impl From<Spanned<io::Error>> for ShellError {
+    fn from(error: Spanned<io::Error>) -> Self {
+        let Spanned { item: error, span } = error;
+        if error.kind() == io::ErrorKind::Other {
+            match error.into_inner() {
+                Some(err) => match err.downcast() {
+                    Ok(err) => *err,
+                    Err(err) => Self::IOErrorSpanned {
+                        msg: err.to_string(),
+                        span,
+                    },
+                },
+                None => Self::IOErrorSpanned {
+                    msg: "unknown error".into(),
+                    span,
+                },
+            }
+        } else {
+            Self::IOErrorSpanned {
+                msg: error.to_string(),
+                span,
+            }
         }
     }
 }
 
-impl std::convert::From<Box<dyn std::error::Error>> for ShellError {
-    fn from(input: Box<dyn std::error::Error>) -> ShellError {
+impl From<ShellError> for io::Error {
+    fn from(error: ShellError) -> Self {
+        io::Error::new(io::ErrorKind::Other, error)
+    }
+}
+
+impl From<Box<dyn std::error::Error>> for ShellError {
+    fn from(error: Box<dyn std::error::Error>) -> ShellError {
         ShellError::IOError {
-            msg: input.to_string(),
+            msg: error.to_string(),
         }
     }
 }
 
 impl From<Box<dyn std::error::Error + Send + Sync>> for ShellError {
-    fn from(input: Box<dyn std::error::Error + Send + Sync>) -> ShellError {
+    fn from(error: Box<dyn std::error::Error + Send + Sync>) -> ShellError {
         ShellError::IOError {
-            msg: format!("{input:?}"),
+            msg: format!("{error:?}"),
         }
     }
 }
 
 impl From<super::LabeledError> for ShellError {
-    fn from(value: super::LabeledError) -> Self {
-        ShellError::LabeledError(Box::new(value))
+    fn from(error: super::LabeledError) -> Self {
+        ShellError::LabeledError(Box::new(error))
     }
 }
 

crates/nu-protocol/src/eval_const.rs

@@ -317,7 +317,7 @@ impl Eval for EvalConst {
     ) -> Result<Value, ShellError> {
         // TODO: Allow debugging const eval
         // TODO: eval.rs uses call.head for the span rather than expr.span
-        Ok(eval_const_call(working_set, call, PipelineData::empty())?.into_value(span))
+        eval_const_call(working_set, call, PipelineData::empty())?.into_value(span)
     }
 
     fn eval_external_call(
@@ -339,10 +339,7 @@ impl Eval for EvalConst {
     ) -> Result<Value, ShellError> {
         // TODO: Allow debugging const eval
         let block = working_set.get_block(block_id);
-        Ok(
-            eval_const_subexpression(working_set, block, PipelineData::empty(), span)?
-                .into_value(span),
-        )
+        eval_const_subexpression(working_set, block, PipelineData::empty(), span)?.into_value(span)
     }
 
     fn regex_match(

crates/nu-protocol/src/lib.rs

@@ -11,14 +11,14 @@ mod example;
 mod id;
 mod lev_distance;
 mod module;
-mod pipeline_data;
+mod pipeline;
 #[cfg(feature = "plugin")]
 mod plugin;
+pub mod process;
 mod signature;
 pub mod span;
 mod syntax_shape;
 mod ty;
-pub mod util;
 mod value;
 
 pub use alias::*;
@@ -31,12 +31,11 @@ pub use example::*;
 pub use id::*;
 pub use lev_distance::levenshtein_distance;
 pub use module::*;
-pub use pipeline_data::*;
+pub use pipeline::*;
 #[cfg(feature = "plugin")]
 pub use plugin::*;
 pub use signature::*;
 pub use span::*;
 pub use syntax_shape::*;
 pub use ty::*;
-pub use util::BufferedReader;
 pub use value::*;

crates/nu-protocol/src/pipeline/byte_stream.rs

@@ -0,0 +1,822 @@
+use crate::{
+    process::{ChildPipe, ChildProcess, ExitStatus},
+    ErrSpan, IntoSpanned, OutDest, PipelineData, ShellError, Span, Value,
+};
+#[cfg(unix)]
+use std::os::fd::OwnedFd;
+#[cfg(windows)]
+use std::os::windows::io::OwnedHandle;
+use std::{
+    fmt::Debug,
+    fs::File,
+    io::{self, BufRead, BufReader, Cursor, ErrorKind, Read, Write},
+    process::Stdio,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+    thread,
+};
+
+/// The source of bytes for a [`ByteStream`].
+///
+/// Currently, there are only three possibilities:
+/// 1. `Read` (any `dyn` type that implements [`Read`])
+/// 2. [`File`]
+/// 3. [`ChildProcess`]
+pub enum ByteStreamSource {
+    Read(Box<dyn Read + Send + 'static>),
+    File(File),
+    Child(Box<ChildProcess>),
+}
+
+impl ByteStreamSource {
+    fn reader(self) -> Option<SourceReader> {
+        match self {
+            ByteStreamSource::Read(read) => Some(SourceReader::Read(read)),
+            ByteStreamSource::File(file) => Some(SourceReader::File(file)),
+            ByteStreamSource::Child(mut child) => child.stdout.take().map(|stdout| match stdout {
+                ChildPipe::Pipe(pipe) => SourceReader::File(convert_file(pipe)),
+                ChildPipe::Tee(tee) => SourceReader::Read(tee),
+            }),
+        }
+    }
+}
+
+enum SourceReader {
+    Read(Box<dyn Read + Send + 'static>),
+    File(File),
+}
+
+impl Read for SourceReader {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        match self {
+            SourceReader::Read(reader) => reader.read(buf),
+            SourceReader::File(file) => file.read(buf),
+        }
+    }
+}
+
+/// A potentially infinite, interruptible stream of bytes.
+///
+/// The data of a [`ByteStream`] can be accessed using one of the following methods:
+/// - [`reader`](ByteStream::reader): returns a [`Read`]-able type to get the raw bytes in the stream.
+/// - [`lines`](ByteStream::lines): splits the bytes on lines and returns an [`Iterator`]
+///   where each item is a `Result<String, ShellError>`.
+/// - [`chunks`](ByteStream::chunks): returns an [`Iterator`] of [`Value`]s where each value is either a string or binary.
+///   Try not to use this method if possible. Rather, please use [`reader`](ByteStream::reader)
+///   (or [`lines`](ByteStream::lines) if it matches the situation).
+///
+/// Additionally, there are few methods to collect a [`Bytestream`] into memory:
+/// - [`into_bytes`](ByteStream::into_bytes): collects all bytes into a [`Vec<u8>`].
+/// - [`into_string`](ByteStream::into_string): collects all bytes into a [`String`], erroring if utf-8 decoding failed.
+/// - [`into_value`](ByteStream::into_value): collects all bytes into a string [`Value`].
+///   If utf-8 decoding failed, then a binary [`Value`] is returned instead.
+///
+/// There are also a few other methods to consume all the data of a [`Bytestream`]:
+/// - [`drain`](ByteStream::drain): consumes all bytes and outputs nothing.
+/// - [`write_to`](ByteStream::write_to): writes all bytes to the given [`Write`] destination.
+/// - [`print`](ByteStream::print): a convenience wrapper around [`write_to`](ByteStream::write_to).
+///   It prints all bytes to stdout or stderr.
+///
+/// Internally, [`ByteStream`]s currently come in three flavors according to [`ByteStreamSource`].
+/// See its documentation for more information.
+pub struct ByteStream {
+    stream: ByteStreamSource,
+    span: Span,
+    ctrlc: Option<Arc<AtomicBool>>,
+    known_size: Option<u64>,
+}
+
+impl ByteStream {
+    /// Create a new [`ByteStream`] from a [`ByteStreamSource`].
+    pub fn new(stream: ByteStreamSource, span: Span, interrupt: Option<Arc<AtomicBool>>) -> Self {
+        Self {
+            stream,
+            span,
+            ctrlc: interrupt,
+            known_size: None,
+        }
+    }
+
+    /// Create a new [`ByteStream`] from a [`ByteStreamSource::Read`].
+    pub fn read(
+        reader: impl Read + Send + 'static,
+        span: Span,
+        interrupt: Option<Arc<AtomicBool>>,
+    ) -> Self {
+        Self::new(ByteStreamSource::Read(Box::new(reader)), span, interrupt)
+    }
+
+    /// Create a new [`ByteStream`] from a [`ByteStreamSource::File`].
+    pub fn file(file: File, span: Span, interrupt: Option<Arc<AtomicBool>>) -> Self {
+        Self::new(ByteStreamSource::File(file), span, interrupt)
+    }
+
+    /// Create a new [`ByteStream`] from a [`ByteStreamSource::Child`].
+    pub fn child(child: ChildProcess, span: Span) -> Self {
+        Self::new(ByteStreamSource::Child(Box::new(child)), span, None)
+    }
+
+    /// Create a new [`ByteStream`] that reads from stdin.
+    pub fn stdin(span: Span) -> Result<Self, ShellError> {
+        let stdin = os_pipe::dup_stdin().err_span(span)?;
+        let source = ByteStreamSource::File(convert_file(stdin));
+        Ok(Self::new(source, span, None))
+    }
+
+    /// Create a new [`ByteStream`] from an [`Iterator`] of bytes slices.
+    ///
+    /// The returned [`ByteStream`] will have a [`ByteStreamSource`] of `Read`.
+    pub fn from_iter<I>(iter: I, span: Span, interrupt: Option<Arc<AtomicBool>>) -> Self
+    where
+        I: IntoIterator,
+        I::IntoIter: Send + 'static,
+        I::Item: AsRef<[u8]> + Default + Send + 'static,
+    {
+        let iter = iter.into_iter();
+        let cursor = Some(Cursor::new(I::Item::default()));
+        Self::read(ReadIterator { iter, cursor }, span, interrupt)
+    }
+
+    /// Create a new [`ByteStream`] from an [`Iterator`] of [`Result`] bytes slices.
+    ///
+    /// The returned [`ByteStream`] will have a [`ByteStreamSource`] of `Read`.
+    pub fn from_result_iter<I, T>(iter: I, span: Span, interrupt: Option<Arc<AtomicBool>>) -> Self
+    where
+        I: IntoIterator<Item = Result<T, ShellError>>,
+        I::IntoIter: Send + 'static,
+        T: AsRef<[u8]> + Default + Send + 'static,
+    {
+        let iter = iter.into_iter();
+        let cursor = Some(Cursor::new(T::default()));
+        Self::read(ReadResultIterator { iter, cursor }, span, interrupt)
+    }
+
+    /// Set the known size, in number of bytes, of the [`ByteStream`].
+    pub fn with_known_size(mut self, size: Option<u64>) -> Self {
+        self.known_size = size;
+        self
+    }
+
+    /// Get a reference to the inner [`ByteStreamSource`] of the [`ByteStream`].
+    pub fn source(&self) -> &ByteStreamSource {
+        &self.stream
+    }
+
+    /// Get a mutable reference to the inner [`ByteStreamSource`] of the [`ByteStream`].
+    pub fn source_mut(&mut self) -> &mut ByteStreamSource {
+        &mut self.stream
+    }
+
+    /// Returns the [`Span`] associated with the [`ByteStream`].
+    pub fn span(&self) -> Span {
+        self.span
+    }
+
+    /// Returns the known size, in number of bytes, of the [`ByteStream`].
+    pub fn known_size(&self) -> Option<u64> {
+        self.known_size
+    }
+
+    /// Convert the [`ByteStream`] into its [`Reader`] which allows one to [`Read`] the raw bytes of the stream.
+    ///
+    /// [`Reader`] is buffered and also implements [`BufRead`].
+    ///
+    /// If the source of the [`ByteStream`] is [`ByteStreamSource::Child`] and the child has no stdout,
+    /// then the stream is considered empty and `None` will be returned.
+    pub fn reader(self) -> Option<Reader> {
+        let reader = self.stream.reader()?;
+        Some(Reader {
+            reader: BufReader::new(reader),
+            span: self.span,
+            ctrlc: self.ctrlc,
+        })
+    }
+
+    /// Convert the [`ByteStream`] into a [`Lines`] iterator where each element is a `Result<String, ShellError>`.
+    ///
+    /// There is no limit on how large each line will be. Ending new lines (`\n` or `\r\n`) are
+    /// stripped from each line. If a line fails to be decoded as utf-8, then it will become a [`ShellError`].
+    ///
+    /// If the source of the [`ByteStream`] is [`ByteStreamSource::Child`] and the child has no stdout,
+    /// then the stream is considered empty and `None` will be returned.
+    pub fn lines(self) -> Option<Lines> {
+        let reader = self.stream.reader()?;
+        Some(Lines {
+            reader: BufReader::new(reader),
+            span: self.span,
+            ctrlc: self.ctrlc,
+        })
+    }
+
+    /// Convert the [`ByteStream`] into a [`Chunks`] iterator where each element is a `Result<Value, ShellError>`.
+    ///
+    /// Each call to [`next`](Iterator::next) reads the currently available data from the byte stream source,
+    /// up to a maximum size. If the chunk of bytes, or an expected portion of it, succeeds utf-8 decoding,
+    /// then it is returned as a [`Value::String`]. Otherwise, it is turned into a [`Value::Binary`].
+    /// Any and all newlines are kept intact in each chunk.
+    ///
+    /// Where possible, prefer [`reader`](ByteStream::reader) or [`lines`](ByteStream::lines) over this method.
+    /// Those methods are more likely to be used in a semantically correct way
+    /// (and [`reader`](ByteStream::reader) is more efficient too).
+    ///
+    /// If the source of the [`ByteStream`] is [`ByteStreamSource::Child`] and the child has no stdout,
+    /// then the stream is considered empty and `None` will be returned.
+    pub fn chunks(self) -> Option<Chunks> {
+        let reader = self.stream.reader()?;
+        Some(Chunks {
+            reader: BufReader::new(reader),
+            span: self.span,
+            ctrlc: self.ctrlc,
+            leftover: Vec::new(),
+        })
+    }
+
+    /// Convert the [`ByteStream`] into its inner [`ByteStreamSource`].
+    pub fn into_source(self) -> ByteStreamSource {
+        self.stream
+    }
+
+    /// Attempt to convert the [`ByteStream`] into a [`Stdio`].
+    ///
+    /// This will succeed if the [`ByteStreamSource`] of the [`ByteStream`] is either:
+    /// - [`File`](ByteStreamSource::File)
+    /// - [`Child`](ByteStreamSource::Child) and the child has a stdout that is `Some(ChildPipe::Pipe(..))`.
+    ///
+    /// All other cases return an `Err` with the original [`ByteStream`] in it.
+    pub fn into_stdio(mut self) -> Result<Stdio, Self> {
+        match self.stream {
+            ByteStreamSource::Read(..) => Err(self),
+            ByteStreamSource::File(file) => Ok(file.into()),
+            ByteStreamSource::Child(child) => {
+                if let ChildProcess {
+                    stdout: Some(ChildPipe::Pipe(stdout)),
+                    stderr,
+                    ..
+                } = *child
+                {
+                    debug_assert!(stderr.is_none(), "stderr should not exist");
+                    Ok(stdout.into())
+                } else {
+                    self.stream = ByteStreamSource::Child(child);
+                    Err(self)
+                }
+            }
+        }
+    }
+
+    /// Attempt to convert the [`ByteStream`] into a [`ChildProcess`].
+    ///
+    /// This will only succeed if the [`ByteStreamSource`] of the [`ByteStream`] is [`Child`](ByteStreamSource::Child).
+    /// All other cases return an `Err` with the original [`ByteStream`] in it.
+    pub fn into_child(self) -> Result<ChildProcess, Self> {
+        if let ByteStreamSource::Child(child) = self.stream {
+            Ok(*child)
+        } else {
+            Err(self)
+        }
+    }
+
+    /// Collect all the bytes of the [`ByteStream`] into a [`Vec<u8>`].
+    ///
+    /// Any trailing new lines are kept in the returned [`Vec`].
+    pub fn into_bytes(self) -> Result<Vec<u8>, ShellError> {
+        // todo!() ctrlc
+        match self.stream {
+            ByteStreamSource::Read(mut read) => {
+                let mut buf = Vec::new();
+                read.read_to_end(&mut buf).err_span(self.span)?;
+                Ok(buf)
+            }
+            ByteStreamSource::File(mut file) => {
+                let mut buf = Vec::new();
+                file.read_to_end(&mut buf).err_span(self.span)?;
+                Ok(buf)
+            }
+            ByteStreamSource::Child(child) => child.into_bytes(),
+        }
+    }
+
+    /// Collect all the bytes of the [`ByteStream`] into a [`String`].
+    ///
+    /// The trailing new line (`\n` or `\r\n`), if any, is removed from the [`String`] prior to being returned.
+    ///
+    /// If utf-8 decoding fails, an error is returned.
+    pub fn into_string(self) -> Result<String, ShellError> {
+        let span = self.span;
+        let bytes = self.into_bytes()?;
+        let mut string = String::from_utf8(bytes).map_err(|_| ShellError::NonUtf8 { span })?;
+        trim_end_newline(&mut string);
+        Ok(string)
+    }
+
+    /// Collect all the bytes of the [`ByteStream`] into a [`Value`].
+    ///
+    /// If the collected bytes are successfully decoded as utf-8, then a [`Value::String`] is returned.
+    /// The trailing new line (`\n` or `\r\n`), if any, is removed from the [`String`] prior to being returned.
+    /// Otherwise, a [`Value::Binary`] is returned with any trailing new lines preserved.
+    pub fn into_value(self) -> Result<Value, ShellError> {
+        let span = self.span;
+        let bytes = self.into_bytes()?;
+        let value = match String::from_utf8(bytes) {
+            Ok(mut str) => {
+                trim_end_newline(&mut str);
+                Value::string(str, span)
+            }
+            Err(err) => Value::binary(err.into_bytes(), span),
+        };
+        Ok(value)
+    }
+
+    /// Consume and drop all bytes of the [`ByteStream`].
+    ///
+    /// If the source of the [`ByteStream`] is [`ByteStreamSource::Child`],
+    /// then the [`ExitStatus`] of the [`ChildProcess`] is returned.
+    pub fn drain(self) -> Result<Option<ExitStatus>, ShellError> {
+        match self.stream {
+            ByteStreamSource::Read(mut read) => {
+                copy_with_interrupt(&mut read, &mut io::sink(), self.span, self.ctrlc.as_deref())?;
+                Ok(None)
+            }
+            ByteStreamSource::File(_) => Ok(None),
+            ByteStreamSource::Child(child) => Ok(Some(child.wait()?)),
+        }
+    }
+
+    /// Print all bytes of the [`ByteStream`] to stdout or stderr.
+    ///
+    /// If the source of the [`ByteStream`] is [`ByteStreamSource::Child`],
+    /// then the [`ExitStatus`] of the [`ChildProcess`] is returned.
+    pub fn print(self, to_stderr: bool) -> Result<Option<ExitStatus>, ShellError> {
+        if to_stderr {
+            self.write_to(&mut io::stderr())
+        } else {
+            self.write_to(&mut io::stdout())
+        }
+    }
+
+    /// Write all bytes of the [`ByteStream`] to `dest`.
+    ///
+    /// If the source of the [`ByteStream`] is [`ByteStreamSource::Child`],
+    /// then the [`ExitStatus`] of the [`ChildProcess`] is returned.
+    pub fn write_to(self, dest: &mut impl Write) -> Result<Option<ExitStatus>, ShellError> {
+        let span = self.span;
+        let ctrlc = self.ctrlc.as_deref();
+        match self.stream {
+            ByteStreamSource::Read(mut read) => {
+                copy_with_interrupt(&mut read, dest, span, ctrlc)?;
+                Ok(None)
+            }
+            ByteStreamSource::File(mut file) => {
+                copy_with_interrupt(&mut file, dest, span, ctrlc)?;
+                Ok(None)
+            }
+            ByteStreamSource::Child(mut child) => {
+                // All `OutDest`s except `OutDest::Capture` will cause `stderr` to be `None`.
+                // Only `save`, `tee`, and `complete` set the stderr `OutDest` to `OutDest::Capture`,
+                // and those commands have proper simultaneous handling of stdout and stderr.
+                debug_assert!(child.stderr.is_none(), "stderr should not exist");
+
+                if let Some(stdout) = child.stdout.take() {
+                    match stdout {
+                        ChildPipe::Pipe(mut pipe) => {
+                            copy_with_interrupt(&mut pipe, dest, span, ctrlc)?;
+                        }
+                        ChildPipe::Tee(mut tee) => {
+                            copy_with_interrupt(&mut tee, dest, span, ctrlc)?;
+                        }
+                    }
+                }
+                Ok(Some(child.wait()?))
+            }
+        }
+    }
+
+    pub(crate) fn write_to_out_dests(
+        self,
+        stdout: &OutDest,
+        stderr: &OutDest,
+    ) -> Result<Option<ExitStatus>, ShellError> {
+        let span = self.span;
+        let ctrlc = self.ctrlc.as_deref();
+
+        match self.stream {
+            ByteStreamSource::Read(read) => {
+                write_to_out_dest(read, stdout, true, span, ctrlc)?;
+                Ok(None)
+            }
+            ByteStreamSource::File(mut file) => {
+                match stdout {
+                    OutDest::Pipe | OutDest::Capture | OutDest::Null => {}
+                    OutDest::Inherit => {
+                        copy_with_interrupt(&mut file, &mut io::stdout(), span, ctrlc)?;
+                    }
+                    OutDest::File(f) => {
+                        copy_with_interrupt(&mut file, &mut f.as_ref(), span, ctrlc)?;
+                    }
+                }
+                Ok(None)
+            }
+            ByteStreamSource::Child(mut child) => {
+                match (child.stdout.take(), child.stderr.take()) {
+                    (Some(out), Some(err)) => {
+                        // To avoid deadlocks, we must spawn a separate thread to wait on stderr.
+                        thread::scope(|s| {
+                            let err_thread = thread::Builder::new()
+                                .name("stderr writer".into())
+                                .spawn_scoped(s, || match err {
+                                    ChildPipe::Pipe(pipe) => {
+                                        write_to_out_dest(pipe, stderr, false, span, ctrlc)
+                                    }
+                                    ChildPipe::Tee(tee) => {
+                                        write_to_out_dest(tee, stderr, false, span, ctrlc)
+                                    }
+                                })
+                                .err_span(span);
+
+                            match out {
+                                ChildPipe::Pipe(pipe) => {
+                                    write_to_out_dest(pipe, stdout, true, span, ctrlc)
+                                }
+                                ChildPipe::Tee(tee) => {
+                                    write_to_out_dest(tee, stdout, true, span, ctrlc)
+                                }
+                            }?;
+
+                            if let Ok(result) = err_thread?.join() {
+                                result?;
+                            } else {
+                                // thread panicked, which should not happen
+                                debug_assert!(false)
+                            }
+
+                            Ok::<_, ShellError>(())
+                        })?;
+                    }
+                    (Some(out), None) => {
+                        // single output stream, we can consume directly
+                        write_to_out_dest(out, stdout, true, span, ctrlc)?;
+                    }
+                    (None, Some(err)) => {
+                        // single output stream, we can consume directly
+                        write_to_out_dest(err, stderr, false, span, ctrlc)?;
+                    }
+                    (None, None) => {}
+                }
+                Ok(Some(child.wait()?))
+            }
+        }
+    }
+}
+
+impl Debug for ByteStream {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ByteStream").finish()
+    }
+}
+
+impl From<ByteStream> for PipelineData {
+    fn from(stream: ByteStream) -> Self {
+        Self::ByteStream(stream, None)
+    }
+}
+
+struct ReadIterator<I>
+where
+    I: Iterator,
+    I::Item: AsRef<[u8]>,
+{
+    iter: I,
+    cursor: Option<Cursor<I::Item>>,
+}
+
+impl<I> Read for ReadIterator<I>
+where
+    I: Iterator,
+    I::Item: AsRef<[u8]>,
+{
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        while let Some(cursor) = self.cursor.as_mut() {
+            let read = cursor.read(buf)?;
+            if read == 0 {
+                self.cursor = self.iter.next().map(Cursor::new);
+            } else {
+                return Ok(read);
+            }
+        }
+        Ok(0)
+    }
+}
+
+struct ReadResultIterator<I, T>
+where
+    I: Iterator<Item = Result<T, ShellError>>,
+    T: AsRef<[u8]>,
+{
+    iter: I,
+    cursor: Option<Cursor<T>>,
+}
+
+impl<I, T> Read for ReadResultIterator<I, T>
+where
+    I: Iterator<Item = Result<T, ShellError>>,
+    T: AsRef<[u8]>,
+{
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        while let Some(cursor) = self.cursor.as_mut() {
+            let read = cursor.read(buf)?;
+            if read == 0 {
+                self.cursor = self.iter.next().transpose()?.map(Cursor::new);
+            } else {
+                return Ok(read);
+            }
+        }
+        Ok(0)
+    }
+}
+
+pub struct Reader {
+    reader: BufReader<SourceReader>,
+    span: Span,
+    ctrlc: Option<Arc<AtomicBool>>,
+}
+
+impl Reader {
+    pub fn span(&self) -> Span {
+        self.span
+    }
+}
+
+impl Read for Reader {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        if nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
+            Err(ShellError::InterruptedByUser {
+                span: Some(self.span),
+            }
+            .into())
+        } else {
+            self.reader.read(buf)
+        }
+    }
+}
+
+impl BufRead for Reader {
+    fn fill_buf(&mut self) -> io::Result<&[u8]> {
+        self.reader.fill_buf()
+    }
+
+    fn consume(&mut self, amt: usize) {
+        self.reader.consume(amt)
+    }
+}
+
+pub struct Lines {
+    reader: BufReader<SourceReader>,
+    span: Span,
+    ctrlc: Option<Arc<AtomicBool>>,
+}
+
+impl Lines {
+    pub fn span(&self) -> Span {
+        self.span
+    }
+}
+
+impl Iterator for Lines {
+    type Item = Result<String, ShellError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
+            None
+        } else {
+            let mut buf = Vec::new();
+            match self.reader.read_until(b'\n', &mut buf) {
+                Ok(0) => None,
+                Ok(_) => {
+                    let Ok(mut string) = String::from_utf8(buf) else {
+                        return Some(Err(ShellError::NonUtf8 { span: self.span }));
+                    };
+                    trim_end_newline(&mut string);
+                    Some(Ok(string))
+                }
+                Err(e) => Some(Err(e.into_spanned(self.span).into())),
+            }
+        }
+    }
+}
+
+pub struct Chunks {
+    reader: BufReader<SourceReader>,
+    span: Span,
+    ctrlc: Option<Arc<AtomicBool>>,
+    leftover: Vec<u8>,
+}
+
+impl Chunks {
+    pub fn span(&self) -> Span {
+        self.span
+    }
+}
+
+impl Iterator for Chunks {
+    type Item = Result<Value, ShellError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
+            None
+        } else {
+            match self.reader.fill_buf() {
+                Ok(buf) => {
+                    self.leftover.extend_from_slice(buf);
+                    let len = buf.len();
+                    self.reader.consume(len);
+                }
+                Err(err) => return Some(Err(err.into_spanned(self.span).into())),
+            };
+
+            if self.leftover.is_empty() {
+                return None;
+            }
+
+            match String::from_utf8(std::mem::take(&mut self.leftover)) {
+                Ok(str) => Some(Ok(Value::string(str, self.span))),
+                Err(err) => {
+                    if err.utf8_error().error_len().is_some() {
+                        Some(Ok(Value::binary(err.into_bytes(), self.span)))
+                    } else {
+                        let i = err.utf8_error().valid_up_to();
+                        let mut bytes = err.into_bytes();
+                        self.leftover = bytes.split_off(i);
+                        let str = String::from_utf8(bytes).expect("valid utf8");
+                        Some(Ok(Value::string(str, self.span)))
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn trim_end_newline(string: &mut String) {
+    if string.ends_with('\n') {
+        string.pop();
+        if string.ends_with('\r') {
+            string.pop();
+        }
+    }
+}
+
+fn write_to_out_dest(
+    mut read: impl Read,
+    stream: &OutDest,
+    stdout: bool,
+    span: Span,
+    ctrlc: Option<&AtomicBool>,
+) -> Result<(), ShellError> {
+    match stream {
+        OutDest::Pipe | OutDest::Capture => return Ok(()),
+        OutDest::Null => copy_with_interrupt(&mut read, &mut io::sink(), span, ctrlc),
+        OutDest::Inherit if stdout => {
+            copy_with_interrupt(&mut read, &mut io::stdout(), span, ctrlc)
+        }
+        OutDest::Inherit => copy_with_interrupt(&mut read, &mut io::stderr(), span, ctrlc),
+        OutDest::File(file) => copy_with_interrupt(&mut read, &mut file.as_ref(), span, ctrlc),
+    }?;
+    Ok(())
+}
+
+#[cfg(unix)]
+pub(crate) fn convert_file<T: From<OwnedFd>>(file: impl Into<OwnedFd>) -> T {
+    file.into().into()
+}
+
+#[cfg(windows)]
+pub(crate) fn convert_file<T: From<OwnedHandle>>(file: impl Into<OwnedHandle>) -> T {
+    file.into().into()
+}
+
+const DEFAULT_BUF_SIZE: usize = 8192;
+
+pub fn copy_with_interrupt<R: ?Sized, W: ?Sized>(
+    reader: &mut R,
+    writer: &mut W,
+    span: Span,
+    interrupt: Option<&AtomicBool>,
+) -> Result<u64, ShellError>
+where
+    R: Read,
+    W: Write,
+{
+    if let Some(interrupt) = interrupt {
+        // #[cfg(any(target_os = "linux", target_os = "android"))]
+        // {
+        //     return crate::sys::kernel_copy::copy_spec(reader, writer);
+        // }
+        match generic_copy(reader, writer, span, interrupt) {
+            Ok(len) => {
+                writer.flush().err_span(span)?;
+                Ok(len)
+            }
+            Err(err) => {
+                let _ = writer.flush();
+                Err(err)
+            }
+        }
+    } else {
+        match io::copy(reader, writer) {
+            Ok(n) => {
+                writer.flush().err_span(span)?;
+                Ok(n)
+            }
+            Err(err) => {
+                let _ = writer.flush();
+                Err(err.into_spanned(span).into())
+            }
+        }
+    }
+}
+
+// Copied from [`std::io::copy`]
+fn generic_copy<R: ?Sized, W: ?Sized>(
+    reader: &mut R,
+    writer: &mut W,
+    span: Span,
+    interrupt: &AtomicBool,
+) -> Result<u64, ShellError>
+where
+    R: Read,
+    W: Write,
+{
+    let buf = &mut [0; DEFAULT_BUF_SIZE];
+    let mut len = 0;
+    loop {
+        if interrupt.load(Ordering::Relaxed) {
+            return Err(ShellError::InterruptedByUser { span: Some(span) });
+        }
+        let n = match reader.read(buf) {
+            Ok(0) => break,
+            Ok(n) => n,
+            Err(e) if e.kind() == ErrorKind::Interrupted => continue,
+            Err(e) => return Err(e.into_spanned(span).into()),
+        };
+        len += n;
+        writer.write_all(&buf[..n]).err_span(span)?;
+    }
+    Ok(len as u64)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_chunks<T>(data: Vec<T>) -> Chunks
+    where
+        T: AsRef<[u8]> + Default + Send + 'static,
+    {
+        let reader = ReadIterator {
+            iter: data.into_iter(),
+            cursor: Some(Cursor::new(T::default())),
+        };
+        Chunks {
+            reader: BufReader::new(SourceReader::Read(Box::new(reader))),
+            span: Span::test_data(),
+            ctrlc: None,
+            leftover: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn chunks_read_string() {
+        let data = vec!["Nushell", "が好きです"];
+        let chunks = test_chunks(data.clone());
+        let actual = chunks.collect::<Result<Vec<_>, _>>().unwrap();
+        let expected = data.into_iter().map(Value::test_string).collect::<Vec<_>>();
+        assert_eq!(expected, actual);
+    }
+
+    #[test]
+    fn chunks_read_string_split_utf8() {
+        let expected = "Nushell最高!";
+        let chunks = test_chunks(vec![&b"Nushell\xe6"[..], b"\x9c\x80\xe9", b"\xab\x98!"]);
+
+        let actual = chunks
+            .into_iter()
+            .map(|value| value.and_then(Value::into_string))
+            .collect::<Result<String, _>>()
+            .unwrap();
+
+        assert_eq!(expected, actual);
+    }
+
+    #[test]
+    fn chunks_returns_string_or_binary() {
+        let chunks = test_chunks(vec![b"Nushell".as_slice(), b"\x9c\x80\xe9abcd", b"efgh"]);
+        let actual = chunks.collect::<Result<Vec<_>, _>>().unwrap();
+        let expected = vec![
+            Value::test_string("Nushell"),
+            Value::test_binary(b"\x9c\x80\xe9abcd"),
+            Value::test_string("efgh"),
+        ];
+        assert_eq!(actual, expected)
+    }
+}

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

@@ -0,0 +1,11 @@
+pub mod byte_stream;
+pub mod list_stream;
+mod metadata;
+mod out_dest;
+mod pipeline_data;
+
+pub use byte_stream::*;
+pub use list_stream::*;
+pub use metadata::*;
+pub use out_dest::*;
+pub use pipeline_data::*;

crates/nu-protocol/src/pipeline/out_dest.rs

@@ -5,17 +5,17 @@ use std::{fs::File, io, process::Stdio, sync::Arc};
 pub enum OutDest {
     /// Redirect the stdout and/or stderr of one command as the input for the next command in the pipeline.
     ///
-    /// The output pipe will be available as the `stdout` of `PipelineData::ExternalStream`.
+    /// The output pipe will be available as the `stdout` of [`ChildProcess`](crate::process::ChildProcess).
     ///
     /// If stdout and stderr are both set to `Pipe`,
-    /// then they will combined into the `stdout` of `PipelineData::ExternalStream`.
+    /// then they will combined into the `stdout` of [`ChildProcess`](crate::process::ChildProcess).
     Pipe,
     /// Capture output to later be collected into a [`Value`](crate::Value), `Vec`, or used in some other way.
     ///
-    /// The output stream(s) will be available in the `stdout` or `stderr` of `PipelineData::ExternalStream`.
+    /// The output stream(s) will be available in the `stdout` or `stderr` of [`ChildProcess`](crate::process::ChildProcess).
     ///
     /// This is similar to `Pipe` but will never combine stdout and stderr
-    /// or place an external command's stderr into `stdout` of `PipelineData::ExternalStream`.
+    /// or place an external command's stderr into `stdout` of [`ChildProcess`](crate::process::ChildProcess).
     Capture,
     /// Ignore output.
     ///

crates/nu-protocol/src/pipeline/pipeline_data.rs

@@ -0,0 +1,725 @@
+use crate::{
+    ast::{Call, PathMember},
+    engine::{EngineState, Stack},
+    process::{ChildPipe, ChildProcess, ExitStatus},
+    ByteStream, Config, ErrSpan, ListStream, OutDest, PipelineMetadata, Range, ShellError, Span,
+    Value,
+};
+use nu_utils::{stderr_write_all_and_flush, stdout_write_all_and_flush};
+use std::{
+    io::{Cursor, Read, Write},
+    sync::{atomic::AtomicBool, Arc},
+};
+
+const LINE_ENDING_PATTERN: &[char] = &['\r', '\n'];
+
+/// The foundational abstraction for input and output to commands
+///
+/// This represents either a single Value or a stream of values coming into the command or leaving a command.
+///
+/// A note on implementation:
+///
+/// We've tried a few variations of this structure. Listing these below so we have a record.
+///
+/// * We tried always assuming a stream in Nushell. This was a great 80% solution, but it had some rough edges.
+/// Namely, how do you know the difference between a single string and a list of one string. How do you know
+/// when to flatten the data given to you from a data source into the stream or to keep it as an unflattened
+/// list?
+///
+/// * We tried putting the stream into Value. This had some interesting properties as now commands "just worked
+/// on values", but lead to a few unfortunate issues.
+///
+/// The first is that you can't easily clone Values in a way that felt largely immutable. For example, if
+/// you cloned a Value which contained a stream, and in one variable drained some part of it, then the second
+/// variable would see different values based on what you did to the first.
+///
+/// To make this kind of mutation thread-safe, we would have had to produce a lock for the stream, which in
+/// practice would have meant always locking the stream before reading from it. But more fundamentally, it
+/// felt wrong in practice that observation of a value at runtime could affect other values which happen to
+/// alias the same stream. By separating these, we don't have this effect. Instead, variables could get
+/// concrete list values rather than streams, and be able to view them without non-local effects.
+///
+/// * A balance of the two approaches is what we've landed on: Values are thread-safe to pass, and we can stream
+/// them into any sources. Streams are still available to model the infinite streams approach of original
+/// Nushell.
+#[derive(Debug)]
+pub enum PipelineData {
+    Empty,
+    Value(Value, Option<PipelineMetadata>),
+    ListStream(ListStream, Option<PipelineMetadata>),
+    ByteStream(ByteStream, Option<PipelineMetadata>),
+}
+
+impl PipelineData {
+    pub fn empty() -> PipelineData {
+        PipelineData::Empty
+    }
+
+    /// create a `PipelineData::ByteStream` with proper exit_code
+    ///
+    /// It's useful to break running without raising error at user level.
+    pub fn new_external_stream_with_only_exit_code(exit_code: i32) -> PipelineData {
+        let span = Span::unknown();
+        let mut child = ChildProcess::from_raw(None, None, None, span);
+        child.set_exit_code(exit_code);
+        PipelineData::ByteStream(ByteStream::child(child, span), None)
+    }
+
+    pub fn metadata(&self) -> Option<PipelineMetadata> {
+        match self {
+            PipelineData::Empty => None,
+            PipelineData::Value(_, meta)
+            | PipelineData::ListStream(_, meta)
+            | PipelineData::ByteStream(_, meta) => meta.clone(),
+        }
+    }
+
+    pub fn set_metadata(mut self, metadata: Option<PipelineMetadata>) -> Self {
+        match &mut self {
+            PipelineData::Empty => {}
+            PipelineData::Value(_, meta)
+            | PipelineData::ListStream(_, meta)
+            | PipelineData::ByteStream(_, meta) => *meta = metadata,
+        }
+        self
+    }
+
+    pub fn is_nothing(&self) -> bool {
+        matches!(self, PipelineData::Value(Value::Nothing { .. }, ..))
+            || matches!(self, PipelineData::Empty)
+    }
+
+    /// PipelineData doesn't always have a Span, but we can try!
+    pub fn span(&self) -> Option<Span> {
+        match self {
+            PipelineData::Empty => None,
+            PipelineData::Value(value, ..) => Some(value.span()),
+            PipelineData::ListStream(stream, ..) => Some(stream.span()),
+            PipelineData::ByteStream(stream, ..) => Some(stream.span()),
+        }
+    }
+
+    pub fn into_value(self, span: Span) -> Result<Value, ShellError> {
+        match self {
+            PipelineData::Empty => Ok(Value::nothing(span)),
+            PipelineData::Value(value, ..) => Ok(value.with_span(span)),
+            PipelineData::ListStream(stream, ..) => Ok(stream.into_value()),
+            PipelineData::ByteStream(stream, ..) => stream.into_value(),
+        }
+    }
+
+    /// Writes all values or redirects all output to the current [`OutDest`]s in `stack`.
+    ///
+    /// For [`OutDest::Pipe`] and [`OutDest::Capture`], this will return the `PipelineData` as is
+    /// without consuming input and without writing anything.
+    ///
+    /// For the other [`OutDest`]s, the given `PipelineData` will be completely consumed
+    /// and `PipelineData::Empty` will be returned.
+    pub fn write_to_out_dests(
+        self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+    ) -> Result<PipelineData, ShellError> {
+        match (self, stack.stdout()) {
+            (PipelineData::ByteStream(stream, ..), stdout) => {
+                stream.write_to_out_dests(stdout, stack.stderr())?;
+            }
+            (data, OutDest::Pipe | OutDest::Capture) => return Ok(data),
+            (PipelineData::Empty, ..) => {}
+            (PipelineData::Value(..), OutDest::Null) => {}
+            (PipelineData::ListStream(stream, ..), OutDest::Null) => {
+                // we need to drain the stream in case there are external commands in the pipeline
+                stream.drain()?;
+            }
+            (PipelineData::Value(value, ..), OutDest::File(file)) => {
+                let bytes = value_to_bytes(value)?;
+                let mut file = file.as_ref();
+                file.write_all(&bytes)?;
+                file.flush()?;
+            }
+            (PipelineData::ListStream(stream, ..), OutDest::File(file)) => {
+                let mut file = file.as_ref();
+                // use BufWriter here?
+                for value in stream {
+                    let bytes = value_to_bytes(value)?;
+                    file.write_all(&bytes)?;
+                    file.write_all(b"\n")?;
+                }
+                file.flush()?;
+            }
+            (data @ (PipelineData::Value(..) | PipelineData::ListStream(..)), OutDest::Inherit) => {
+                data.print(engine_state, stack, false, false)?;
+            }
+        }
+        Ok(PipelineData::Empty)
+    }
+
+    pub fn drain(self) -> Result<Option<ExitStatus>, ShellError> {
+        match self {
+            PipelineData::Empty => Ok(None),
+            PipelineData::Value(Value::Error { error, .. }, ..) => Err(*error),
+            PipelineData::Value(..) => Ok(None),
+            PipelineData::ListStream(stream, ..) => {
+                stream.drain()?;
+                Ok(None)
+            }
+            PipelineData::ByteStream(stream, ..) => stream.drain(),
+        }
+    }
+
+    /// Try convert from self into iterator
+    ///
+    /// It returns Err if the `self` cannot be converted to an iterator.
+    pub fn into_iter_strict(self, span: Span) -> Result<PipelineIterator, ShellError> {
+        Ok(PipelineIterator(match self {
+            PipelineData::Value(value, ..) => {
+                let val_span = value.span();
+                match value {
+                    Value::List { vals, .. } => PipelineIteratorInner::ListStream(
+                        ListStream::new(vals.into_iter(), val_span, None).into_iter(),
+                    ),
+                    Value::Binary { val, .. } => PipelineIteratorInner::ListStream(
+                        ListStream::new(
+                            val.into_iter().map(move |x| Value::int(x as i64, val_span)),
+                            val_span,
+                            None,
+                        )
+                        .into_iter(),
+                    ),
+                    Value::Range { val, .. } => PipelineIteratorInner::ListStream(
+                        ListStream::new(val.into_range_iter(val_span, None), val_span, None)
+                            .into_iter(),
+                    ),
+                    // Propagate errors by explicitly matching them before the final case.
+                    Value::Error { error, .. } => return Err(*error),
+                    other => {
+                        return Err(ShellError::OnlySupportsThisInputType {
+                            exp_input_type: "list, binary, range, or byte stream".into(),
+                            wrong_type: other.get_type().to_string(),
+                            dst_span: span,
+                            src_span: val_span,
+                        })
+                    }
+                }
+            }
+            PipelineData::ListStream(stream, ..) => {
+                PipelineIteratorInner::ListStream(stream.into_iter())
+            }
+            PipelineData::Empty => {
+                return Err(ShellError::OnlySupportsThisInputType {
+                    exp_input_type: "list, binary, range, or byte stream".into(),
+                    wrong_type: "null".into(),
+                    dst_span: span,
+                    src_span: span,
+                })
+            }
+            PipelineData::ByteStream(stream, ..) => {
+                if let Some(chunks) = stream.chunks() {
+                    PipelineIteratorInner::ByteStream(chunks)
+                } else {
+                    PipelineIteratorInner::Empty
+                }
+            }
+        }))
+    }
+
+    pub fn collect_string(self, separator: &str, config: &Config) -> Result<String, ShellError> {
+        match self {
+            PipelineData::Empty => Ok(String::new()),
+            PipelineData::Value(value, ..) => Ok(value.to_expanded_string(separator, config)),
+            PipelineData::ListStream(stream, ..) => Ok(stream.into_string(separator, config)),
+            PipelineData::ByteStream(stream, ..) => stream.into_string(),
+        }
+    }
+
+    /// Retrieves string from pipeline data.
+    ///
+    /// As opposed to `collect_string` this raises error rather than converting non-string values.
+    /// The `span` will be used if `ListStream` is encountered since it doesn't carry a span.
+    pub fn collect_string_strict(
+        self,
+        span: Span,
+    ) -> Result<(String, Span, Option<PipelineMetadata>), ShellError> {
+        match self {
+            PipelineData::Empty => Ok((String::new(), span, None)),
+            PipelineData::Value(Value::String { val, .. }, metadata) => Ok((val, span, metadata)),
+            PipelineData::Value(val, ..) => Err(ShellError::TypeMismatch {
+                err_message: "string".into(),
+                span: val.span(),
+            }),
+            PipelineData::ListStream(..) => Err(ShellError::TypeMismatch {
+                err_message: "string".into(),
+                span,
+            }),
+            PipelineData::ByteStream(stream, metadata) => {
+                let span = stream.span();
+                Ok((stream.into_string()?, span, metadata))
+            }
+        }
+    }
+
+    pub fn follow_cell_path(
+        self,
+        cell_path: &[PathMember],
+        head: Span,
+        insensitive: bool,
+    ) -> Result<Value, ShellError> {
+        match self {
+            // FIXME: there are probably better ways of doing this
+            PipelineData::ListStream(stream, ..) => Value::list(stream.into_iter().collect(), head)
+                .follow_cell_path(cell_path, insensitive),
+            PipelineData::Value(v, ..) => v.follow_cell_path(cell_path, insensitive),
+            PipelineData::Empty => Err(ShellError::IncompatiblePathAccess {
+                type_name: "empty pipeline".to_string(),
+                span: head,
+            }),
+            PipelineData::ByteStream(stream, ..) => Err(ShellError::IncompatiblePathAccess {
+                type_name: "byte stream".to_string(),
+                span: stream.span(),
+            }),
+        }
+    }
+
+    /// Simplified mapper to help with simple values also. For full iterator support use `.into_iter()` instead
+    pub fn map<F>(
+        self,
+        mut f: F,
+        ctrlc: Option<Arc<AtomicBool>>,
+    ) -> Result<PipelineData, ShellError>
+    where
+        Self: Sized,
+        F: FnMut(Value) -> Value + 'static + Send,
+    {
+        match self {
+            PipelineData::Value(value, ..) => {
+                let span = value.span();
+                match value {
+                    Value::List { vals, .. } => {
+                        Ok(vals.into_iter().map(f).into_pipeline_data(span, ctrlc))
+                    }
+                    Value::Range { val, .. } => Ok(val
+                        .into_range_iter(span, ctrlc.clone())
+                        .map(f)
+                        .into_pipeline_data(span, ctrlc)),
+                    value => match f(value) {
+                        Value::Error { error, .. } => Err(*error),
+                        v => Ok(v.into_pipeline_data()),
+                    },
+                }
+            }
+            PipelineData::Empty => Ok(PipelineData::Empty),
+            PipelineData::ListStream(stream, ..) => {
+                Ok(PipelineData::ListStream(stream.map(f), None))
+            }
+            PipelineData::ByteStream(stream, ..) => {
+                // TODO: is this behavior desired / correct ?
+                let span = stream.span();
+                match String::from_utf8(stream.into_bytes()?) {
+                    Ok(mut str) => {
+                        str.truncate(str.trim_end_matches(LINE_ENDING_PATTERN).len());
+                        Ok(f(Value::string(str, span)).into_pipeline_data())
+                    }
+                    Err(err) => Ok(f(Value::binary(err.into_bytes(), span)).into_pipeline_data()),
+                }
+            }
+        }
+    }
+
+    /// Simplified flatmapper. For full iterator support use `.into_iter()` instead
+    pub fn flat_map<U, F>(
+        self,
+        mut f: F,
+        ctrlc: Option<Arc<AtomicBool>>,
+    ) -> Result<PipelineData, ShellError>
+    where
+        Self: Sized,
+        U: IntoIterator<Item = Value> + 'static,
+        <U as IntoIterator>::IntoIter: 'static + Send,
+        F: FnMut(Value) -> U + 'static + Send,
+    {
+        match self {
+            PipelineData::Empty => Ok(PipelineData::Empty),
+            PipelineData::Value(value, ..) => {
+                let span = value.span();
+                match value {
+                    Value::List { vals, .. } => {
+                        Ok(vals.into_iter().flat_map(f).into_pipeline_data(span, ctrlc))
+                    }
+                    Value::Range { val, .. } => Ok(val
+                        .into_range_iter(span, ctrlc.clone())
+                        .flat_map(f)
+                        .into_pipeline_data(span, ctrlc)),
+                    value => Ok(f(value).into_iter().into_pipeline_data(span, ctrlc)),
+                }
+            }
+            PipelineData::ListStream(stream, ..) => {
+                Ok(stream.modify(|iter| iter.flat_map(f)).into())
+            }
+            PipelineData::ByteStream(stream, ..) => {
+                // TODO: is this behavior desired / correct ?
+                let span = stream.span();
+                match String::from_utf8(stream.into_bytes()?) {
+                    Ok(mut str) => {
+                        str.truncate(str.trim_end_matches(LINE_ENDING_PATTERN).len());
+                        Ok(f(Value::string(str, span))
+                            .into_iter()
+                            .into_pipeline_data(span, ctrlc))
+                    }
+                    Err(err) => Ok(f(Value::binary(err.into_bytes(), span))
+                        .into_iter()
+                        .into_pipeline_data(span, ctrlc)),
+                }
+            }
+        }
+    }
+
+    pub fn filter<F>(
+        self,
+        mut f: F,
+        ctrlc: Option<Arc<AtomicBool>>,
+    ) -> Result<PipelineData, ShellError>
+    where
+        Self: Sized,
+        F: FnMut(&Value) -> bool + 'static + Send,
+    {
+        match self {
+            PipelineData::Empty => Ok(PipelineData::Empty),
+            PipelineData::Value(value, ..) => {
+                let span = value.span();
+                match value {
+                    Value::List { vals, .. } => {
+                        Ok(vals.into_iter().filter(f).into_pipeline_data(span, ctrlc))
+                    }
+                    Value::Range { val, .. } => Ok(val
+                        .into_range_iter(span, ctrlc.clone())
+                        .filter(f)
+                        .into_pipeline_data(span, ctrlc)),
+                    value => {
+                        if f(&value) {
+                            Ok(value.into_pipeline_data())
+                        } else {
+                            Ok(Value::nothing(span).into_pipeline_data())
+                        }
+                    }
+                }
+            }
+            PipelineData::ListStream(stream, ..) => Ok(stream.modify(|iter| iter.filter(f)).into()),
+            PipelineData::ByteStream(stream, ..) => {
+                // TODO: is this behavior desired / correct ?
+                let span = stream.span();
+                let value = match String::from_utf8(stream.into_bytes()?) {
+                    Ok(mut str) => {
+                        str.truncate(str.trim_end_matches(LINE_ENDING_PATTERN).len());
+                        Value::string(str, span)
+                    }
+                    Err(err) => Value::binary(err.into_bytes(), span),
+                };
+                if f(&value) {
+                    Ok(value.into_pipeline_data())
+                } else {
+                    Ok(Value::nothing(span).into_pipeline_data())
+                }
+            }
+        }
+    }
+
+    /// Try to catch the external command exit status and detect if it failed.
+    ///
+    /// This is useful for external commands with semicolon, we can detect errors early to avoid
+    /// commands after the semicolon running.
+    ///
+    /// Returns `self` and a flag that indicates if the external command run failed. If `self` is
+    /// not [`PipelineData::ByteStream`], the flag will be `false`.
+    ///
+    /// Currently this will consume an external command to completion.
+    pub fn check_external_failed(self) -> Result<(Self, bool), ShellError> {
+        if let PipelineData::ByteStream(stream, metadata) = self {
+            let span = stream.span();
+            match stream.into_child() {
+                Ok(mut child) => {
+                    // Only check children without stdout. This means that nothing
+                    // later in the pipeline can possibly consume output from this external command.
+                    if child.stdout.is_none() {
+                        // Note:
+                        // In run-external's implementation detail, the result sender thread
+                        // send out stderr message first, then stdout message, then exit_code.
+                        //
+                        // In this clause, we already make sure that `stdout` is None
+                        // But not the case of `stderr`, so if `stderr` is not None
+                        // We need to consume stderr message before reading external commands' exit code.
+                        //
+                        // Or we'll never have a chance to read exit_code if stderr producer produce too much stderr message.
+                        // So we consume stderr stream and rebuild it.
+                        let stderr = child
+                            .stderr
+                            .take()
+                            .map(|mut stderr| {
+                                let mut buf = Vec::new();
+                                stderr.read_to_end(&mut buf).err_span(span)?;
+                                Ok::<_, ShellError>(buf)
+                            })
+                            .transpose()?;
+
+                        let code = child.wait()?.code();
+                        let mut child = ChildProcess::from_raw(None, None, None, span);
+                        if let Some(stderr) = stderr {
+                            child.stderr = Some(ChildPipe::Tee(Box::new(Cursor::new(stderr))));
+                        }
+                        child.set_exit_code(code);
+                        let stream = ByteStream::child(child, span);
+                        Ok((PipelineData::ByteStream(stream, metadata), code != 0))
+                    } else {
+                        let stream = ByteStream::child(child, span);
+                        Ok((PipelineData::ByteStream(stream, metadata), false))
+                    }
+                }
+                Err(stream) => Ok((PipelineData::ByteStream(stream, metadata), false)),
+            }
+        } else {
+            Ok((self, false))
+        }
+    }
+
+    /// Try to convert Value from Value::Range to Value::List.
+    /// This is useful to expand Value::Range into array notation, specifically when
+    /// converting `to json` or `to nuon`.
+    /// `1..3 | to XX -> [1,2,3]`
+    pub fn try_expand_range(self) -> Result<PipelineData, ShellError> {
+        match self {
+            PipelineData::Value(v, metadata) => {
+                let span = v.span();
+                match v {
+                    Value::Range { val, .. } => {
+                        match *val {
+                            Range::IntRange(range) => {
+                                if range.is_unbounded() {
+                                    return Err(ShellError::GenericError {
+                                        error: "Cannot create range".into(),
+                                        msg: "Unbounded ranges are not allowed when converting to this format".into(),
+                                        span: Some(span),
+                                        help: Some("Consider using ranges with valid start and end point.".into()),
+                                        inner: vec![],
+                                    });
+                                }
+                            }
+                            Range::FloatRange(range) => {
+                                if range.is_unbounded() {
+                                    return Err(ShellError::GenericError {
+                                        error: "Cannot create range".into(),
+                                        msg: "Unbounded ranges are not allowed when converting to this format".into(),
+                                        span: Some(span),
+                                        help: Some("Consider using ranges with valid start and end point.".into()),
+                                        inner: vec![],
+                                    });
+                                }
+                            }
+                        }
+                        let range_values: Vec<Value> = val.into_range_iter(span, None).collect();
+                        Ok(PipelineData::Value(Value::list(range_values, span), None))
+                    }
+                    x => Ok(PipelineData::Value(x, metadata)),
+                }
+            }
+            _ => Ok(self),
+        }
+    }
+
+    /// Consume and print self data immediately.
+    ///
+    /// `no_newline` controls if we need to attach newline character to output.
+    /// `to_stderr` controls if data is output to stderr, when the value is false, the data is output to stdout.
+    pub fn print(
+        self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+        no_newline: bool,
+        to_stderr: bool,
+    ) -> Result<Option<ExitStatus>, ShellError> {
+        if let PipelineData::ByteStream(stream, ..) = self {
+            stream.print(to_stderr)
+        } else {
+            // If the table function is in the declarations, then we can use it
+            // to create the table value that will be printed in the terminal
+            if let Some(decl_id) = engine_state.table_decl_id {
+                let command = engine_state.get_decl(decl_id);
+                if command.get_block_id().is_some() {
+                    self.write_all_and_flush(engine_state, no_newline, to_stderr)?;
+                } else {
+                    let call = Call::new(Span::new(0, 0));
+                    let table = command.run(engine_state, stack, &call, self)?;
+                    table.write_all_and_flush(engine_state, no_newline, to_stderr)?;
+                }
+            } else {
+                self.write_all_and_flush(engine_state, no_newline, to_stderr)?;
+            }
+            Ok(None)
+        }
+    }
+
+    fn write_all_and_flush(
+        self,
+        engine_state: &EngineState,
+        no_newline: bool,
+        to_stderr: bool,
+    ) -> Result<(), ShellError> {
+        let config = engine_state.get_config();
+        for item in self {
+            let mut out = if let Value::Error { error, .. } = item {
+                return Err(*error);
+            } else {
+                item.to_expanded_string("\n", config)
+            };
+
+            if !no_newline {
+                out.push('\n');
+            }
+
+            if to_stderr {
+                stderr_write_all_and_flush(out)?
+            } else {
+                stdout_write_all_and_flush(out)?
+            }
+        }
+
+        Ok(())
+    }
+}
+
+enum PipelineIteratorInner {
+    Empty,
+    Value(Value),
+    ListStream(crate::list_stream::IntoIter),
+    ByteStream(crate::byte_stream::Chunks),
+}
+
+pub struct PipelineIterator(PipelineIteratorInner);
+
+impl IntoIterator for PipelineData {
+    type Item = Value;
+
+    type IntoIter = PipelineIterator;
+
+    fn into_iter(self) -> Self::IntoIter {
+        PipelineIterator(match self {
+            PipelineData::Empty => PipelineIteratorInner::Empty,
+            PipelineData::Value(value, ..) => {
+                let span = value.span();
+                match value {
+                    Value::List { vals, .. } => PipelineIteratorInner::ListStream(
+                        ListStream::new(vals.into_iter(), span, None).into_iter(),
+                    ),
+                    Value::Range { val, .. } => PipelineIteratorInner::ListStream(
+                        ListStream::new(val.into_range_iter(span, None), span, None).into_iter(),
+                    ),
+                    x => PipelineIteratorInner::Value(x),
+                }
+            }
+            PipelineData::ListStream(stream, ..) => {
+                PipelineIteratorInner::ListStream(stream.into_iter())
+            }
+            PipelineData::ByteStream(stream, ..) => stream.chunks().map_or(
+                PipelineIteratorInner::Empty,
+                PipelineIteratorInner::ByteStream,
+            ),
+        })
+    }
+}
+
+impl Iterator for PipelineIterator {
+    type Item = Value;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match &mut self.0 {
+            PipelineIteratorInner::Empty => None,
+            PipelineIteratorInner::Value(Value::Nothing { .. }, ..) => None,
+            PipelineIteratorInner::Value(v, ..) => Some(std::mem::take(v)),
+            PipelineIteratorInner::ListStream(stream, ..) => stream.next(),
+            PipelineIteratorInner::ByteStream(stream) => stream.next().map(|x| match x {
+                Ok(x) => x,
+                Err(err) => Value::error(
+                    err,
+                    Span::unknown(), //FIXME: unclear where this span should come from
+                ),
+            }),
+        }
+    }
+}
+
+pub trait IntoPipelineData {
+    fn into_pipeline_data(self) -> PipelineData;
+
+    fn into_pipeline_data_with_metadata(
+        self,
+        metadata: impl Into<Option<PipelineMetadata>>,
+    ) -> PipelineData;
+}
+
+impl<V> IntoPipelineData for V
+where
+    V: Into<Value>,
+{
+    fn into_pipeline_data(self) -> PipelineData {
+        PipelineData::Value(self.into(), None)
+    }
+
+    fn into_pipeline_data_with_metadata(
+        self,
+        metadata: impl Into<Option<PipelineMetadata>>,
+    ) -> PipelineData {
+        PipelineData::Value(self.into(), metadata.into())
+    }
+}
+
+pub trait IntoInterruptiblePipelineData {
+    fn into_pipeline_data(self, span: Span, ctrlc: Option<Arc<AtomicBool>>) -> PipelineData;
+    fn into_pipeline_data_with_metadata(
+        self,
+        span: Span,
+        ctrlc: Option<Arc<AtomicBool>>,
+        metadata: impl Into<Option<PipelineMetadata>>,
+    ) -> PipelineData;
+}
+
+impl<I> IntoInterruptiblePipelineData for I
+where
+    I: IntoIterator + Send + 'static,
+    I::IntoIter: Send + 'static,
+    <I::IntoIter as Iterator>::Item: Into<Value>,
+{
+    fn into_pipeline_data(self, span: Span, ctrlc: Option<Arc<AtomicBool>>) -> PipelineData {
+        ListStream::new(self.into_iter().map(Into::into), span, ctrlc).into()
+    }
+
+    fn into_pipeline_data_with_metadata(
+        self,
+        span: Span,
+        ctrlc: Option<Arc<AtomicBool>>,
+        metadata: impl Into<Option<PipelineMetadata>>,
+    ) -> PipelineData {
+        PipelineData::ListStream(
+            ListStream::new(self.into_iter().map(Into::into), span, ctrlc),
+            metadata.into(),
+        )
+    }
+}
+
+fn value_to_bytes(value: Value) -> Result<Vec<u8>, ShellError> {
+    let bytes = match value {
+        Value::String { val, .. } => val.into_bytes(),
+        Value::Binary { val, .. } => val,
+        Value::List { vals, .. } => {
+            let val = vals
+                .into_iter()
+                .map(Value::coerce_into_string)
+                .collect::<Result<Vec<String>, ShellError>>()?
+                .join("\n")
+                + "\n";
+
+            val.into_bytes()
+        }
+        // Propagate errors by explicitly matching them before the final case.
+        Value::Error { error, .. } => return Err(*error),
+        value => value.coerce_into_string()?.into_bytes(),
+    };
+    Ok(bytes)
+}

crates/nu-protocol/src/pipeline_data/raw_stream.rs

@@ -1,176 +0,0 @@
-use crate::*;
-use std::{
-    fmt::Debug,
-    sync::{atomic::AtomicBool, Arc},
-};
-
-pub struct RawStream {
-    pub stream: Box<dyn Iterator<Item = Result<Vec<u8>, ShellError>> + Send + 'static>,
-    pub leftover: Vec<u8>,
-    pub ctrlc: Option<Arc<AtomicBool>>,
-    pub is_binary: bool,
-    pub span: Span,
-    pub known_size: Option<u64>, // (bytes)
-}
-
-impl RawStream {
-    pub fn new(
-        stream: Box<dyn Iterator<Item = Result<Vec<u8>, ShellError>> + Send + 'static>,
-        ctrlc: Option<Arc<AtomicBool>>,
-        span: Span,
-        known_size: Option<u64>,
-    ) -> Self {
-        Self {
-            stream,
-            leftover: vec![],
-            ctrlc,
-            is_binary: false,
-            span,
-            known_size,
-        }
-    }
-
-    pub fn into_bytes(self) -> Result<Spanned<Vec<u8>>, ShellError> {
-        let mut output = vec![];
-
-        for item in self.stream {
-            if nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
-                break;
-            }
-            output.extend(item?);
-        }
-
-        Ok(Spanned {
-            item: output,
-            span: self.span,
-        })
-    }
-
-    pub fn into_string(self) -> Result<Spanned<String>, ShellError> {
-        let mut output = String::new();
-        let span = self.span;
-        let ctrlc = &self.ctrlc.clone();
-
-        for item in self {
-            if nu_utils::ctrl_c::was_pressed(ctrlc) {
-                break;
-            }
-            output.push_str(&item?.coerce_into_string()?);
-        }
-
-        Ok(Spanned { item: output, span })
-    }
-
-    pub fn chain(self, stream: RawStream) -> RawStream {
-        RawStream {
-            stream: Box::new(self.stream.chain(stream.stream)),
-            leftover: self.leftover.into_iter().chain(stream.leftover).collect(),
-            ctrlc: self.ctrlc,
-            is_binary: self.is_binary,
-            span: self.span,
-            known_size: self.known_size,
-        }
-    }
-
-    pub fn drain(self) -> Result<(), ShellError> {
-        for next in self {
-            match next {
-                Ok(val) => {
-                    if let Value::Error { error, .. } = val {
-                        return Err(*error);
-                    }
-                }
-                Err(err) => return Err(err),
-            }
-        }
-        Ok(())
-    }
-}
-impl Debug for RawStream {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("RawStream").finish()
-    }
-}
-impl Iterator for RawStream {
-    type Item = Result<Value, ShellError>;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        if nu_utils::ctrl_c::was_pressed(&self.ctrlc) {
-            return None;
-        }
-
-        // If we know we're already binary, just output that
-        if self.is_binary {
-            self.stream.next().map(|buffer| {
-                buffer.map(|mut v| {
-                    if !self.leftover.is_empty() {
-                        for b in self.leftover.drain(..).rev() {
-                            v.insert(0, b);
-                        }
-                    }
-                    Value::binary(v, self.span)
-                })
-            })
-        } else {
-            // We *may* be text. We're only going to try utf-8. Other decodings
-            // needs to be taken as binary first, then passed through `decode`.
-            if let Some(buffer) = self.stream.next() {
-                match buffer {
-                    Ok(mut v) => {
-                        if !self.leftover.is_empty() {
-                            while let Some(b) = self.leftover.pop() {
-                                v.insert(0, b);
-                            }
-                        }
-
-                        match String::from_utf8(v.clone()) {
-                            Ok(s) => {
-                                // Great, we have a complete string, let's output it
-                                Some(Ok(Value::string(s, self.span)))
-                            }
-                            Err(err) => {
-                                // Okay, we *might* have a string but we've also got some errors
-                                if v.is_empty() {
-                                    // We can just end here
-                                    None
-                                } else if v.len() > 3
-                                    && (v.len() - err.utf8_error().valid_up_to() > 3)
-                                {
-                                    // As UTF-8 characters are max 4 bytes, if we have more than that in error we know
-                                    // that it's not just a character spanning two frames.
-                                    // We now know we are definitely binary, so switch to binary and stay there.
-                                    self.is_binary = true;
-                                    Some(Ok(Value::binary(v, self.span)))
-                                } else {
-                                    // Okay, we have a tiny bit of error at the end of the buffer. This could very well be
-                                    // a character that spans two frames. Since this is the case, remove the error from
-                                    // the current frame an dput it in the leftover buffer.
-                                    self.leftover = v[err.utf8_error().valid_up_to()..].to_vec();
-
-                                    let buf = v[0..err.utf8_error().valid_up_to()].to_vec();
-
-                                    match String::from_utf8(buf) {
-                                        Ok(s) => Some(Ok(Value::string(s, self.span))),
-                                        Err(_) => {
-                                            // Something is definitely wrong. Switch to binary, and stay there
-                                            self.is_binary = true;
-                                            Some(Ok(Value::binary(v, self.span)))
-                                        }
-                                    }
-                                }
-                            }
-                        }
-                    }
-                    Err(e) => Some(Err(e)),
-                }
-            } else if !self.leftover.is_empty() {
-                let output = Ok(Value::binary(self.leftover.clone(), self.span));
-                self.leftover.clear();
-
-                Some(output)
-            } else {
-                None
-            }
-        }
-    }
-}

crates/nu-protocol/src/process/child.rs

@@ -0,0 +1,294 @@
+use crate::{
+    byte_stream::convert_file, process::ExitStatus, ErrSpan, IntoSpanned, ShellError, Span,
+};
+use nu_system::ForegroundChild;
+use os_pipe::PipeReader;
+use std::{
+    fmt::Debug,
+    io::{self, Read},
+    sync::mpsc::{self, Receiver, RecvError, TryRecvError},
+    thread,
+};
+
+#[derive(Debug)]
+enum ExitStatusFuture {
+    Finished(Result<ExitStatus, Box<ShellError>>),
+    Running(Receiver<io::Result<ExitStatus>>),
+}
+
+impl ExitStatusFuture {
+    fn wait(&mut self, span: Span) -> Result<ExitStatus, ShellError> {
+        match self {
+            ExitStatusFuture::Finished(Ok(status)) => Ok(*status),
+            ExitStatusFuture::Finished(Err(err)) => Err(err.as_ref().clone()),
+            ExitStatusFuture::Running(receiver) => {
+                let code = match receiver.recv() {
+                    Ok(Ok(status)) => Ok(status),
+                    Ok(Err(err)) => Err(ShellError::IOErrorSpanned {
+                        msg: format!("failed to get exit code: {err:?}"),
+                        span,
+                    }),
+                    Err(RecvError) => Err(ShellError::IOErrorSpanned {
+                        msg: "failed to get exit code".into(),
+                        span,
+                    }),
+                };
+
+                *self = ExitStatusFuture::Finished(code.clone().map_err(Box::new));
+
+                code
+            }
+        }
+    }
+
+    fn try_wait(&mut self, span: Span) -> Result<Option<ExitStatus>, ShellError> {
+        match self {
+            ExitStatusFuture::Finished(Ok(code)) => Ok(Some(*code)),
+            ExitStatusFuture::Finished(Err(err)) => Err(err.as_ref().clone()),
+            ExitStatusFuture::Running(receiver) => {
+                let code = match receiver.try_recv() {
+                    Ok(Ok(status)) => Ok(Some(status)),
+                    Ok(Err(err)) => Err(ShellError::IOErrorSpanned {
+                        msg: format!("failed to get exit code: {err:?}"),
+                        span,
+                    }),
+                    Err(TryRecvError::Disconnected) => Err(ShellError::IOErrorSpanned {
+                        msg: "failed to get exit code".into(),
+                        span,
+                    }),
+                    Err(TryRecvError::Empty) => Ok(None),
+                };
+
+                if let Some(code) = code.clone().transpose() {
+                    *self = ExitStatusFuture::Finished(code.map_err(Box::new));
+                }
+
+                code
+            }
+        }
+    }
+}
+
+pub enum ChildPipe {
+    Pipe(PipeReader),
+    Tee(Box<dyn Read + Send + 'static>),
+}
+
+impl Debug for ChildPipe {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ChildPipe").finish()
+    }
+}
+
+impl From<PipeReader> for ChildPipe {
+    fn from(pipe: PipeReader) -> Self {
+        Self::Pipe(pipe)
+    }
+}
+
+impl Read for ChildPipe {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        match self {
+            ChildPipe::Pipe(pipe) => pipe.read(buf),
+            ChildPipe::Tee(tee) => tee.read(buf),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct ChildProcess {
+    pub stdout: Option<ChildPipe>,
+    pub stderr: Option<ChildPipe>,
+    exit_status: ExitStatusFuture,
+    span: Span,
+}
+
+impl ChildProcess {
+    pub fn new(
+        mut child: ForegroundChild,
+        reader: Option<PipeReader>,
+        swap: bool,
+        span: Span,
+    ) -> Result<Self, ShellError> {
+        let (stdout, stderr) = if let Some(combined) = reader {
+            (Some(combined), None)
+        } else {
+            let stdout = child.as_mut().stdout.take().map(convert_file);
+            let stderr = child.as_mut().stderr.take().map(convert_file);
+
+            if swap {
+                (stderr, stdout)
+            } else {
+                (stdout, stderr)
+            }
+        };
+
+        // Create a thread to wait for the exit status.
+        let (exit_status_sender, exit_status) = mpsc::channel();
+
+        thread::Builder::new()
+            .name("exit status waiter".into())
+            .spawn(move || exit_status_sender.send(child.wait().map(Into::into)))
+            .err_span(span)?;
+
+        Ok(Self::from_raw(stdout, stderr, Some(exit_status), span))
+    }
+
+    pub fn from_raw(
+        stdout: Option<PipeReader>,
+        stderr: Option<PipeReader>,
+        exit_status: Option<Receiver<io::Result<ExitStatus>>>,
+        span: Span,
+    ) -> Self {
+        Self {
+            stdout: stdout.map(Into::into),
+            stderr: stderr.map(Into::into),
+            exit_status: exit_status
+                .map(ExitStatusFuture::Running)
+                .unwrap_or(ExitStatusFuture::Finished(Ok(ExitStatus::Exited(0)))),
+            span,
+        }
+    }
+
+    pub fn set_exit_code(&mut self, exit_code: i32) {
+        self.exit_status = ExitStatusFuture::Finished(Ok(ExitStatus::Exited(exit_code)));
+    }
+
+    pub fn span(&self) -> Span {
+        self.span
+    }
+
+    pub fn into_bytes(mut self) -> Result<Vec<u8>, ShellError> {
+        if self.stderr.is_some() {
+            debug_assert!(false, "stderr should not exist");
+            return Err(ShellError::IOErrorSpanned {
+                msg: "internal error".into(),
+                span: self.span,
+            });
+        }
+
+        let bytes = if let Some(stdout) = self.stdout {
+            collect_bytes(stdout).err_span(self.span)?
+        } else {
+            Vec::new()
+        };
+
+        // TODO: check exit_status
+        self.exit_status.wait(self.span)?;
+
+        Ok(bytes)
+    }
+
+    pub fn wait(mut self) -> Result<ExitStatus, ShellError> {
+        if let Some(stdout) = self.stdout.take() {
+            let stderr = self
+                .stderr
+                .take()
+                .map(|stderr| {
+                    thread::Builder::new()
+                        .name("stderr consumer".into())
+                        .spawn(move || consume_pipe(stderr))
+                })
+                .transpose()
+                .err_span(self.span)?;
+
+            let res = consume_pipe(stdout);
+
+            if let Some(handle) = stderr {
+                handle
+                    .join()
+                    .map_err(|e| match e.downcast::<io::Error>() {
+                        Ok(io) => ShellError::from((*io).into_spanned(self.span)),
+                        Err(err) => ShellError::GenericError {
+                            error: "Unknown error".into(),
+                            msg: format!("{err:?}"),
+                            span: Some(self.span),
+                            help: None,
+                            inner: Vec::new(),
+                        },
+                    })?
+                    .err_span(self.span)?;
+            }
+
+            res.err_span(self.span)?;
+        } else if let Some(stderr) = self.stderr.take() {
+            consume_pipe(stderr).err_span(self.span)?;
+        }
+
+        self.exit_status.wait(self.span)
+    }
+
+    pub fn try_wait(&mut self) -> Result<Option<ExitStatus>, ShellError> {
+        self.exit_status.try_wait(self.span)
+    }
+
+    pub fn wait_with_output(mut self) -> Result<ProcessOutput, ShellError> {
+        let (stdout, stderr) = if let Some(stdout) = self.stdout {
+            let stderr = self
+                .stderr
+                .map(|stderr| thread::Builder::new().spawn(move || collect_bytes(stderr)))
+                .transpose()
+                .err_span(self.span)?;
+
+            let stdout = collect_bytes(stdout).err_span(self.span)?;
+
+            let stderr = stderr
+                .map(|handle| {
+                    handle.join().map_err(|e| match e.downcast::<io::Error>() {
+                        Ok(io) => ShellError::from((*io).into_spanned(self.span)),
+                        Err(err) => ShellError::GenericError {
+                            error: "Unknown error".into(),
+                            msg: format!("{err:?}"),
+                            span: Some(self.span),
+                            help: None,
+                            inner: Vec::new(),
+                        },
+                    })
+                })
+                .transpose()?
+                .transpose()
+                .err_span(self.span)?;
+
+            (Some(stdout), stderr)
+        } else {
+            let stderr = self
+                .stderr
+                .map(collect_bytes)
+                .transpose()
+                .err_span(self.span)?;
+
+            (None, stderr)
+        };
+
+        let exit_status = self.exit_status.wait(self.span)?;
+
+        Ok(ProcessOutput {
+            stdout,
+            stderr,
+            exit_status,
+        })
+    }
+}
+
+fn collect_bytes(pipe: ChildPipe) -> io::Result<Vec<u8>> {
+    let mut buf = Vec::new();
+    match pipe {
+        ChildPipe::Pipe(mut pipe) => pipe.read_to_end(&mut buf),
+        ChildPipe::Tee(mut tee) => tee.read_to_end(&mut buf),
+    }?;
+    Ok(buf)
+}
+
+fn consume_pipe(pipe: ChildPipe) -> io::Result<()> {
+    match pipe {
+        ChildPipe::Pipe(mut pipe) => io::copy(&mut pipe, &mut io::sink()),
+        ChildPipe::Tee(mut tee) => io::copy(&mut tee, &mut io::sink()),
+    }?;
+    Ok(())
+}
+
+pub struct ProcessOutput {
+    pub stdout: Option<Vec<u8>>,
+    pub stderr: Option<Vec<u8>>,
+    pub exit_status: ExitStatus,
+}

crates/nu-protocol/src/process/exit_status.rs

@@ -0,0 +1,64 @@
+use std::process;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ExitStatus {
+    Exited(i32),
+    #[cfg(unix)]
+    Signaled {
+        signal: i32,
+        core_dumped: bool,
+    },
+}
+
+impl ExitStatus {
+    pub fn code(self) -> i32 {
+        match self {
+            ExitStatus::Exited(code) => code,
+            #[cfg(unix)]
+            ExitStatus::Signaled { signal, .. } => -signal,
+        }
+    }
+}
+
+#[cfg(unix)]
+impl From<process::ExitStatus> for ExitStatus {
+    fn from(status: process::ExitStatus) -> Self {
+        use std::os::unix::process::ExitStatusExt;
+
+        match (status.code(), status.signal()) {
+            (Some(code), None) => Self::Exited(code),
+            (None, Some(signal)) => Self::Signaled {
+                signal,
+                core_dumped: status.core_dumped(),
+            },
+            (None, None) => {
+                debug_assert!(false, "ExitStatus should have either a code or a signal");
+                Self::Exited(-1)
+            }
+            (Some(code), Some(signal)) => {
+                // Should be unreachable, as `code()` will be `None` if `signal()` is `Some`
+                // according to the docs for `ExitStatus::code`.
+                debug_assert!(
+                    false,
+                    "ExitStatus cannot have both a code ({code}) and a signal ({signal})"
+                );
+                Self::Signaled {
+                    signal,
+                    core_dumped: status.core_dumped(),
+                }
+            }
+        }
+    }
+}
+
+#[cfg(not(unix))]
+impl From<process::ExitStatus> for ExitStatus {
+    fn from(status: process::ExitStatus) -> Self {
+        let code = status.code();
+        debug_assert!(
+            code.is_some(),
+            "`ExitStatus::code` cannot return `None` on windows"
+        );
+        Self::Exited(code.unwrap_or(-1))
+    }
+}

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

@@ -0,0 +1,5 @@
+mod child;
+mod exit_status;
+
+pub use child::*;
+pub use exit_status::ExitStatus;

crates/nu-protocol/src/util.rs

@@ -1,52 +0,0 @@
-use crate::ShellError;
-use std::io::{BufRead, BufReader, Read};
-
-pub struct BufferedReader<R: Read> {
-    input: BufReader<R>,
-    error: bool,
-}
-
-impl<R: Read> BufferedReader<R> {
-    pub fn new(input: BufReader<R>) -> Self {
-        Self {
-            input,
-            error: false,
-        }
-    }
-
-    pub fn into_inner(self) -> BufReader<R> {
-        self.input
-    }
-}
-
-impl<R: Read> Iterator for BufferedReader<R> {
-    type Item = Result<Vec<u8>, ShellError>;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        // Don't try to read more data if an error occurs
-        if self.error {
-            return None;
-        }
-
-        let buffer = self.input.fill_buf();
-        match buffer {
-            Ok(s) => {
-                let result = s.to_vec();
-
-                let buffer_len = s.len();
-
-                if buffer_len == 0 {
-                    None
-                } else {
-                    self.input.consume(buffer_len);
-
-                    Some(Ok(result))
-                }
-            }
-            Err(e) => {
-                self.error = true;
-                Some(Err(ShellError::IOError { msg: e.to_string() }))
-            }
-        }
-    }
-}

crates/nu-protocol/tests/test_pipeline_data.rs

@@ -11,5 +11,5 @@ fn test_convert_pipeline_data_to_value() {
     let new_span = Span::new(5, 6);
     let converted_value = pipeline_data.into_value(new_span);
 
-    assert_eq!(converted_value, Value::int(value_val, new_span));
+    assert_eq!(converted_value, Ok(Value::int(value_val, new_span)));
 }