diff --git a/crates/nu-cli/src/commands/split/command.rs b/crates/nu-cli/src/commands/split/command.rs
index 7f0e0e30c..0164ea23f 100644
--- a/crates/nu-cli/src/commands/split/command.rs
+++ b/crates/nu-cli/src/commands/split/command.rs
@@ -26,14 +26,10 @@ impl WholeStreamCommand for Command {
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
         let registry = registry.clone();
-        let stream = async_stream! {
-            yield Ok(ReturnSuccess::Value(
-                UntaggedValue::string(crate::commands::help::get_help(&Command, &registry))
-                    .into_value(Tag::unknown()),
-            ));
-        };
-
-        Ok(stream.to_output_stream())
+        Ok(OutputStream::one(Ok(ReturnSuccess::Value(
+            UntaggedValue::string(crate::commands::help::get_help(&Command, &registry))
+                .into_value(Tag::unknown()),
+        ))))
     }
 }
 
diff --git a/crates/nu-cli/src/commands/split/row.rs b/crates/nu-cli/src/commands/split/row.rs
index 34179d046..df73bcdd3 100644
--- a/crates/nu-cli/src/commands/split/row.rs
+++ b/crates/nu-cli/src/commands/split/row.rs
@@ -35,41 +35,52 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        split_row(args, registry)
+        split_row(args, registry).await
     }
 }
 
