Reduce unwraps

Remove a number of unwraps. In some cases, a `?` just worked as is. I also made it possible to use `?` to go from Result<OutputStream, ShellError> to OutputStream. Finally, started updating PerItemCommand to be able to use the signature deserialization logic, which substantially reduces unwraps.

This is still in-progress work, but tests pass and it should be clear to merge and keep iterating on master.

src/cli.rs

@@ -52,9 +52,9 @@ fn load_plugin(path: &std::path::Path, context: &mut Context) -> Result<(), Shel
     let mut reader = BufReader::new(stdout);
 
     let request = JsonRpc::new("config", Vec::<Value>::new());
-    let request_raw = serde_json::to_string(&request).unwrap();
+    let request_raw = serde_json::to_string(&request)?;
     stdin.write(format!("{}\n", request_raw).as_bytes())?;
-    let path = dunce::canonicalize(path).unwrap();
+    let path = dunce::canonicalize(path)?;
 
     let mut input = String::new();
     match reader.read_line(&mut input) {
@@ -90,13 +90,13 @@ fn load_plugin(path: &std::path::Path, context: &mut Context) -> Result<(), Shel
 }
 
 fn load_plugins_in_dir(path: &std::path::PathBuf, context: &mut Context) -> Result<(), ShellError> {
-    let re_bin = Regex::new(r"^nu_plugin_[A-Za-z_]+$").unwrap();
+    let re_bin = Regex::new(r"^nu_plugin_[A-Za-z_]+$")?;
-    let re_exe = Regex::new(r"^nu_plugin_[A-Za-z_]+\.exe$").unwrap();
+    let re_exe = Regex::new(r"^nu_plugin_[A-Za-z_]+\.exe$")?;
 
     match std::fs::read_dir(path) {
         Ok(p) => {
             for entry in p {
-                let entry = entry.unwrap();
+                let entry = entry?;
                 let filename = entry.file_name();
                 let f_name = filename.to_string_lossy();
                 if re_bin.is_match(&f_name) || re_exe.is_match(&f_name) {
@@ -270,8 +270,7 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
                     &files,
                     &diag,
                     &language_reporting::DefaultConfig,
-                )
+                )?;
-                .unwrap();
             }
 
             LineResult::Break => {
@@ -288,7 +287,7 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
         }
         ctrlcbreak = false;
     }
-    rl.save_history("history.txt").unwrap();
+    rl.save_history("history.txt")?;
 
     Ok(())
 }

src/commands/autoview.rs

@@ -17,7 +17,7 @@ impl WholeStreamCommand for Autoview {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        args.process_raw(registry, autoview)?.run()
+        Ok(args.process_raw(registry, autoview)?.run())
     }
 
     fn signature(&self) -> Signature {
@@ -40,28 +40,20 @@ pub fn autoview(
             } = input[0usize]
             {
                 let binary = context.expect_command("binaryview");
-                let result = binary.run(raw.with_input(input), &context.commands).await.unwrap();
+                let result = binary.run(raw.with_input(input), &context.commands);
                 result.collect::<Vec<_>>().await;
             } else if is_single_text_value(&input) {
                 let text = context.expect_command("textview");
-                let result = text.run(raw.with_input(input), &context.commands).await.unwrap();
+                let result = text.run(raw.with_input(input), &context.commands);
                 result.collect::<Vec<_>>().await;
             } else if equal_shapes(&input) {
                 let table = context.expect_command("table");
-                let result = table.run(raw.with_input(input), &context.commands).await.unwrap();
+                let result = table.run(raw.with_input(input), &context.commands);
                 result.collect::<Vec<_>>().await;
             } else {
                 let table = context.expect_command("table");
-                let result = table.run(raw.with_input(input), &context.commands).await.unwrap();
+                let result = table.run(raw.with_input(input), &context.commands);
                 result.collect::<Vec<_>>().await;
-                //println!("TODO!")
-                // TODO
-                // let mut host = context.host.lock().unwrap();
-                // for i in input.iter() {
-                //     let view = GenericView::new(&i);
-                //     handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host));
-                //     host.stdout("");
-                // }
             }
         }
     }))

src/commands/classified.rs

