Removing async_stream! from some more commands (#1973)

* Removing async_stream! from some more commands

* Fix await error

* Fix Clippy warnings

crates/nu-cli/src/commands/mv.rs

@@ -43,7 +43,7 @@ impl WholeStreamCommand for Move {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        mv(args, registry)
+        mv(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -67,20 +67,13 @@ impl WholeStreamCommand for Move {
     }
 }
 
-fn mv(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn mv(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
+    let name = args.call_info.name_tag.clone();
-        let name = args.call_info.name_tag.clone();
+    let shell_manager = args.shell_manager.clone();
-        let shell_manager = args.shell_manager.clone();
+    let (args, _) = args.process(&registry).await?;
-        let (args, _) = args.process(&registry).await?;
-        let mut result = shell_manager.mv(args, name)?;
 
-        while let Some(item) = result.next().await {
+    shell_manager.mv(args, name)
-            yield item;
-        }
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]

crates/nu-cli/src/commands/reject.rs

@@ -31,7 +31,7 @@ impl WholeStreamCommand for Reject {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        reject(args, registry)
+        reject(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -43,28 +43,24 @@ impl WholeStreamCommand for Reject {
     }
 }
 
-fn reject(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn reject(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
+    let name = args.call_info.name_tag.clone();
-        let name = args.call_info.name_tag.clone();
+    let (RejectArgs { rest: fields }, input) = args.process(&registry).await?;
-        let (RejectArgs { rest: fields }, mut input) = args.process(&registry).await?;
-        if fields.is_empty() {
-            yield Err(ShellError::labeled_error(
-                "Reject requires fields",
-                "needs parameter",
-                name,
-            ));
-            return;
-        }
 
-        let fields: Vec<_> = fields.iter().map(|f| f.item.clone()).collect();
+    if fields.is_empty() {
+        return Err(ShellError::labeled_error(
+            "Reject requires fields",
+            "needs parameter",
+            name,
+        ));
+    }
 
-        while let Some(item) = input.next().await {
+    let fields: Vec<_> = fields.iter().map(|f| f.item.clone()).collect();
-            yield ReturnSuccess::value(reject_fields(&item, &fields, &item.tag));
-        }
-    };
 
-    Ok(stream.to_output_stream())
+    Ok(input
+        .map(move |item| ReturnSuccess::value(reject_fields(&item, &fields, &item.tag)))
+        .to_output_stream())
 }
 
 #[cfg(test)]

crates/nu-cli/src/commands/save.rs

@@ -154,11 +154,14 @@ impl WholeStreamCommand for Save {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        save(args, registry)
+        save(args, registry).await
     }
 }
 
-fn save(raw_args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn save(
+    raw_args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let mut full_path = PathBuf::from(raw_args.shell_manager.path());
     let name_tag = raw_args.call_info.name_tag.clone();
     let name = raw_args.call_info.name_tag.clone();
@@ -169,101 +172,96 @@ fn save(raw_args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStrea
     let current_errors = raw_args.current_errors.clone();
     let shell_manager = raw_args.shell_manager.clone();
 
-    let stream = async_stream! {
+    let head = raw_args.call_info.args.head.clone();
-        let head = raw_args.call_info.args.head.clone();
+    let (
-        let (SaveArgs { path, raw: save_raw }, mut input) = raw_args.process(&registry).await?;
+        SaveArgs {
-        let input: Vec<Value> = input.collect().await;
+            path,
-        if path.is_none() {
+            raw: save_raw,
-            // If there is no filename, check the metadata for the anchor filename
+        },
-            if input.len() > 0 {
+        input,
-                let anchor = input[0].tag.anchor();
+    ) = raw_args.process(&registry).await?;
-                match anchor {
+    let input: Vec<Value> = input.collect().await;
-                    Some(path) => match path {
+    if path.is_none() {
-                        AnchorLocation::File(file) => {
+        let mut should_return_file_path_error = true;
-                            full_path.push(Path::new(&file));
+
-                        }
+        // If there is no filename, check the metadata for the anchor filename
-                        _ => {
+        if !input.is_empty() {
-                            yield Err(ShellError::labeled_error(
+            let anchor = input[0].tag.anchor();
-                                "Save requires a filepath",
+
-                                "needs path",
+            if let Some(path) = anchor {
-                                name_tag.clone(),
+                if let AnchorLocation::File(file) = path {
-                            ));
+                    should_return_file_path_error = false;
-                        }
+                    full_path.push(Path::new(&file));
-                    },
-                    None => {
-                        yield Err(ShellError::labeled_error(
-                            "Save requires a filepath",
-                            "needs path",
-                            name_tag.clone(),
-                        ));
-                    }
                 }
-            } else {
-                yield Err(ShellError::labeled_error(
-                    "Save requires a filepath",
-                    "needs path",
-                    name_tag.clone(),
-                ));
-            }
-        } else {
-            if let Some(file) = path {
-                full_path.push(file.item());
             }
         }
 
-        // TODO use label_break_value once it is stable:
+        if should_return_file_path_error {
-        // https://github.com/rust-lang/rust/issues/48594
+            return Err(ShellError::labeled_error(
-        let content : Result<Vec<u8>, ShellError> = 'scope: loop {
+                "Save requires a filepath",
-            break if !save_raw {
+                "needs path",
-                if let Some(extension) = full_path.extension() {
+                name_tag.clone(),
-                    let command_name = format!("to {}", extension.to_string_lossy());
+            ));
-                    if let Some(converter) = registry.get_command(&command_name) {
+        }
-                        let new_args = RawCommandArgs {
+    } else if let Some(file) = path {
-                            host,
+        full_path.push(file.item());
-                            ctrl_c,
+    }
-                            current_errors,
+
-                            shell_manager,
+    // TODO use label_break_value once it is stable:
-                            call_info: UnevaluatedCallInfo {
+    // https://github.com/rust-lang/rust/issues/48594
-                                args: nu_protocol::hir::Call {
+    #[allow(clippy::never_loop)]
-                                    head,
+    let content: Result<Vec<u8>, ShellError> = 'scope: loop {
-                                    positional: None,
+        break if !save_raw {
-                                    named: None,
+            if let Some(extension) = full_path.extension() {
-                                    span: Span::unknown(),
+                let command_name = format!("to {}", extension.to_string_lossy());
-                                    is_last: false,
+                if let Some(converter) = registry.get_command(&command_name) {
-                                },
+                    let new_args = RawCommandArgs {
-                                name_tag: name_tag.clone(),
+                        host,
-                                scope,
+                        ctrl_c,
-                            }
+                        current_errors,
-                        };
+                        shell_manager,
-                        let mut result = converter.run(new_args.with_input(input), &registry).await;
+                        call_info: UnevaluatedCallInfo {
-                        let result_vec: Vec<Result<ReturnSuccess, ShellError>> = result.drain_vec().await;
+                            args: nu_protocol::hir::Call {
-                        if converter.is_binary() {
+                                head,
-                            process_binary_return_success!('scope, result_vec, name_tag)
+                                positional: None,
-                        } else {
+                                named: None,
-                            process_string_return_success!('scope, result_vec, name_tag)
+                                span: Span::unknown(),
-                        }
+                                is_last: false,
+                            },
+                            name_tag: name_tag.clone(),
+                            scope,
+                        },
+                    };
+                    let mut result = converter.run(new_args.with_input(input), &registry).await;
+                    let result_vec: Vec<Result<ReturnSuccess, ShellError>> =
+                        result.drain_vec().await;
+                    if converter.is_binary() {
+                        process_binary_return_success!('scope, result_vec, name_tag)
                     } else {
-                        process_unknown!('scope, input, name_tag)
+                        process_string_return_success!('scope, result_vec, name_tag)
                     }
                 } else {
                     process_unknown!('scope, input, name_tag)
                 }
             } else {
-                Ok(string_from(&input).into_bytes())
+                process_unknown!('scope, input, name_tag)
-            };
+            }
+        } else {
+            Ok(string_from(&input).into_bytes())
         };
-
-        match content {
-            Ok(save_data) => match std::fs::write(full_path, save_data) {
-                Ok(o) => o,
-                Err(e) => yield Err(ShellError::labeled_error(e.to_string(), "IO error while saving", name)),
-            },
-            Err(e) => yield Err(e),
-        }
-
     };
 
-    Ok(OutputStream::new(stream))
+    match content {
+        Ok(save_data) => match std::fs::write(full_path, save_data) {
+            Ok(_) => Ok(OutputStream::empty()),
+            Err(e) => Err(ShellError::labeled_error(
+                e.to_string(),
+                "IO error while saving",
+                name,
+            )),
+        },
+        Err(e) => Err(e),
+    }
 }
 
 fn string_from(input: &[Value]) -> String {

crates/nu-cli/src/commands/touch.rs

@@ -33,7 +33,7 @@ impl WholeStreamCommand for Touch {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        touch(args, registry)
+        touch(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -45,21 +45,18 @@ impl WholeStreamCommand for Touch {
     }
 }
 
-fn touch(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn touch(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
+    let (TouchArgs { target }, _) = args.process(&registry).await?;
-        let (TouchArgs { target }, _) = args.process(&registry).await?;
-        match OpenOptions::new().write(true).create(true).open(&target) {
-            Ok(_) => {},
-            Err(err) => yield Err(ShellError::labeled_error(
-                "File Error",
-                err.to_string(),
-                &target.tag,
-            )),
-        }
-    };
 
-    Ok(stream.to_output_stream())
+    match OpenOptions::new().write(true).create(true).open(&target) {
+        Ok(_) => Ok(OutputStream::empty()),
+        Err(err) => Err(ShellError::labeled_error(
+            "File Error",
+            err.to_string(),
+            &target.tag,
+        )),
+    }
 }
 
 #[cfg(test)]