Batch of moving commands off async_stream #3 (#1919)

* Batch of moving commands off async_stream #3

* remove commented-out section

* merge master

crates/nu-cli/src/cli.rs

@@ -909,6 +909,7 @@ async fn process_line(
                         shell_manager: ctx.shell_manager.clone(),
                         host: ctx.host.clone(),
                         ctrl_c: ctx.ctrl_c.clone(),
+                        current_errors: ctx.current_errors.clone(),
                         registry: ctx.registry.clone(),
                         name: Tag::unknown(),
                         raw_input: line.to_string(),

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

@@ -5,6 +5,7 @@ use crate::prelude::*;
 use nu_errors::ShellError;
 use nu_protocol::{hir, hir::Expression, hir::Literal, hir::SpannedExpression};
 use nu_protocol::{Primitive, Scope, Signature, UntaggedValue, Value};
+use parking_lot::Mutex;
 use prettytable::format::{FormatBuilder, LinePosition, LineSeparator};
 use prettytable::{color, Attr, Cell, Row, Table};
 use std::sync::atomic::AtomicBool;
@@ -38,6 +39,7 @@ impl WholeStreamCommand for Autoview {
             shell_manager: args.shell_manager,
             host: args.host,
             ctrl_c: args.ctrl_c,
+            current_errors: args.current_errors,
             name: args.call_info.name_tag,
             raw_input: args.raw_input,
         })
