Add streaming support to save for ExternalStream data (#4985)

* Add streaming support to save for ExternalStream data

Prior to this change, save would collect data from an ExternalStream (data
originating from externals) consuming memory for the full amount of data piped
to it,

This change adds streaming support for ExternalStream allowing saving of
arbitrarily large files and bounding memory usage.

* Remove broken save test

This test passes but not for the right reasons, since this test was
written filename has become a required parameter.  The parser outputs
an error but the test still passes as is checking the original un-mutated
file assuming save has re-written the contents.

This change removes the test.

```
running 1 test
=== stderr
Error: nu::parser::missing_positional (https://docs.rs/nu-parser/0.60.0/nu-parser/enum.ParseError.html#variant.MissingPositional)

  × Missing required positional argument.
   ╭─[source:1:1]
 1 │ open save_test_1/cargo_sample.toml | save
   ·                                          ▲
   ·                                          ╰── missing filename
   ╰────
  help: Usage: save {flags} <filename>

test commands::save::figures_out_intelligently_where_to_write_out_with_metadata ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 515 filtered out; finished in 0.10s
```

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

@@ -4,7 +4,7 @@ use nu_protocol::engine::{Command, EngineState, Stack};
 use nu_protocol::{
     Category, Example, PipelineData, ShellError, Signature, Spanned, SyntaxShape, Value,
 };
-use std::io::Write;
+use std::io::{BufWriter, Write};
 
 use std::path::Path;
 
@@ -125,39 +125,71 @@ impl Command for Save {
                 )),
             }
         } else {
-            match input.into_value(span) {
-                Value::String { val, .. } => {
-                    if let Err(err) = file.write_all(val.as_bytes()) {
-                        return Err(ShellError::IOError(err.to_string()));
-                    }
+            match input {
+                PipelineData::ExternalStream { stdout: None, .. } => Ok(PipelineData::new(span)),
+                PipelineData::ExternalStream {
+                    stdout: Some(mut stream),
+                    ..
+                } => {
+                    let mut writer = BufWriter::new(file);
 
-                    Ok(PipelineData::new(span))
+                    stream
+                        .try_for_each(move |result| {
+                            let buf = match result {
+                                Ok(v) => match v {
+                                    Value::String { val, .. } => val.into_bytes(),
+                                    Value::Binary { val, .. } => val,
+                                    _ => {
+                                        return Err(ShellError::UnsupportedInput(
+                                            format!("{:?} not supported", v.get_type()),
+                                            v.span()?,
+                                        ));
+                                    }
+                                },
+                                Err(err) => return Err(err),
+                            };
+
+                            if let Err(err) = writer.write(&buf) {
+                                return Err(ShellError::IOError(err.to_string()));
+                            }
+                            Ok(())
+                        })
+                        .map(|_| PipelineData::new(span))
                 }
-                Value::Binary { val, .. } => {
-                    if let Err(err) = file.write_all(&val) {
-                        return Err(ShellError::IOError(err.to_string()));
+                input => match input.into_value(span) {
+                    Value::String { val, .. } => {
+                        if let Err(err) = file.write_all(val.as_bytes()) {
+                            return Err(ShellError::IOError(err.to_string()));
+                        }
+
+                        Ok(PipelineData::new(span))
                     }
+                    Value::Binary { val, .. } => {
+                        if let Err(err) = file.write_all(&val) {
+                            return Err(ShellError::IOError(err.to_string()));
+                        }
 
-                    Ok(PipelineData::new(span))
-                }
-                Value::List { vals, .. } => {
-                    let val = vals
-                        .into_iter()
-                        .map(|it| it.as_string())
-                        .collect::<Result<Vec<String>, ShellError>>()?
-                        .join("\n")
-                        + "\n";
-
-                    if let Err(err) = file.write_all(val.as_bytes()) {
-                        return Err(ShellError::IOError(err.to_string()));
+                        Ok(PipelineData::new(span))
                     }
+                    Value::List { vals, .. } => {
+                        let val = vals
+                            .into_iter()
+                            .map(|it| it.as_string())
+                            .collect::<Result<Vec<String>, ShellError>>()?
+                            .join("\n")
+                            + "\n";
 
-                    Ok(PipelineData::new(span))
-                }
-                v => Err(ShellError::UnsupportedInput(
-                    format!("{:?} not supported", v.get_type()),
-                    span,
-                )),
+                        if let Err(err) = file.write_all(val.as_bytes()) {
+                            return Err(ShellError::IOError(err.to_string()));
+                        }
+
+                        Ok(PipelineData::new(span))
+                    }
+                    v => Err(ShellError::UnsupportedInput(
+                        format!("{:?} not supported", v.get_type()),
+                        span,
+                    )),
+                },
             }
         }
     }

crates/nu-command/tests/commands/save.rs

@@ -1,36 +1,8 @@
-use nu_test_support::fs::{file_contents, Stub::FileWithContent};
+use nu_test_support::fs::file_contents;
 use nu_test_support::nu;
 use nu_test_support::playground::Playground;
 use std::io::Write;
 
-#[test]
-fn figures_out_intelligently_where_to_write_out_with_metadata() {
-    Playground::setup("save_test_1", |dirs, sandbox| {
-        sandbox.with_files(vec![FileWithContent(
-            "cargo_sample.toml",
-            r#"
-                [package]
-                name = "nu"
-                version = "0.1.1"
-                authors = ["Yehuda Katz <wycats@gmail.com>"]
-                description = "A shell for the GitHub era"
-                license = "ISC"
-                edition = "2018"
-            "#,
-        )]);
-
-        let subject_file = dirs.test().join("cargo_sample.toml");
-
-        nu!(
-            cwd: dirs.root(),
-            "open save_test_1/cargo_sample.toml | save"
-        );
-
-        let actual = file_contents(&subject_file);
-        assert!(actual.contains("0.1.1"));
-    })
-}
-
 #[test]
 fn writes_out_csv() {
     Playground::setup("save_test_2", |dirs, sandbox| {