-fn split_row(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn split_row(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
-        let name = args.call_info.name_tag.clone();
-        let (SplitRowArgs { separator }, mut input) = args.process(&registry).await?;
-        while let Some(v) = input.next().await {
+    let name = args.call_info.name_tag.clone();
+    let (SplitRowArgs { separator }, input) = args.process(&registry).await?;
+    Ok(input
+        .flat_map(move |v| {
             if let Ok(s) = v.as_string() {
                 let splitter = separator.item.replace("\\n", "\n");
                 trace!("splitting with {:?}", splitter);
-                let split_result: Vec<_> = s.split(&splitter).filter(|s| s.trim() != "").collect();
+                let split_result: Vec<String> = s
+                    .split(&splitter)
+                    .filter_map(|s| {
+                        if s.trim() != "" {
+                            Some(s.to_string())
+                        } else {
+                            None
+                        }
+                    })
+                    .collect();
 
                 trace!("split result = {:?}", split_result);
 
-                for s in split_result {
-                    yield ReturnSuccess::value(
-                        UntaggedValue::Primitive(Primitive::String(s.into())).into_value(&v.tag),
-                    );
-                }
+                futures::stream::iter(split_result.into_iter().map(move |s| {
+                    ReturnSuccess::value(
+                        UntaggedValue::Primitive(Primitive::String(s)).into_value(&v.tag),
+                    )
+                }))
+                .to_output_stream()
             } else {
-                yield Err(ShellError::labeled_error_with_secondary(
+                OutputStream::one(Err(ShellError::labeled_error_with_secondary(
                     "Expected a string from pipeline",
                     "requires string input",
                     name.span,
                     "value originates from here",
                     v.tag.span,
-                ));
+                )))
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 #[cfg(test)]
diff --git a/crates/nu-cli/src/commands/str_/capitalize.rs b/crates/nu-cli/src/commands/str_/capitalize.rs
index 2fcb7c8ba..3b2c51548 100644
--- a/crates/nu-cli/src/commands/str_/capitalize.rs
+++ b/crates/nu-cli/src/commands/str_/capitalize.rs
@@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        operate(args, registry)
+        operate(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -49,47 +49,44 @@ impl WholeStreamCommand for SubCommand {
     }
 }
 
-fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn operate(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (Arguments { rest }, mut input) = args.process(&registry).await?;
+    let (Arguments { rest }, input) = args.process(&registry).await?;
 
-        let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
+    let column_paths: Vec<_> = rest;
 
-        while let Some(v) = input.next().await {
+    Ok(input
+        .map(move |v| {
             if column_paths.is_empty() {
                 match action(&v, v.tag()) {
-                    Ok(out) => yield ReturnSuccess::value(out),
-                    Err(err) => {
-                        yield Err(err);
-                        return;
-                    }
+                    Ok(out) => ReturnSuccess::value(out),
+                    Err(err) => Err(err),
                 }
             } else {
-
-                let mut ret = v.clone();
+                let mut ret = v;
 
                 for path in &column_paths {
-                    let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
+                    let swapping = ret.swap_data_by_column_path(
+                        path,
+                        Box::new(move |old| action(old, old.tag())),
+                    );
 
                     match swapping {
                         Ok(new_value) => {
                             ret = new_value;
                         }
-                        Err(err) => {
-                            yield Err(err);
-                            return;
-                        }
+                        Err(err) => return Err(err),
                     }
                 }
 
-                yield ReturnSuccess::value(ret);
+                ReturnSuccess::value(ret)
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {
diff --git a/crates/nu-cli/src/commands/str_/downcase.rs b/crates/nu-cli/src/commands/str_/downcase.rs
index b334003d3..0e8ca216e 100644
--- a/crates/nu-cli/src/commands/str_/downcase.rs
+++ b/crates/nu-cli/src/commands/str_/downcase.rs
@@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        operate(args, registry)
+        operate(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -49,47 +49,46 @@ impl WholeStreamCommand for SubCommand {
     }
 }
 
-fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn operate(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (Arguments { rest }, mut input) = args.process(&registry).await?;
+    let (Arguments { rest }, input) = args.process(&registry).await?;
 
-        let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
+    let column_paths: Vec<_> = rest;
 
-        while let Some(v) = input.next().await {
+    Ok(input
+        .map(move |v| {
             if column_paths.is_empty() {
                 match action(&v, v.tag()) {
-                    Ok(out) => yield ReturnSuccess::value(out),
-                    Err(err) => {
-                        yield Err(err);
-                        return;
-                    }
+                    Ok(out) => ReturnSuccess::value(out),
+                    Err(err) => Err(err),
                 }
             } else {
-
-                let mut ret = v.clone();
+                let mut ret = v;
 
                 for path in &column_paths {
-                    let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
+                    let swapping = ret.swap_data_by_column_path(
+                        path,
+                        Box::new(move |old| action(old, old.tag())),
+                    );
 
                     match swapping {
                         Ok(new_value) => {
                             ret = new_value;
                         }
                         Err(err) => {
-                            yield Err(err);
-                            return;
+                            return Err(err);
                         }
                     }
                 }
 
-                yield ReturnSuccess::value(ret);
+                ReturnSuccess::value(ret)
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {
diff --git a/crates/nu-cli/src/commands/str_/set.rs b/crates/nu-cli/src/commands/str_/set.rs
index 8c719f3ee..809fab0ae 100644
--- a/crates/nu-cli/src/commands/str_/set.rs
+++ b/crates/nu-cli/src/commands/str_/set.rs
@@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        operate(args, registry)
+        operate(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -59,50 +59,49 @@ impl WholeStreamCommand for SubCommand {
 #[derive(Clone)]
 struct Replace(String);
 
-fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn operate(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (Arguments { replace, rest }, mut input) = args.process(&registry).await?;
-        let options = Replace(replace.item);
+    let (Arguments { replace, rest }, input) = args.process(&registry).await?;
+    let options = Replace(replace.item);
 
-        let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
+    let column_paths: Vec<_> = rest;
 
-        while let Some(v) = input.next().await {
+    Ok(input
+        .map(move |v| {
             if column_paths.is_empty() {
                 match action(&v, &options, v.tag()) {
-                    Ok(out) => yield ReturnSuccess::value(out),
-                    Err(err) => {
-                        yield Err(err);
-                        return;
-                    }
+                    Ok(out) => ReturnSuccess::value(out),
+                    Err(err) => Err(err),
                 }
             } else {
-
-                let mut ret = v.clone();
+                let mut ret = v;
 
                 for path in &column_paths {
                     let options = options.clone();
 
-                    let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, &options, old.tag())));
+                    let swapping = ret.swap_data_by_column_path(
+                        path,
+                        Box::new(move |old| action(old, &options, old.tag())),
+                    );
 
                     match swapping {
                         Ok(new_value) => {
                             ret = new_value;
                         }
                         Err(err) => {
-                            yield Err(err);
-                            return;
+                            return Err(err);
                         }
                     }
                 }
 
-                yield ReturnSuccess::value(ret);
+                ReturnSuccess::value(ret)
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 fn action(_input: &Value, options: &Replace, tag: impl Into<Tag>) -> Result<Value, ShellError> {
diff --git a/crates/nu-cli/src/commands/str_/substring.rs b/crates/nu-cli/src/commands/str_/substring.rs
index ae9f3000a..5e4c649af 100644
--- a/crates/nu-cli/src/commands/str_/substring.rs
+++ b/crates/nu-cli/src/commands/str_/substring.rs
@@ -46,7 +46,7 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        operate(args, registry)
+        operate(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -73,91 +73,83 @@ impl WholeStreamCommand for SubCommand {
 #[derive(Clone)]
 struct Substring(usize, usize);
 
-fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn operate(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let name = args.call_info.name_tag.clone();
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (Arguments { range, rest }, mut input) = args.process(&registry).await?;
+    let (Arguments { range, rest }, input) = args.process(&registry).await?;
 
-        let v: Vec<&str> = range.item.split(',').collect();
+    let v: Vec<&str> = range.item.split(',').collect();
 
-        let start = match v[0] {
-            "" => 0,
-            _ => v[0]
-                .trim()
-                .parse()
-                .map_err(|_| {
-                    ShellError::labeled_error(
-                        "could not perform substring",
-                        "could not perform substring",
-                        name.span,
-                    )
-                })?
-        };
-
-        let end = match v[1] {
-            "" => usize::max_value(),
-            _ => v[1]
-                .trim()
-                .parse()
-                .map_err(|_| {
-                    ShellError::labeled_error(
-                        "could not perform substring",
-                        "could not perform substring",
-                        name.span,
-                    )
-                })?
-        };
-
-        if start > end {
-            yield Err(ShellError::labeled_error(
-                "End must be greater than or equal to Start",
-                "End must be greater than or equal to Start",
+    let start = match v[0] {
+        "" => 0,
+        _ => v[0].trim().parse().map_err(|_| {
+            ShellError::labeled_error(
+                "could not perform substring",
+                "could not perform substring",
                 name.span,
-            ));
-            return;
-        }
+            )
+        })?,
+    };
 
-        let options = Substring(start, end);
+    let end = match v[1] {
+        "" => usize::max_value(),
+        _ => v[1].trim().parse().map_err(|_| {
+            ShellError::labeled_error(
+                "could not perform substring",
+                "could not perform substring",
+                name.span,
+            )
+        })?,
+    };
 
-        let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
+    if start > end {
+        return Err(ShellError::labeled_error(
+            "End must be greater than or equal to Start",
+            "End must be greater than or equal to Start",
+            name.span,
+        ));
+    }
 
-        while let Some(v) = input.next().await {
+    let options = Substring(start, end);
+
+    let column_paths: Vec<_> = rest;
+
+    Ok(input
+        .map(move |v| {
             if column_paths.is_empty() {
                 match action(&v, &options, v.tag()) {
-                    Ok(out) => yield ReturnSuccess::value(out),
-                    Err(err) => {
-                        yield Err(err);
-                        return;
-                    }
+                    Ok(out) => ReturnSuccess::value(out),
+                    Err(err) => Err(err),
                 }
             } else {
-
-                let mut ret = v.clone();
+                let mut ret = v;
 
                 for path in &column_paths {
                     let options = options.clone();
 
-                    let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, &options, old.tag())));
+                    let swapping = ret.swap_data_by_column_path(
+                        path,
+                        Box::new(move |old| action(old, &options, old.tag())),
+                    );
 
                     match swapping {
                         Ok(new_value) => {
                             ret = new_value;
                         }
                         Err(err) => {
-                            yield Err(err);
-                            return;
+                            return Err(err);
                         }
                     }
                 }
 
-                yield ReturnSuccess::value(ret);
+                ReturnSuccess::value(ret)
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 fn action(input: &Value, options: &Substring, tag: impl Into<Tag>) -> Result<Value, ShellError> {
diff --git a/crates/nu-cli/src/commands/str_/to_decimal.rs b/crates/nu-cli/src/commands/str_/to_decimal.rs
index a101b723a..3e908d22c 100644
--- a/crates/nu-cli/src/commands/str_/to_decimal.rs
+++ b/crates/nu-cli/src/commands/str_/to_decimal.rs
@@ -40,7 +40,7 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        operate(args, registry)
+        operate(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -52,47 +52,46 @@ impl WholeStreamCommand for SubCommand {
     }
 }
 
-fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn operate(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (Arguments { rest }, mut input) = args.process(&registry).await?;
+    let (Arguments { rest }, input) = args.process(&registry).await?;
 
-        let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
+    let column_paths: Vec<_> = rest;
 
-        while let Some(v) = input.next().await {
+    Ok(input
+        .map(move |v| {
             if column_paths.is_empty() {
                 match action(&v, v.tag()) {
-                    Ok(out) => yield ReturnSuccess::value(out),
-                    Err(err) => {
-                        yield Err(err);
-                        return;
-                    }
+                    Ok(out) => ReturnSuccess::value(out),
+                    Err(err) => Err(err),
                 }
             } else {
-
-                let mut ret = v.clone();
+                let mut ret = v;
 
                 for path in &column_paths {
-                    let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
+                    let swapping = ret.swap_data_by_column_path(
+                        path,
+                        Box::new(move |old| action(old, old.tag())),
+                    );
 
                     match swapping {
                         Ok(new_value) => {
                             ret = new_value;
                         }
                         Err(err) => {
-                            yield Err(err);
-                            return;
+                            return Err(err);
                         }
                     }
                 }
 
-                yield ReturnSuccess::value(ret);
+                ReturnSuccess::value(ret)
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {
diff --git a/crates/nu-cli/src/commands/str_/to_integer.rs b/crates/nu-cli/src/commands/str_/to_integer.rs
index ce4f20e7d..41277401a 100644
--- a/crates/nu-cli/src/commands/str_/to_integer.rs
+++ b/crates/nu-cli/src/commands/str_/to_integer.rs
@@ -40,7 +40,7 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        operate(args, registry)
+        operate(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -52,47 +52,46 @@ impl WholeStreamCommand for SubCommand {
     }
 }
 
-fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn operate(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (Arguments { rest }, mut input) = args.process(&registry).await?;
+    let (Arguments { rest }, input) = args.process(&registry).await?;
 
-        let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
+    let column_paths: Vec<_> = rest;
 
-        while let Some(v) = input.next().await {
+    Ok(input
+        .map(move |v| {
             if column_paths.is_empty() {
                 match action(&v, v.tag()) {
-                    Ok(out) => yield ReturnSuccess::value(out),
-                    Err(err) => {
-                        yield Err(err);
-                        return;
-                    }
+                    Ok(out) => ReturnSuccess::value(out),
+                    Err(err) => Err(err),
                 }
             } else {
-
-                let mut ret = v.clone();
+                let mut ret = v;
 
                 for path in &column_paths {
-                    let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
+                    let swapping = ret.swap_data_by_column_path(
+                        path,
+                        Box::new(move |old| action(old, old.tag())),
+                    );
 
                     match swapping {
                         Ok(new_value) => {
                             ret = new_value;
                         }
                         Err(err) => {
-                            yield Err(err);
-                            return;
+                            return Err(err);
                         }
                     }
                 }
 
-                yield ReturnSuccess::value(ret);
+                ReturnSuccess::value(ret)
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {
diff --git a/crates/nu-cli/src/commands/str_/trim.rs b/crates/nu-cli/src/commands/str_/trim.rs
index b8e91da70..3d5cf5b2d 100644
--- a/crates/nu-cli/src/commands/str_/trim.rs
+++ b/crates/nu-cli/src/commands/str_/trim.rs
@@ -37,7 +37,7 @@ impl WholeStreamCommand for SubCommand {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        operate(args, registry)
+        operate(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -49,47 +49,46 @@ impl WholeStreamCommand for SubCommand {
     }
 }
 
-fn operate(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn operate(
+    args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (Arguments { rest }, mut input) = args.process(&registry).await?;
+    let (Arguments { rest }, input) = args.process(&registry).await?;
 
-        let column_paths: Vec<_> = rest.iter().map(|x| x.clone()).collect();
+    let column_paths: Vec<_> = rest;
 
-        while let Some(v) = input.next().await {
+    Ok(input
+        .map(move |v| {
             if column_paths.is_empty() {
                 match action(&v, v.tag()) {
-                    Ok(out) => yield ReturnSuccess::value(out),
-                    Err(err) => {
-                        yield Err(err);
-                        return;
-                    }
+                    Ok(out) => ReturnSuccess::value(out),
+                    Err(err) => Err(err),
                 }
             } else {
-
-                let mut ret = v.clone();
+                let mut ret = v;
 
                 for path in &column_paths {
-                    let swapping = ret.swap_data_by_column_path(path, Box::new(move |old| action(old, old.tag())));
+                    let swapping = ret.swap_data_by_column_path(
+                        path,
+                        Box::new(move |old| action(old, old.tag())),
+                    );
 
                     match swapping {
                         Ok(new_value) => {
                             ret = new_value;
                         }
                         Err(err) => {
-                            yield Err(err);
-                            return;
+                            return Err(err);
                         }
                     }
                 }
 
-                yield ReturnSuccess::value(ret);
+                ReturnSuccess::value(ret)
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .to_output_stream())
 }
 
 fn action(input: &Value, tag: impl Into<Tag>) -> Result<Value, ShellError> {
diff --git a/crates/nu-cli/src/commands/what.rs b/crates/nu-cli/src/commands/what.rs
index c8eabe027..3af287c0d 100644
--- a/crates/nu-cli/src/commands/what.rs
+++ b/crates/nu-cli/src/commands/what.rs
@@ -2,7 +2,7 @@ use crate::commands::WholeStreamCommand;
 use crate::prelude::*;
 
 use nu_errors::ShellError;
-use nu_protocol::{ReturnSuccess, ReturnValue, Signature, UntaggedValue};
+use nu_protocol::{ReturnSuccess, Signature, UntaggedValue};
 
 pub struct What;
 
@@ -28,23 +28,23 @@ impl WholeStreamCommand for What {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        // args.process(registry, what)?.run()
-        what(args, registry)
+        what(args, registry).await
     }
 }
 
-pub fn what(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let stream = async_stream! {
-        let mut input = args.input;
-        while let Some(row) = input.next().await {
+pub async fn what(
+    args: CommandArgs,
+    _registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
+    Ok(args
+        .input
+        .map(|row| {
             let name = value::format_type(&row, 100);
-            yield ReturnSuccess::value(UntaggedValue::string(name).into_value(Tag::unknown_anchor(row.tag.span)));
-        }
-    };
-
-    let stream: BoxStream<'static, ReturnValue> = stream.boxed();
-
-    Ok(OutputStream::from(stream))
+            ReturnSuccess::value(
+                UntaggedValue::string(name).into_value(Tag::unknown_anchor(row.tag.span)),
+            )
+        })
+        .to_output_stream())
 }
 
 #[cfg(test)]
diff --git a/crates/nu-cli/src/commands/wrap.rs b/crates/nu-cli/src/commands/wrap.rs
index 8f6ae3e3c..27ed89ebb 100644
--- a/crates/nu-cli/src/commands/wrap.rs
+++ b/crates/nu-cli/src/commands/wrap.rs
@@ -38,8 +38,7 @@ impl WholeStreamCommand for Wrap {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        // args.process(registry, wrap)?.run()
-        wrap(args, registry)
+        wrap(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -84,64 +83,57 @@ impl WholeStreamCommand for Wrap {
     }
 }
 
-fn wrap(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn wrap(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
 
-    let stream = async_stream! {
-        let (WrapArgs { column }, mut input) = args.process(&registry).await?;
-        let mut result_table = vec![];
-        let mut are_all_rows = true;
+    let (WrapArgs { column }, mut input) = args.process(&registry).await?;
+    let mut result_table = vec![];
+    let mut are_all_rows = true;
 
-        while let Some(value) = input.next().await {
-            match value {
-                Value {
-                    value: UntaggedValue::Row(_),
-                    ..
-                } => {
-                    result_table.push(value);
-                }
-                _ => {
-                    are_all_rows = false;
+    while let Some(value) = input.next().await {
+        match value {
+            Value {
+                value: UntaggedValue::Row(_),
+                ..
+            } => {
+                result_table.push(value);
+            }
+            _ => {
+                are_all_rows = false;
 
-                    let mut index_map = IndexMap::new();
-                    index_map.insert(
-                        match &column {
-                            Some(key) => key.item.clone(),
-                            None => DEFAULT_COLUMN_NAME.to_string(),
-                        },
-                        value,
-                    );
-
-                    result_table.push(UntaggedValue::row(index_map).into_value(Tag::unknown()));
-                }
+                let mut index_map = IndexMap::new();
+                index_map.insert(
+                    match &column {
+                        Some(key) => key.item.clone(),
+                        None => DEFAULT_COLUMN_NAME.to_string(),
+                    },
+                    value,
+                );
 
+                result_table.push(UntaggedValue::row(index_map).into_value(Tag::unknown()));
             }
         }
+    }
 
-        if are_all_rows {
-            let mut index_map = IndexMap::new();
-            index_map.insert(
-                match &column {
-                    Some(key) => key.item.clone(),
-                    None => DEFAULT_COLUMN_NAME.to_string(),
-                },
-                UntaggedValue::table(&result_table).into_value(Tag::unknown()),
-            );
+    if are_all_rows {
+        let mut index_map = IndexMap::new();
+        index_map.insert(
+            match &column {
+                Some(key) => key.item.clone(),
+                None => DEFAULT_COLUMN_NAME.to_string(),
+            },
+            UntaggedValue::table(&result_table).into_value(Tag::unknown()),
+        );
 
-            let row = UntaggedValue::row(index_map).into_untagged_value();
+        let row = UntaggedValue::row(index_map).into_untagged_value();
 
-            yield ReturnSuccess::value(row);
-        } else {
-            for item in result_table
-                .iter()
-                .map(|row| ReturnSuccess::value(row.clone())) {
-
-                yield item;
-            }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        Ok(OutputStream::one(ReturnSuccess::value(row)))
+    } else {
+        Ok(
+            futures::stream::iter(result_table.into_iter().map(ReturnSuccess::value))
+                .to_output_stream(),
+        )
+    }
 }
 
 #[cfg(test)]
diff --git a/crates/nu-cli/src/shell/filesystem_shell.rs b/crates/nu-cli/src/shell/filesystem_shell.rs
index 141fd00ce..4001e9b00 100644
--- a/crates/nu-cli/src/shell/filesystem_shell.rs
+++ b/crates/nu-cli/src/shell/filesystem_shell.rs
@@ -562,54 +562,54 @@ impl Shell for FilesystemShell {
             ));
         }
 
-        let stream = async_stream! {
-            for (f, tag) in all_targets.iter() {
+        Ok(
+            futures::stream::iter(all_targets.into_iter().map(move |(f, tag)| {
                 let is_empty = || match f.read_dir() {
                     Ok(mut p) => p.next().is_none(),
-                    Err(_) => false
+                    Err(_) => false,
                 };
 
                 if let Ok(metadata) = f.symlink_metadata() {
-                    if metadata.is_file() || metadata.file_type().is_symlink() || recursive.item || is_empty() {
+                    if metadata.is_file()
+                        || metadata.file_type().is_symlink()
+                        || recursive.item
+                        || is_empty()
+                    {
                         let result;
                         #[cfg(feature = "trash-support")]
                         {
-                            let rm_always_trash = config::config(Tag::unknown())?.get("rm_always_trash").map(|val| val.is_true()).unwrap_or(false);
+                            let rm_always_trash = config::config(Tag::unknown())?
+                                .get("rm_always_trash")
+                                .map(|val| val.is_true())
+                                .unwrap_or(false);
                             result = if _trash.item || (rm_always_trash && !_permanent.item) {
-                                trash::remove(f)
-                                   .map_err(|e| f.to_string_lossy())
+                                trash::remove(&f).map_err(|_| f.to_string_lossy())
                             } else if metadata.is_file() {
-                                std::fs::remove_file(f)
-                                    .map_err(|e| f.to_string_lossy())
+                                std::fs::remove_file(&f).map_err(|_| f.to_string_lossy())
                             } else {
-                                std::fs::remove_dir_all(f)
-                                    .map_err(|e| f.to_string_lossy())
+                                std::fs::remove_dir_all(&f).map_err(|_| f.to_string_lossy())
                             };
                         }
                         #[cfg(not(feature = "trash-support"))]
                         {
                             result = if metadata.is_file() {
-                                std::fs::remove_file(f)
-                                    .map_err(|e| f.to_string_lossy())
+                                std::fs::remove_file(&f).map_err(|_| f.to_string_lossy())
                             } else {
-                                std::fs::remove_dir_all(f)
-                                    .map_err(|e| f.to_string_lossy())
+                                std::fs::remove_dir_all(&f).map_err(|_| f.to_string_lossy())
                             };
                         }
 
                         if let Err(e) = result {
                             let msg = format!("Could not delete {:}", e);
-                            yield Err(ShellError::labeled_error(msg, e, tag))
+                            Err(ShellError::labeled_error(msg, e, tag))
                         } else {
                             let val = format!("deleted {:}", f.to_string_lossy()).into();
-                            yield Ok(ReturnSuccess::Value(val))
+                            Ok(ReturnSuccess::Value(val))
                         }
                     } else {
-                        let msg = format!(
-                            "Cannot remove {:}. try --recursive",
-                            f.to_string_lossy()
-                        );
-                        yield Err(ShellError::labeled_error(
+                        let msg =
+                            format!("Cannot remove {:}. try --recursive", f.to_string_lossy());
+                        Err(ShellError::labeled_error(
                             msg,
                             "cannot remove non-empty directory",
                             tag,
@@ -617,16 +617,15 @@ impl Shell for FilesystemShell {
                     }
                 } else {
                     let msg = format!("no such file or directory: {:}", f.to_string_lossy());
-                    yield Err(ShellError::labeled_error(
+                    Err(ShellError::labeled_error(
                         msg,
                         "no such file or directory",
                         tag,
                     ))
                 }
-            }
-        };
-
-        Ok(stream.to_output_stream())
+            }))
+            .to_output_stream(),
+        )
     }
 
     fn path(&self) -> String {