Only spawn external once when no $it argument (#1358)

crates/nu-test-support/src/bins/chop.rs

@@ -8,12 +8,9 @@ fn main() {
 
     // if no arguments given, chop from standard input and exit.
     let stdin = io::stdin();
-    let mut input = stdin.lock().lines();
+    for line in stdin.lock().lines() {
-
+        if let Ok(given) = line {
-    if let Some(Ok(given)) = input.next() {
-        if !given.is_empty() {
             println!("{}", chop(&given));
-            std::process::exit(0);
         }
     }
 
@@ -21,9 +18,12 @@ fn main() {
 }
 
 fn chop(word: &str) -> &str {
-    let to = word.len() - 1;
+    if word.is_empty() {
-
+        word
-    &word[..to]
+    } else {
+        let to = word.len() - 1;
+        &word[..to]
+    }
 }
 
 fn did_chop_arguments() -> bool {

src/commands/classified/external.rs

@@ -121,7 +121,7 @@ async fn run_with_iterator_arg(
                 }
             }).collect::<Vec<String>>();
 
-            match spawn(&command, &path, &process_args[..], None, is_last).await {
+            match spawn(&command, &path, &process_args[..], None, is_last) {
                 Ok(res) => {
                     if let Some(mut res) = res {
                         while let Some(item) = res.next().await {
@@ -151,83 +151,46 @@ async fn run_with_stdin(
 ) -> Result<Option<InputStream>, ShellError> {
     let path = context.shell_manager.path();
 
-    let mut inputs: InputStream = if let Some(input) = input {
+    let input = input
-        trace_stream!(target: "nu::trace_stream::external::stdin", "input" = input)
+        .map(|input| trace_stream!(target: "nu::trace_stream::external::stdin", "input" = input));
-    } else {
-        InputStream::empty()
-    };
 
-    let stream = async_stream! {
+    let process_args = command
-        while let Some(value) = inputs.next().await {
+        .args
-            let name = command.name.clone();
+        .iter()
-            let name_tag = command.name_tag.clone();
+        .map(|arg| {
-            let home_dir = dirs::home_dir();
+            let arg = expand_tilde(arg.deref(), dirs::home_dir);
-            let path = &path;
-            let args = command.args.clone();
 
-            let value_for_stdin = match nu_value_to_string_for_stdin(&command, &value) {
+            #[cfg(not(windows))]
-                Ok(value) => value,
+            {
-                Err(reason) => {
+                if argument_contains_whitespace(&arg) && argument_is_quoted(&arg) {
-                    yield Ok(Value {
-                        value: UntaggedValue::Error(reason),
-                        tag: name_tag
-                    });
-                    return;
-                }
-            };
-
-            let process_args = args.iter().map(|arg| {
-                let arg = expand_tilde(arg.deref(), || home_dir.as_ref());
-
-                #[cfg(not(windows))]
-                {
-                    if argument_contains_whitespace(&arg) && argument_is_quoted(&arg) {
-                        if let Some(unquoted) = remove_quotes(&arg) {
-                            format!(r#""{}""#, unquoted)
-                        } else {
-                            arg.as_ref().to_string()
-                        }
-                    } else {
-                        arg.as_ref().to_string()
-                    }
-                }
-                #[cfg(windows)]
-                {
                     if let Some(unquoted) = remove_quotes(&arg) {
-                        unquoted.to_string()
+                        format!(r#""{}""#, unquoted)
                     } else {
                         arg.as_ref().to_string()
                     }
-                }
+                } else {
-            }).collect::<Vec<String>>();
+                    arg.as_ref().to_string()
-
-            match spawn(&command, &path, &process_args[..], value_for_stdin, is_last).await {
-                Ok(res) => {
-                    if let Some(mut res) = res {
-                        while let Some(item) = res.next().await {
-                            yield Ok(item)
-                        }
-                    }
-                }
-                Err(reason) => {
-                    yield Ok(Value {
-                        value: UntaggedValue::Error(reason),
-                        tag: name_tag
-                    });
-                    return;
                 }
             }
-        }
+            #[cfg(windows)]
-    };
+            {
+                if let Some(unquoted) = remove_quotes(&arg) {
+                    unquoted.to_string()
+                } else {
+                    arg.as_ref().to_string()
+                }
+            }
+        })
+        .collect::<Vec<String>>();
 
-    Ok(Some(stream.to_input_stream()))
+    spawn(&command, &path, &process_args[..], input, is_last)
 }
 
-async fn spawn(
+fn spawn(
     command: &ExternalCommand,
     path: &str,
     args: &[String],
-    stdin_contents: Option<String>,
+    input: Option<InputStream>,
     is_last: bool,
 ) -> Result<Option<InputStream>, ShellError> {
     let command = command.clone();
@@ -265,7 +228,7 @@ async fn spawn(
     }
 
     // open since we have some contents for stdin
-    if stdin_contents.is_some() {
+    if input.is_some() {
         process.stdin(Stdio::piped());
         trace!(target: "nu::run::external", "set up stdin pipe");
     }
@@ -274,52 +237,37 @@ async fn spawn(
 
     if let Ok(mut child) = process.spawn() {
         let stream = async_stream! {
-            if let Some(mut input) = stdin_contents.as_ref() {
+            if let Some(mut input) = input {
                 let mut stdin_write = child.stdin
                     .take()
                     .expect("Internal error: could not get stdin pipe for external command");
 
-                if let Err(e) = stdin_write.write(input.as_bytes()) {
+                while let Some(value) = input.next().await {
-                    let message = format!("Unable to write to stdin (error = {})", e);
+                    let input_string = match nu_value_to_string_for_stdin(&command, &value) {
+                        Ok(None) => continue,
+                        Ok(Some(v)) => v,
+                        Err(e) => {
+                            yield Ok(Value {
+                                value: UntaggedValue::Error(e),
+                                tag: name_tag
+                            });
+                            return;
+                        }
+                    };
 
-                    yield Ok(Value {
+                    if let Err(e) = stdin_write.write(input_string.as_bytes()) {
-                        value: UntaggedValue::Error(ShellError::labeled_error(
+                        let message = format!("Unable to write to stdin (error = {})", e);
-                            message,
-                            "application may have closed before completing pipeline",
-                            &name_tag)),
-                        tag: name_tag
-                    });
-                    return;
-                }
 
-                drop(stdin_write);
+                        yield Ok(Value {
-            }
+                            value: UntaggedValue::Error(ShellError::labeled_error(
-
+                                message,
-            if is_last && command.has_it_argument() {
+                                "application may have closed before completing pipeline",
-                if let Ok(status) = child.wait() {
+                                &name_tag)),
-                    if status.success() {
+                            tag: name_tag
+                        });
                         return;
                     }
                 }
-
-                // We can give an error when we see a non-zero exit code, but this is different
-                // than what other shells will do.
-                let cfg = crate::data::config::config(Tag::unknown());
-                if let Ok(cfg) = cfg {
-                    if cfg.contains_key("nonzero_exit_errors") {
-                        yield Ok(Value {
-                            value: UntaggedValue::Error(
-                                ShellError::labeled_error(
-                                    "External command failed",
-                                    "command failed",
-                                    &name_tag,
-                                )
-                            ),
-                            tag: name_tag,
-                        });
-                    }
-                }
-                return;
             }
 
             if !is_last {
@@ -340,7 +288,7 @@ async fn spawn(
                 let mut stream = FramedRead::new(file, LinesCodec).map(|line| {
                     if let Ok(line) = line {
                         Value {
-                            value: UntaggedValue::Primitive(Primitive::String(line)),
+                            value: UntaggedValue::Primitive(Primitive::Line(line)),
                             tag: name_tag.clone(),
                         }
                     } else {
@@ -353,6 +301,8 @@ async fn spawn(
                 }
             }
 
+            // We can give an error when we see a non-zero exit code, but this is different
+            // than what other shells will do.
             if child.wait().is_err() {
                 let cfg = crate::data::config::config(Tag::unknown());
                 if let Ok(cfg) = cfg {

tests/shell/pipeline/commands/internal.rs

@@ -21,8 +21,8 @@ fn takes_rows_of_nu_value_strings_and_pipes_it_to_stdin_of_external() {
         r#"
             open nu_times.csv
             | get name
+            | ^echo $it
             | chop
-            | lines
             | nth 3
             | echo $it
             "#