@@ -63,6 +65,7 @@ impl WholeStreamCommand for Autoview {
 pub struct RunnableContextWithoutInput {
     pub shell_manager: ShellManager,
     pub host: Arc<parking_lot::Mutex<Box<dyn Host>>>,
+    pub current_errors: Arc<Mutex<Vec<ShellError>>>,
     pub ctrl_c: Arc<AtomicBool>,
     pub registry: CommandRegistry,
     pub name: Tag,
@@ -74,6 +77,7 @@ impl RunnableContextWithoutInput {
             shell_manager: context.shell_manager,
             host: context.host,
             ctrl_c: context.ctrl_c,
+            current_errors: context.current_errors,
             registry: context.registry,
             name: context.name,
         };
@@ -389,6 +393,7 @@ fn create_default_command_args(context: &RunnableContextWithoutInput) -> RawComm
     RawCommandArgs {
         host: context.host.clone(),
         ctrl_c: context.ctrl_c.clone(),
+        current_errors: context.current_errors.clone(),
         shell_manager: context.shell_manager.clone(),
         call_info: UnevaluatedCallInfo {
             args: hir::Call {

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

@@ -38,6 +38,7 @@ impl WholeStreamCommand for Average {
             shell_manager: args.shell_manager,
             host: args.host,
             ctrl_c: args.ctrl_c,
+            current_errors: args.current_errors,
             name: args.call_info.name_tag,
             raw_input: args.raw_input,
         })

crates/nu-cli/src/commands/classified/internal.rs

@@ -66,6 +66,7 @@ pub(crate) async fn run_internal_command(
                             let new_args = RawCommandArgs {
                                 host: context.host.clone(),
                                 ctrl_c: context.ctrl_c.clone(),
+                                current_errors: context.current_errors.clone(),
                                 shell_manager: context.shell_manager.clone(),
                                 call_info: UnevaluatedCallInfo {
                                     args: nu_protocol::hir::Call {

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

@@ -8,6 +8,7 @@ use getset::Getters;
 use nu_errors::ShellError;
 use nu_protocol::hir;
 use nu_protocol::{CallInfo, EvaluatedArgs, ReturnSuccess, Scope, Signature, UntaggedValue, Value};
+use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
 use std::ops::Deref;
 use std::sync::atomic::AtomicBool;
@@ -54,6 +55,7 @@ impl UnevaluatedCallInfo {
 pub struct CommandArgs {
     pub host: Arc<parking_lot::Mutex<Box<dyn Host>>>,
     pub ctrl_c: Arc<AtomicBool>,
+    pub current_errors: Arc<Mutex<Vec<ShellError>>>,
     pub shell_manager: ShellManager,
     pub call_info: UnevaluatedCallInfo,
     pub input: InputStream,
@@ -65,6 +67,7 @@ pub struct CommandArgs {
 pub struct RawCommandArgs {
     pub host: Arc<parking_lot::Mutex<Box<dyn Host>>>,
     pub ctrl_c: Arc<AtomicBool>,
+    pub current_errors: Arc<Mutex<Vec<ShellError>>>,
     pub shell_manager: ShellManager,
     pub call_info: UnevaluatedCallInfo,
 }
@@ -74,6 +77,7 @@ impl RawCommandArgs {
         CommandArgs {
             host: self.host,
             ctrl_c: self.ctrl_c,
+            current_errors: self.current_errors,
             shell_manager: self.shell_manager,
             call_info: self.call_info,
             input: input.into(),
@@ -151,6 +155,7 @@ pub struct RunnableContext {
     pub shell_manager: ShellManager,
     pub host: Arc<parking_lot::Mutex<Box<dyn Host>>>,
     pub ctrl_c: Arc<AtomicBool>,
+    pub current_errors: Arc<Mutex<Vec<ShellError>>>,
     pub registry: CommandRegistry,
     pub name: Tag,
     pub raw_input: String,
@@ -342,10 +347,9 @@ impl Command {
         if args.call_info.switch_present("help") {
             let cl = self.0.clone();
             let registry = registry.clone();
-            let stream = async_stream! {
-                yield Ok(ReturnSuccess::Value(UntaggedValue::string(get_help(&*cl, &registry)).into_value(Tag::unknown())));
-            };
-            stream.to_output_stream()
+            OutputStream::one(Ok(ReturnSuccess::Value(
+                UntaggedValue::string(get_help(&*cl, &registry)).into_value(Tag::unknown()),
+            )))
         } else {
             match self.0.run(args, registry).await {
                 Ok(stream) => stream,

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

@@ -6,8 +6,8 @@ use crate::prelude::*;
 use futures::stream::once;
 use nu_errors::ShellError;
 use nu_protocol::{
-    hir::Block, hir::Expression, hir::SpannedExpression, hir::Synthetic, ReturnSuccess, Signature,
-    SyntaxShape, UntaggedValue,
+    hir::Block, hir::Expression, hir::SpannedExpression, hir::Synthetic, Scope, Signature,
+    SyntaxShape, UntaggedValue, Value,
 };
 
 pub struct Each;
@@ -40,7 +40,7 @@ impl WholeStreamCommand for Each {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        each(args, registry)
+        each(args, registry).await
     }
 
     fn examples(&self) -> Vec<Example> {
@@ -76,50 +76,56 @@ fn is_expanded_it_usage(head: &SpannedExpression) -> bool {
     }
 }
 
-fn each(raw_args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+async fn process_row(
+    block: Arc<Block>,
+    scope: Arc<Scope>,
+    head: Arc<Box<SpannedExpression>>,
+    mut context: Arc<Context>,
+    input: Value,
+) -> Result<OutputStream, ShellError> {
+    let input_clone = input.clone();
+    let input_stream = if is_expanded_it_usage(&head) {
+        InputStream::empty()
+    } else {
+        once(async { Ok(input_clone) }).to_input_stream()
+    };
+    Ok(run_block(
+        &block,
+        Arc::make_mut(&mut context),
+        input_stream,
+        &input,
+        &scope.vars,
+        &scope.env,
+    )
+    .await?
+    .to_output_stream())
+}
+
+async fn each(
+    raw_args: CommandArgs,
+    registry: &CommandRegistry,
+) -> Result<OutputStream, ShellError> {
     let registry = registry.clone();
-    let stream = async_stream! {
-        let head = raw_args.call_info.args.head.clone();
-        let scope = raw_args.call_info.scope.clone();
-        let mut context = Context::from_raw(&raw_args, &registry);
-        let (each_args, mut input): (EachArgs, _) = raw_args.process(&registry).await?;
-        let block = each_args.block;
-        while let Some(input) = input.next().await {
-            let input_clone = input.clone();
-            let input_stream = if is_expanded_it_usage(&head) {
-                InputStream::empty()
-            } else {
-                once(async { Ok(input_clone) }).to_input_stream()
-            };
-
-            let result = run_block(
-                &block,
-                &mut context,
-                input_stream,
-                &input,
-                &scope.vars,
-                &scope.env
-            ).await;
-
-            match result {
-                Ok(mut stream) => {
-                    while let Some(result) = stream.next().await {
-                        yield Ok(ReturnSuccess::Value(result));
-                    }
-
-                    let errors = context.get_errors();
-                    if let Some(error) = errors.first() {
-                        yield Err(error.clone());
-                    }
-                }
-                Err(e) => {
-                    yield Err(e);
+    let head = Arc::new(raw_args.call_info.args.head.clone());
+    let scope = Arc::new(raw_args.call_info.scope.clone());
+    let context = Arc::new(Context::from_raw(&raw_args, &registry));
+    let (each_args, input): (EachArgs, _) = raw_args.process(&registry).await?;
+    let block = Arc::new(each_args.block);
+    Ok(input
+        .then(move |input| {
+            let block = block.clone();
+            let scope = scope.clone();
+            let head = head.clone();
+            let context = context.clone();
+            async {
+                match process_row(block, scope, head, context, input).await {
+                    Ok(s) => s,
+                    Err(e) => OutputStream::one(Err(e)),
                 }
             }
-        }
-    };
-
-    Ok(stream.to_output_stream())
+        })
+        .flatten()
+        .to_output_stream())
 }
 
 #[cfg(test)]

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

@@ -65,6 +65,7 @@ fn enter(raw_args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStre
         let shell_manager = raw_args.shell_manager.clone();
         let head = raw_args.call_info.args.head.clone();
         let ctrl_c = raw_args.ctrl_c.clone();
+        let current_errors = raw_args.current_errors.clone();
         let host = raw_args.host.clone();
         let tag = raw_args.call_info.name_tag.clone();
         let (EnterArgs { location }, _) = raw_args.process(&registry).await?;
@@ -116,6 +117,7 @@ fn enter(raw_args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStre
                             let new_args = RawCommandArgs {
                                 host,
                                 ctrl_c,
+                                current_errors,
                                 shell_manager,
                                 call_info: UnevaluatedCallInfo {
                                     args: nu_protocol::hir::Call {

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

@@ -166,6 +166,7 @@ fn save(raw_args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStrea
     let registry = registry.clone();
     let host = raw_args.host.clone();
     let ctrl_c = raw_args.ctrl_c.clone();
+    let current_errors = raw_args.current_errors.clone();
     let shell_manager = raw_args.shell_manager.clone();
 
     let stream = async_stream! {
@@ -220,6 +221,7 @@ fn save(raw_args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStrea
                         let new_args = RawCommandArgs {
                             host,
                             ctrl_c,
+                            current_errors,
                             shell_manager,
                             call_info: UnevaluatedCallInfo {
                                 args: nu_protocol::hir::Call {

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

@@ -34,6 +34,7 @@ impl WholeStreamCommand for Sum {
             shell_manager: args.shell_manager,
             host: args.host,
             ctrl_c: args.ctrl_c,
+            current_errors: args.current_errors,
             name: args.call_info.name_tag,
             raw_input: args.raw_input,
         })

crates/nu-cli/src/context.rs

@@ -93,7 +93,7 @@ impl Context {
             Context {
                 registry: registry.clone(),
                 host: raw_args.host.clone(),
-                current_errors: Arc::new(Mutex::new(vec![])),
+                current_errors: raw_args.current_errors.clone(),
                 ctrl_c: raw_args.ctrl_c.clone(),
                 shell_manager: raw_args.shell_manager.clone(),
                 windows_drives_previous_cwd: Arc::new(Mutex::new(std::collections::HashMap::new())),
@@ -105,7 +105,7 @@ impl Context {
             Context {
                 registry: registry.clone(),
                 host: raw_args.host.clone(),
-                current_errors: Arc::new(Mutex::new(vec![])),
+                current_errors: raw_args.current_errors.clone(),
                 ctrl_c: raw_args.ctrl_c.clone(),
                 shell_manager: raw_args.shell_manager.clone(),
                 raw_input: String::default(),
@@ -119,7 +119,7 @@ impl Context {
             Context {
                 registry: registry.clone(),
                 host: args.host.clone(),
-                current_errors: Arc::new(Mutex::new(vec![])),
+                current_errors: args.current_errors.clone(),
                 ctrl_c: args.ctrl_c.clone(),
                 shell_manager: args.shell_manager.clone(),
                 windows_drives_previous_cwd: Arc::new(Mutex::new(std::collections::HashMap::new())),
@@ -131,7 +131,7 @@ impl Context {
             Context {
                 registry: registry.clone(),
                 host: args.host.clone(),
-                current_errors: Arc::new(Mutex::new(vec![])),
+                current_errors: args.current_errors.clone(),
                 ctrl_c: args.ctrl_c.clone(),
                 shell_manager: args.shell_manager.clone(),
                 raw_input: String::default(),
@@ -255,6 +255,7 @@ impl Context {
         CommandArgs {
             host: self.host.clone(),
             ctrl_c: self.ctrl_c.clone(),
+            current_errors: self.current_errors.clone(),
             shell_manager: self.shell_manager.clone(),
             call_info: self.call_info(args, name_tag, scope),
             input,