Remove async_stream! from some commands (#1951)

* Remove async_stream! from open.rs

* Ran rustfmt

* Fix Clippy warning

* Removed async_stream! from evaluate_by.rs

* Removed async_stream! from exit.rs

* Removed async_stream! from from_eml.rs

* Removed async_stream! from group_by_date.rs

* Removed async_stream! from group_by.rs

* Removed async_stream! from map_max.rs

* Removed async_stream! from to_sqlite.rs

* Removed async_stream! from to_md.rs

* Removed async_stream! from to_html.rs

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

@@ -37,28 +37,26 @@ impl WholeStreamCommand for EvaluateBy {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        evaluate_by(args, registry)
+        evaluate_by(args, registry).await
     }
 }
 
-pub fn evaluate_by(
+pub async fn evaluate_by(
     args: CommandArgs,
     registry: &CommandRegistry,
 ) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
     let name = args.call_info.name_tag.clone();
     let (EvaluateByArgs { evaluate_with }, mut input) = args.process(&registry).await?;
     let values: Vec<Value> = input.collect().await;
 
     if values.is_empty() {
-            yield Err(ShellError::labeled_error(
+        Err(ShellError::labeled_error(
             "Expected table from pipeline",
             "requires a table input",
-                    name
+            name,
         ))
     } else {
-
         let evaluate_with = if let Some(evaluator) = evaluate_with {
             Some(evaluator.item().clone())
         } else {
@@ -66,13 +64,10 @@ pub fn evaluate_by(
         };
 
         match evaluate(&values[0], evaluate_with, name) {
-                Ok(evaluated) => yield ReturnSuccess::value(evaluated),
+            Ok(evaluated) => Ok(OutputStream::one(ReturnSuccess::value(evaluated))),
-                Err(err) => yield Err(err)
+            Err(err) => Err(err),
         }
     }
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]

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

@@ -25,7 +25,7 @@ impl WholeStreamCommand for Exit {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        exit(args, registry)
+        exit(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -44,19 +44,20 @@ impl WholeStreamCommand for Exit {
     }
 }
 
-pub fn exit(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub async fn exit(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
     let args = args.evaluate_once(&registry).await?;
 
-        if args.call_info.args.has("now") {
+    let command_action = if args.call_info.args.has("now") {
-            yield Ok(ReturnSuccess::Action(CommandAction::Exit));
+        CommandAction::Exit
     } else {
-            yield Ok(ReturnSuccess::Action(CommandAction::LeaveShell));
+        CommandAction::LeaveShell
-        }
     };
 
-    Ok(stream.to_output_stream())
+    Ok(OutputStream::one(ReturnSuccess::action(command_action)))
 }
 
 #[cfg(test)]

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

@@ -40,7 +40,7 @@ impl WholeStreamCommand for FromEML {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        from_eml(args, registry)
+        from_eml(args, registry).await
     }
 }
 
@@ -77,19 +77,30 @@ fn headerfieldvalue_to_value(tag: &Tag, value: &HeaderFieldValue) -> UntaggedVal
     }
 }
 
-fn from_eml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn from_eml(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let tag = args.call_info.name_tag.clone();
     let registry = registry.clone();
-    let stream = async_stream! {
+    let (eml_args, input): (FromEMLArgs, _) = args.process(&registry).await?;
-        let (eml_args, mut input): (FromEMLArgs, _) = args.process(&registry).await?;
     let value = input.collect_string(tag.clone()).await?;
 
-        let body_preview = eml_args.preview_body.map(|b| b.item).unwrap_or(DEFAULT_BODY_PREVIEW);
+    let body_preview = eml_args
+        .preview_body
+        .map(|b| b.item)
+        .unwrap_or(DEFAULT_BODY_PREVIEW);
 
     let eml = EmlParser::from_string(value.item)
         .with_body_preview(body_preview)
         .parse()
-        .map_err(|_| ShellError::labeled_error("Could not parse .eml file", "could not parse .eml file", &tag))?;
+        .map_err(|_| {
+            ShellError::labeled_error(
+                "Could not parse .eml file",
+                "could not parse .eml file",
+                &tag,
+            )
+        })?;
 
     let mut dict = TaggedDictBuilder::new(&tag);
 
@@ -113,10 +124,7 @@ fn from_eml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStrea
         dict.insert_untagged("Body", UntaggedValue::string(body));
     }
 
-        yield ReturnSuccess::value(dict.into_value());
+    Ok(OutputStream::one(ReturnSuccess::value(dict.into_value())))
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]

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

@@ -35,7 +35,7 @@ impl WholeStreamCommand for GroupBy {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        group_by(args, registry)
+        group_by(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -71,30 +71,27 @@ impl WholeStreamCommand for GroupBy {
     }
 }
 
-pub fn group_by(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub async fn group_by(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
     let name = args.call_info.name_tag.clone();
-    let stream = async_stream! {
+    let (GroupByArgs { column_name }, input) = args.process(&registry).await?;
-        let (GroupByArgs { column_name }, mut input) = args.process(&registry).await?;
     let values: Vec<Value> = input.collect().await;
 
     if values.is_empty() {
-            yield Err(ShellError::labeled_error(
+        Err(ShellError::labeled_error(
             "Expected table from pipeline",
             "requires a table input",
-                    name
+            name,
         ))
     } else {
-
         match crate::utils::data::group(column_name, &values, None, &name) {
-                Ok(grouped) => yield ReturnSuccess::value(grouped),
+            Ok(grouped) => Ok(OutputStream::one(ReturnSuccess::value(grouped))),
-                Err(err) => yield Err(err),
+            Err(err) => Err(err),
         }
-
     }
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 pub fn group(

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

@@ -42,7 +42,7 @@ impl WholeStreamCommand for GroupByDate {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        group_by_date(args, registry)
+        group_by_date(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -58,25 +58,29 @@ enum Grouper {
     ByDate(Option<String>),
 }
 
-pub fn group_by_date(
+pub async fn group_by_date(
     args: CommandArgs,
     registry: &CommandRegistry,
 ) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
     let name = args.call_info.name_tag.clone();
-    let stream = async_stream! {
+    let (
-        let (GroupByDateArgs { column_name, format }, mut input) = args.process(&registry).await?;
+        GroupByDateArgs {
+            column_name,
+            format,
+        },
+        input,
+    ) = args.process(&registry).await?;
     let values: Vec<Value> = input.collect().await;
 
     if values.is_empty() {
-            yield Err(ShellError::labeled_error(
+        Err(ShellError::labeled_error(
             "Expected table from pipeline",
             "requires a table input",
-                    name
+            name,
         ))
     } else {
-
+        let grouper = if let Some(Tagged { item: fmt, tag: _ }) = format {
-            let grouper = if let Some(Tagged { item: fmt, tag }) = format {
             Grouper::ByDate(Some(fmt))
         } else {
             Grouper::ByDate(None)
@@ -84,24 +88,29 @@ pub fn group_by_date(
 
         match grouper {
             Grouper::ByDate(None) => {
-                    match crate::utils::data::group(column_name, &values, Some(Box::new(|row: &Value| row.format("%Y-%b-%d"))), &name) {
+                match crate::utils::data::group(
-                        Ok(grouped) => yield ReturnSuccess::value(grouped),
+                    column_name,
-                        Err(err) => yield Err(err),
+                    &values,
+                    Some(Box::new(|row: &Value| row.format("%Y-%b-%d"))),
+                    &name,
+                ) {
+                    Ok(grouped) => Ok(OutputStream::one(ReturnSuccess::value(grouped))),
+                    Err(err) => Err(err),
                 }
             }
             Grouper::ByDate(Some(fmt)) => {
-                    match crate::utils::data::group(column_name, &values, Some(Box::new(move |row: &Value| {
+                match crate::utils::data::group(
-                        row.format(&fmt)
+                    column_name,
-                    })), &name) {
+                    &values,
-                        Ok(grouped) => yield ReturnSuccess::value(grouped),
+                    Some(Box::new(move |row: &Value| row.format(&fmt))),
-                        Err(err) => yield Err(err),
+                    &name,
+                ) {
+                    Ok(grouped) => Ok(OutputStream::one(ReturnSuccess::value(grouped))),
+                    Err(err) => Err(err),
                 }
             }
         }
     }
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]

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

@@ -38,28 +38,26 @@ impl WholeStreamCommand for MapMaxBy {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        map_max_by(args, registry)
+        map_max_by(args, registry).await
     }
 }
 
-pub fn map_max_by(
+pub async fn map_max_by(
     args: CommandArgs,
     registry: &CommandRegistry,
 ) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
     let name = args.call_info.name_tag.clone();
-    let stream = async_stream! {
     let (MapMaxByArgs { column_name }, mut input) = args.process(&registry).await?;
     let values: Vec<Value> = input.collect().await;
 
     if values.is_empty() {
-            yield Err(ShellError::labeled_error(
+        Err(ShellError::labeled_error(
             "Expected table from pipeline",
             "requires a table input",
-                    name
+            name,
         ))
     } else {
-
         let map_by_column = if let Some(column_to_map) = column_name {
             Some(column_to_map.item().clone())
         } else {
@@ -67,13 +65,10 @@ pub fn map_max_by(
         };
 
         match map_max(&values[0], map_by_column, name) {
-                Ok(table_maxed) => yield ReturnSuccess::value(table_maxed),
+            Ok(table_maxed) => Ok(OutputStream::one(ReturnSuccess::value(table_maxed))),
-                Err(err) => yield Err(err)
+            Err(err) => Err(err),
         }
     }
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]

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

@@ -42,7 +42,7 @@ impl WholeStreamCommand for Open {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        open(args, registry)
+        open(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -54,19 +54,14 @@ impl WholeStreamCommand for Open {
     }
 }
 
-fn open(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn open(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let cwd = PathBuf::from(args.shell_manager.path());
     let full_path = cwd;
     let registry = registry.clone();
 
-    let stream = async_stream! {
     let (OpenArgs { path, raw }, _) = args.process(&registry).await?;
     let result = fetch(&full_path, &path.item, path.tag.span).await;
 
-        if let Err(e) = result {
-            yield Err(e);
-            return;
-        }
     let (file_extension, contents, contents_tag) = result?;
 
     let file_extension = if raw.item {
@@ -74,19 +69,18 @@ fn open(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, S
     } else {
         // If the extension could not be determined via mimetype, try to use the path
         // extension. Some file types do not declare their mimetypes (such as bson files).
-            file_extension.or(path.extension().map(|x| x.to_string_lossy().to_string()))
+        file_extension.or_else(|| path.extension().map(|x| x.to_string_lossy().to_string()))
     };
 
     let tagged_contents = contents.into_value(&contents_tag);
 
     if let Some(extension) = file_extension {
-            yield Ok(ReturnSuccess::Action(CommandAction::AutoConvert(tagged_contents, extension)))
+        Ok(OutputStream::one(ReturnSuccess::action(
+            CommandAction::AutoConvert(tagged_contents, extension),
+        )))
     } else {
-            yield ReturnSuccess::value(tagged_contents);
+        Ok(OutputStream::one(ReturnSuccess::value(tagged_contents)))
     }
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 pub async fn fetch(

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

@@ -27,13 +27,15 @@ impl WholeStreamCommand for ToHTML {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        to_html(args, registry)
+        to_html(args, registry).await
     }
 }
 
-fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn to_html(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
     let args = args.evaluate_once(&registry).await?;
     let name_tag = args.name_tag();
     let input: Vec<Value> = args.input.collect().await;
@@ -57,11 +59,13 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
             UntaggedValue::Primitive(Primitive::Binary(b)) => {
                 // This might be a bit much, but it's fun :)
                 match row.tag.anchor {
-                        Some(AnchorLocation::Url(f)) |
+                    Some(AnchorLocation::Url(f)) | Some(AnchorLocation::File(f)) => {
-                        Some(AnchorLocation::File(f)) => {
                         let extension = f.split('.').last().map(String::from);
                         match extension {
-                                Some(s) if ["png", "jpg", "bmp", "gif", "tiff", "jpeg"].contains(&s.to_lowercase().as_str()) => {
+                            Some(s)
+                                if ["png", "jpg", "bmp", "gif", "tiff", "jpeg"]
+                                    .contains(&s.to_lowercase().as_str()) =>
+                            {
                                 output_string.push_str("<img src=\"data:image/");
                                 output_string.push_str(&s);
                                 output_string.push_str(";base64,");
@@ -77,8 +81,7 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
             UntaggedValue::Primitive(Primitive::String(ref b)) => {
                 // This might be a bit much, but it's fun :)
                 match row.tag.anchor {
-                        Some(AnchorLocation::Url(f)) |
+                    Some(AnchorLocation::Url(f)) | Some(AnchorLocation::File(f)) => {
-                        Some(AnchorLocation::File(f)) => {
                         let extension = f.split('.').last().map(String::from);
                         match extension {
                             Some(s) if s.to_lowercase() == "svg" => {
@@ -92,7 +95,10 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
                     }
                     _ => {}
                 }
-                    output_string.push_str(&(htmlescape::encode_minimal(&format_leaf(&row.value).plain_string(100_000)).replace("\n", "<br>")));
+                output_string.push_str(
+                    &(htmlescape::encode_minimal(&format_leaf(&row.value).plain_string(100_000))
+                        .replace("\n", "<br>")),
+                );
             }
             UntaggedValue::Row(row) => {
                 output_string.push_str("<tr>");
@@ -105,7 +111,10 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
                 output_string.push_str("</tr>");
             }
             p => {
-                    output_string.push_str(&(htmlescape::encode_minimal(&format_leaf(&p).plain_string(100_000)).replace("\n", "<br>")));
+                output_string.push_str(
+                    &(htmlescape::encode_minimal(&format_leaf(&p).plain_string(100_000))
+                        .replace("\n", "<br>")),
+                );
             }
         }
     }
@@ -115,10 +124,9 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
     }
     output_string.push_str("</body></html>");
 
-        yield ReturnSuccess::value(UntaggedValue::string(output_string).into_value(name_tag));
+    Ok(OutputStream::one(ReturnSuccess::value(
-    };
+        UntaggedValue::string(output_string).into_value(name_tag),
-
+    )))
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]

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

@@ -26,13 +26,15 @@ impl WholeStreamCommand for ToMarkdown {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        to_html(args, registry)
+        to_html(args, registry).await
     }
 }
 
-fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn to_html(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
     let args = args.evaluate_once(&registry).await?;
     let name_tag = args.name_tag();
     let input: Vec<Value> = args.input.collect().await;
@@ -65,16 +67,17 @@ fn to_html(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream
                 output_string.push_str("\n");
             }
             p => {
-                    output_string.push_str(&(htmlescape::encode_minimal(&format_leaf(&p).plain_string(100_000))));
+                output_string.push_str(
+                    &(htmlescape::encode_minimal(&format_leaf(&p).plain_string(100_000))),
+                );
                 output_string.push_str("\n");
             }
         }
     }
 
-        yield ReturnSuccess::value(UntaggedValue::string(output_string).into_value(name_tag));
+    Ok(OutputStream::one(ReturnSuccess::value(
-    };
+        UntaggedValue::string(output_string).into_value(name_tag),
-
+    )))
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]

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

@@ -27,7 +27,7 @@ impl WholeStreamCommand for ToSQLite {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        to_sqlite(args, registry)
+        to_sqlite(args, registry).await
     }
 
     fn is_binary(&self) -> bool {
@@ -56,7 +56,7 @@ impl WholeStreamCommand for ToDB {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        to_sqlite(args, registry)
+        to_sqlite(args, registry).await
     }
 
     fn is_binary(&self) -> bool {
@@ -203,26 +203,23 @@ fn sqlite_input_stream_to_bytes(values: Vec<Value>) -> Result<Value, std::io::Er
     Ok(UntaggedValue::binary(out).into_value(tag))
 }
 
-fn to_sqlite(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn to_sqlite(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
     let args = args.evaluate_once(&registry).await?;
     let name_tag = args.name_tag();
     let input: Vec<Value> = args.input.collect().await;
 
     match sqlite_input_stream_to_bytes(input) {
-            Ok(out) => yield ReturnSuccess::value(out),
+        Ok(out) => Ok(OutputStream::one(ReturnSuccess::value(out))),
-            _ => {
+        _ => Err(ShellError::labeled_error(
-                yield Err(ShellError::labeled_error(
             "Expected a table with SQLite-compatible structure from pipeline",
             "requires SQLite-compatible input",
             name_tag,
-                ))
+        )),
-            },
     }
-    };
-
-    Ok(stream.to_output_stream())
 }
 
 #[cfg(test)]