@@ -117,16 +117,14 @@ impl InternalCommand {
         let objects: InputStream =
             trace_stream!(target: "nu::trace_stream::internal", "input" = input.objects);
 
-        let result = context
+        let result = context.run_command(
-            .run_command(
+            self.command,
-                self.command,
+            self.name_span.clone(),
-                self.name_span.clone(),
+            context.source_map.clone(),
-                context.source_map.clone(),
+            self.args,
-                self.args,
+            source,
-                source,
+            objects,
-                objects,
+        );
-            )
-            .await?;
 
         let mut result = result.values;
 
@@ -285,7 +283,7 @@ impl ExternalCommand {
                             continue;
                         }
 
-                        process = process.arg(&arg.replace("$it", &i.as_string().unwrap()));
+                        process = process.arg(&arg.replace("$it", &i.as_string()?));
                     }
                 }
             } else {

src/commands/command.rs

@@ -77,6 +77,25 @@ pub struct CallInfo {
     pub name_span: Span,
 }
 
+impl CallInfo {
+    pub fn process<'de, T: Deserialize<'de>>(
+        &self,
+        shell_manager: &ShellManager,
+        callback: fn(T, &RunnablePerItemContext) -> Result<VecDeque<ReturnValue>, ShellError>,
+    ) -> Result<RunnablePerItemArgs<T>, ShellError> {
+        let mut deserializer = ConfigDeserializer::from_call_info(self.clone());
+
+        Ok(RunnablePerItemArgs {
+            args: T::deserialize(&mut deserializer)?,
+            context: RunnablePerItemContext {
+                shell_manager: shell_manager.clone(),
+                name: self.name_span,
+            },
+            callback,
+        })
+    }
+}
+
 #[derive(Getters)]
 #[get = "crate"]
 pub struct CommandArgs {
@@ -147,7 +166,7 @@ impl CommandArgs {
         let args = self.evaluate_once(registry)?;
         let (input, args) = args.split();
         let name_span = args.call_info.name_span;
-        let mut deserializer = ConfigDeserializer::from_call_node(args);
+        let mut deserializer = ConfigDeserializer::from_call_info(args.call_info);
 
         Ok(RunnableArgs {
             args: T::deserialize(&mut deserializer)?,
@@ -180,7 +199,7 @@ impl CommandArgs {
         let args = self.evaluate_once(registry)?;
         let (input, args) = args.split();
         let name_span = args.call_info.name_span;
-        let mut deserializer = ConfigDeserializer::from_call_node(args);
+        let mut deserializer = ConfigDeserializer::from_call_info(args.call_info);
 
         Ok(RunnableRawArgs {
             args: T::deserialize(&mut deserializer)?,
@@ -198,6 +217,17 @@ impl CommandArgs {
     }
 }
 
+pub struct RunnablePerItemContext {
+    pub shell_manager: ShellManager,
+    pub name: Span,
+}
+
+impl RunnablePerItemContext {
+    pub fn cwd(&self) -> PathBuf {
+        PathBuf::from(self.shell_manager.path())
+    }
+}
+
 pub struct RunnableContext {
     pub input: InputStream,
     pub shell_manager: ShellManager,
@@ -219,6 +249,18 @@ impl RunnableContext {
     }
 }
 
+pub struct RunnablePerItemArgs<T> {
+    args: T,
+    context: RunnablePerItemContext,
+    callback: fn(T, &RunnablePerItemContext) -> Result<VecDeque<ReturnValue>, ShellError>,
+}
+
+impl<T> RunnablePerItemArgs<T> {
+    pub fn run(self) -> Result<VecDeque<ReturnValue>, ShellError> {
+        (self.callback)(self.args, &self.context)
+    }
+}
+
 pub struct RunnableArgs<T> {
     args: T,
     context: RunnableContext,
@@ -239,8 +281,11 @@ pub struct RunnableRawArgs<T> {
 }
 
 impl<T> RunnableRawArgs<T> {
-    pub fn run(self) -> Result<OutputStream, ShellError> {
+    pub fn run(self) -> OutputStream {
-        (self.callback)(self.args, self.context, self.raw_args)
+        match (self.callback)(self.args, self.context, self.raw_args) {
+            Ok(stream) => stream,
+            Err(err) => OutputStream::one(Err(err)),
+        }
     }
 }
 
@@ -475,13 +520,12 @@ impl Command {
         }
     }
 
-    pub async fn run(
+    pub fn run(&self, args: CommandArgs, registry: &registry::CommandRegistry) -> OutputStream {
-        &self,
-        args: CommandArgs,
-        registry: &registry::CommandRegistry,
-    ) -> Result<OutputStream, ShellError> {
         match self {
-            Command::WholeStream(command) => command.run(args, registry),
+            Command::WholeStream(command) => match command.run(args, registry) {
+                Ok(stream) => stream,
+                Err(err) => OutputStream::one(Err(err)),
+            },
             Command::PerItem(command) => self.run_helper(command.clone(), args, registry.clone()),
         }
     }
@@ -491,7 +535,7 @@ impl Command {
         command: Arc<dyn PerItemCommand>,
         args: CommandArgs,
         registry: CommandRegistry,
-    ) -> Result<OutputStream, ShellError> {
+    ) -> OutputStream {
         let raw_args = RawCommandArgs {
             host: args.host,
             shell_manager: args.shell_manager,
@@ -515,7 +559,7 @@ impl Command {
                 })
                 .flatten();
 
-            Ok(out.to_output_stream())
+            out.to_output_stream()
         } else {
             let nothing = Value::nothing().tagged(Tag::unknown());
             let call_info = raw_args
@@ -524,11 +568,15 @@ impl Command {
                 .evaluate(&registry, &Scope::it_value(nothing.clone()))
                 .unwrap();
             // We don't have an $it or block, so just execute what we have
-            let out = match command.run(&call_info, &registry, &raw_args.shell_manager, nothing) {
+
-                Ok(o) => o,
+            command
-                Err(e) => VecDeque::from(vec![ReturnValue::Err(e)]),
+                .run(&call_info, &registry, &raw_args.shell_manager, nothing)?
-            };
+                .into()
-            Ok(out.to_output_stream())
+            // let out = match command.run(&call_info, &registry, &raw_args.shell_manager, nothing) {
+            //     Ok(o) => o,
+            //     Err(e) => VecDeque::from(vec![ReturnValue::Err(e)]),
+            // };
+            // Ok(out.to_output_stream())
         }
     }
 }

src/commands/cp.rs

@@ -1,3 +1,4 @@
+use crate::commands::command::RunnablePerItemContext;
 use crate::errors::ShellError;
 use crate::parser::hir::SyntaxType;
 use crate::parser::registry::{CommandRegistry, Signature};
@@ -7,15 +8,22 @@ use std::path::PathBuf;
 
 pub struct Cpy;
 
+#[derive(Deserialize)]
+pub struct CopyArgs {
+    source: Tagged<PathBuf>,
+    destination: Tagged<PathBuf>,
+    recursive: Tagged<bool>,
+}
+
 impl PerItemCommand for Cpy {
     fn run(
         &self,
         call_info: &CallInfo,
         _registry: &CommandRegistry,
         shell_manager: &ShellManager,
-        _input: Tagged<Value>,
+        input: Tagged<Value>,
     ) -> Result<VecDeque<ReturnValue>, ShellError> {
-        cp(call_info, shell_manager)
+        call_info.process(shell_manager, cp)?.run()
     }
 
     fn name(&self) -> &str {
@@ -24,42 +32,24 @@ impl PerItemCommand for Cpy {
 
     fn signature(&self) -> Signature {
         Signature::build("cp")
+            .required("source", SyntaxType::Path)
+            .required("destination", SyntaxType::Path)
             .named("file", SyntaxType::Any)
             .switch("recursive")
     }
 }
 
 pub fn cp(
-    call_info: &CallInfo,
+    args: CopyArgs,
-    shell_manager: &ShellManager,
+    context: &RunnablePerItemContext,
 ) -> Result<VecDeque<ReturnValue>, ShellError> {
-    let mut source = PathBuf::from(shell_manager.path());
+    let mut source = PathBuf::from(context.shell_manager.path());
-    let mut destination = PathBuf::from(shell_manager.path());
+    let mut destination = PathBuf::from(context.shell_manager.path());
-    let name_span = call_info.name_span;
+    let name_span = context.name;
 
-    match call_info
+    source.push(&args.source.item);
-        .args
-        .nth(0)
-        .ok_or_else(|| ShellError::string(&format!("No file or directory specified")))?
-        .as_string()?
-        .as_str()
-    {
-        file => {
-            source.push(file);
-        }
-    }
 
-    match call_info
+    destination.push(&args.destination.item);
-        .args
-        .nth(1)
-        .ok_or_else(|| ShellError::string(&format!("No file or directory specified")))?
-        .as_string()?
-        .as_str()
-    {
-        file => {
-            destination.push(file);
-        }
-    }
 
     let sources = glob::glob(&source.to_string_lossy());
 
@@ -67,7 +57,7 @@ pub fn cp(
         return Err(ShellError::labeled_error(
             "Invalid pattern.",
             "Invalid pattern.",
-            call_info.args.nth(0).unwrap().span(),
+            args.source.tag,
         ));
     }
 
@@ -75,11 +65,11 @@ pub fn cp(
 
     if sources.len() == 1 {
         if let Ok(entry) = &sources[0] {
-            if entry.is_dir() && !call_info.args.has("recursive") {
+            if entry.is_dir() && !args.recursive.item {
                 return Err(ShellError::labeled_error(
                     "is a directory (not copied). Try using \"--recursive\".",
                     "is a directory (not copied). Try using \"--recursive\".",
-                    call_info.args.nth(0).unwrap().span(),
+                    args.source.tag,
                 ));
             }
 
@@ -248,7 +238,7 @@ pub fn cp(
                 return Err(ShellError::labeled_error(
                     "Copy aborted (directories found). Recursive copying in patterns not supported yet (try copying the directory directly)",
                     "Copy aborted (directories found). Recursive copying in patterns not supported yet (try copying the directory directly)",
-                    call_info.args.nth(0).unwrap().span(),
+                    args.source.tag,
                 ));
             }
 
@@ -263,7 +253,7 @@ pub fn cp(
                                 return Err(ShellError::labeled_error(
                                     e.to_string(),
                                     e.to_string(),
-                                    call_info.args.nth(0).unwrap().span(),
+                                    args.source.tag,
                                 ));
                             }
                             Ok(o) => o,
@@ -281,7 +271,7 @@ pub fn cp(
                     "Copy aborted. (Does {:?} exist?)",
                     &destination.file_name().unwrap()
                 ),
-                call_info.args.nth(1).unwrap().span(),
+                args.destination.span(),
             ));
         }
     }

src/context.rs

@@ -115,7 +115,7 @@ impl Context {
         self.registry.get_command(name).unwrap()
     }
 
-    crate async fn run_command<'a>(
+    crate fn run_command<'a>(
         &mut self,
         command: Arc<Command>,
         name_span: Span,
@@ -123,9 +123,9 @@ impl Context {
         args: hir::Call,
         source: Text,
         input: InputStream,
-    ) -> Result<OutputStream, ShellError> {
+    ) -> OutputStream {
         let command_args = self.command_args(args, input, source, source_map, name_span);
-        command.run(command_args, self.registry()).await
+        command.run(command_args, self.registry())
     }
 
     fn call_info(

src/errors.rs

@@ -106,6 +106,14 @@ impl ShellError {
         ProximateShellError::MissingProperty { subpath, expr }.start()
     }
 
+    crate fn missing_value(span: Option<Span>, reason: impl Into<String>) -> ShellError {
+        ProximateShellError::MissingValue {
+            span,
+            reason: reason.into(),
+        }
+        .start()
+    }
+
     crate fn argument_error(
         command: impl Into<String>,
         kind: ArgumentError,
@@ -148,6 +156,18 @@ impl ShellError {
                 Diagnostic::new(Severity::Error, "Invalid command")
                     .with_label(Label::new_primary(command.span))
             }
+            ProximateShellError::MissingValue { span, reason } => {
+                let mut d = Diagnostic::new(
+                    Severity::Bug,
+                    format!("Internal Error (missing value) :: {}", reason),
+                );
+
+                if let Some(span) = span {
+                    d = d.with_label(Label::new_primary(span));
+                }
+
+                d
+            }
             ProximateShellError::ArgumentError {
                 command,
                 error,
@@ -245,11 +265,11 @@ impl ShellError {
     pub fn labeled_error(
         msg: impl Into<String>,
         label: impl Into<String>,
-        span: Span,
+        span: impl Into<Span>,
     ) -> ShellError {
         ShellError::diagnostic(
             Diagnostic::new(Severity::Error, msg.into())
-                .with_label(Label::new_primary(span).with_message(label.into())),
+                .with_label(Label::new_primary(span.into()).with_message(label.into())),
         )
     }
 
@@ -299,6 +319,10 @@ pub enum ProximateShellError {
         subpath: Description,
         expr: Description,
     },
+    MissingValue {
+        span: Option<Span>,
+        reason: String,
+    },
     ArgumentError {
         command: String,
         error: ArgumentError,
@@ -372,6 +396,7 @@ impl std::fmt::Display for ShellError {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         match &self.error {
             ProximateShellError::String(s) => write!(f, "{}", &s.title),
+            ProximateShellError::MissingValue { .. } => write!(f, "MissingValue"),
             ProximateShellError::InvalidCommand { .. } => write!(f, "InvalidCommand"),
             ProximateShellError::TypeError { .. } => write!(f, "TypeError"),
             ProximateShellError::SyntaxError { .. } => write!(f, "SyntaxError"),
@@ -414,3 +439,36 @@ impl std::convert::From<toml::ser::Error> for ShellError {
         .start()
     }
 }
+
+impl std::convert::From<serde_json::Error> for ShellError {
+    fn from(input: serde_json::Error) -> ShellError {
+        ProximateShellError::String(StringError {
+            title: format!("{:?}", input),
+            error: Value::nothing(),
+        })
+        .start()
+    }
+}
+
+impl std::convert::From<regex::Error> for ShellError {
+    fn from(input: regex::Error) -> ShellError {
+        ProximateShellError::String(StringError {
+            title: format!("{:?}", input),
+            error: Value::nothing(),
+        })
+        .start()
+    }
+}
+
+pub trait ShellErrorUtils<T> {
+    fn unwrap_error(self, desc: impl Into<String>) -> Result<T, ShellError>;
+}
+
+impl<T> ShellErrorUtils<Tagged<T>> for Option<Tagged<T>> {
+    fn unwrap_error(self, desc: impl Into<String>) -> Result<Tagged<T>, ShellError> {
+        match self {
+            Some(value) => Ok(value),
+            None => Err(ShellError::missing_value(None, desc.into())),
+        }
+    }
+}

src/object/meta.rs

@@ -197,6 +197,18 @@ impl From<&Span> for Tag {
     }
 }
 
+impl From<Tag> for Span {
+    fn from(tag: Tag) -> Self {
+        tag.span
+    }
+}
+
+impl From<&Tag> for Span {
+    fn from(tag: &Tag) -> Self {
+        tag.span
+    }
+}
+
 impl Tag {
     pub fn unknown_origin(span: Span) -> Tag {
         Tag { origin: None, span }

src/parser/deserializer.rs

@@ -1,4 +1,5 @@
 use crate::commands::command::EvaluatedCommandArgs;
+use crate::parser::registry::EvaluatedArgs;
 use crate::prelude::*;
 use log::trace;
 use serde::{de, forward_to_deserialize_any};
@@ -11,16 +12,16 @@ pub struct DeserializerItem<'de> {
 }
 
 pub struct ConfigDeserializer<'de> {
-    args: EvaluatedCommandArgs,
+    call: CallInfo,
     stack: Vec<DeserializerItem<'de>>,
     saw_root: bool,
     position: usize,
 }
 
 impl ConfigDeserializer<'de> {
-    pub fn from_call_node(args: EvaluatedCommandArgs) -> ConfigDeserializer<'de> {
+    pub fn from_call_info(call: CallInfo) -> ConfigDeserializer<'de> {
         ConfigDeserializer {
-            args,
+            call,
             stack: vec![],
             saw_root: false,
             position: 0,
@@ -29,16 +30,16 @@ impl ConfigDeserializer<'de> {
 
     pub fn push(&mut self, name: &'static str) -> Result<(), ShellError> {
         let value: Option<Tagged<Value>> = if name == "rest" {
-            let positional = self.args.slice_from(self.position);
+            let positional = self.call.args.slice_from(self.position);
             self.position += positional.len();
             Some(Value::List(positional).tagged_unknown()) // TODO: correct span
         } else {
-            if self.args.has(name) {
+            if self.call.args.has(name) {
-                self.args.get(name).map(|x| x.clone())
+                self.call.args.get(name).map(|x| x.clone())
             } else {
                 let position = self.position;
                 self.position += 1;
-                self.args.nth(position).map(|x| x.clone())
+                self.call.args.nth(position).map(|x| x.clone())
             }
         };
 
@@ -48,7 +49,7 @@ impl ConfigDeserializer<'de> {
             key: name.to_string(),
             struct_field: name,
             val: value.unwrap_or_else(|| {
-                Value::nothing().tagged(Tag::unknown_origin(self.args.call_info.name_span))
+                Value::nothing().tagged(Tag::unknown_origin(self.call.name_span))
             }),
         });
 

src/parser/registry.rs

@@ -142,6 +142,17 @@ pub struct EvaluatedArgs {
     pub named: Option<IndexMap<String, Tagged<Value>>>,
 }
 
+impl EvaluatedArgs {
+    pub fn slice_from(&self, from: usize) -> Vec<Tagged<Value>> {
+        let positional = &self.positional;
+
+        match positional {
+            None => vec![],
+            Some(list) => list[from..].to_vec(),
+        }
+    }
+}
+
 #[derive(new)]
 pub struct DebugEvaluatedPositional<'a> {
     positional: &'a Option<Vec<Tagged<Value>>>,

src/prelude.rs

@@ -41,7 +41,7 @@ crate use crate::context::CommandRegistry;
 crate use crate::context::{Context, SpanSource};
 crate use crate::env::host::handle_unexpected;
 crate use crate::env::Host;
-crate use crate::errors::ShellError;
+crate use crate::errors::{ShellError, ShellErrorUtils};
 crate use crate::object::base as value;
 crate use crate::object::meta::{Tag, Tagged, TaggedItem};
 crate use crate::object::types::ExtractType;

src/shell/filesystem_shell.rs

@@ -7,12 +7,19 @@ use crate::shell::shell::Shell;
 use rustyline::completion::FilenameCompleter;
 use rustyline::hint::{Hinter, HistoryHinter};
 use std::path::{Path, PathBuf};
+
 pub struct FilesystemShell {
     crate path: String,
     completer: NuCompleter,
     hinter: HistoryHinter,
 }
 
+impl std::fmt::Debug for FilesystemShell {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "FilesystemShell @ {}", self.path)
+    }
+}
+
 impl Clone for FilesystemShell {
     fn clone(&self) -> Self {
         FilesystemShell {

src/shell/shell.rs

@@ -3,7 +3,7 @@ use crate::context::SourceMap;
 use crate::errors::ShellError;
 use crate::stream::OutputStream;
 
-pub trait Shell {
+pub trait Shell: std::fmt::Debug {
     fn name(&self, source_map: &SourceMap) -> String;
     fn ls(&self, args: EvaluatedWholeStreamCommandArgs) -> Result<OutputStream, ShellError>;
     fn cd(&self, args: EvaluatedWholeStreamCommandArgs) -> Result<OutputStream, ShellError>;

src/shell/shell_manager.rs

@@ -7,7 +7,7 @@ use crate::stream::OutputStream;
 use std::error::Error;
 use std::sync::{Arc, Mutex};
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct ShellManager {
     crate shells: Arc<Mutex<Vec<Box<dyn Shell + Send>>>>,
 }

src/shell/value_shell.rs

@@ -5,7 +5,7 @@ use crate::shell::shell::Shell;
 use std::ffi::OsStr;
 use std::path::PathBuf;
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct ValueShell {
     crate path: String,
     crate value: Tagged<Value>,

src/stream.rs

@@ -75,6 +75,29 @@ impl OutputStream {
             values: input.map(ReturnSuccess::value).boxed(),
         }
     }
+
+    pub fn drain_vec(&mut self) -> impl Future<Output = Vec<ReturnValue>> {
+        let mut values: BoxStream<'static, ReturnValue> = VecDeque::new().boxed();
+        std::mem::swap(&mut values, &mut self.values);
+
+        values.collect()
+    }
+}
+
+impl std::ops::Try for OutputStream {
+    type Ok = OutputStream;
+    type Error = ShellError;
+    fn into_result(self) -> Result<Self::Ok, Self::Error> {
+        Ok(self)
+    }
+
+    fn from_error(v: Self::Error) -> Self {
+        OutputStream::one(Err(v))
+    }
+
+    fn from_ok(v: Self::Ok) -> Self {
+        v
+    }
 }
 
 impl Stream for OutputStream {