WIP - types check

src/cli.rs

@@ -175,9 +175,9 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
             static_command(SkipWhile),
             static_command(Clip),
             static_command(Autoview),
-            // command("save", Box::new(save::save)),
-            // command("table", Box::new(table::table)),
-            // command("vtable", Box::new(vtable::vtable)),
+            static_command(Save),
+            static_command(Table),
+            static_command(VTable),
         ]);
     }
     let _ = load_plugins(&mut context);
@@ -337,7 +337,7 @@ async fn process_line(readline: Result<String, ReadlineError>, ctx: &mut Context
                     .push(ClassifiedCommand::Internal(InternalCommand {
                         command: static_command(autoview::Autoview),
                         name_span: None,
-                        source_map: ctx.source_map,
+                        source_map: ctx.source_map.clone(),
                         args: hir::Call::new(
                             Box::new(hir::Expression::synthetic_string("autoview")),
                             None,

src/commands.rs

@@ -47,11 +47,14 @@ crate use cd::Cd;
 crate use clip::Clip;
 crate use command::{
     command, static_command, CallInfo, Command, CommandArgs, EvaluatedStaticCommandArgs,
-    StaticCommand, UnevaluatedCallInfo,
+    RawCommandArgs, StaticCommand, UnevaluatedCallInfo,
 };
 crate use config::Config;
 crate use get::Get;
 crate use open::Open;
 crate use rm::Remove;
+crate use save::Save;
 crate use skip_while::SkipWhile;
+crate use table::Table;
+crate use vtable::VTable;
 crate use where_::Where;

src/commands/autoview.rs

@@ -1,4 +1,4 @@
-use crate::commands::StaticCommand;
+use crate::commands::{RawCommandArgs, StaticCommand};
 use crate::context::{SourceMap, SpanSource};
 use crate::errors::ShellError;
 use crate::format::GenericView;
@@ -7,6 +7,9 @@ use std::path::Path;
 
 pub struct Autoview;
 
+#[derive(Deserialize)]
+pub struct AutoviewArgs {}
+
 impl StaticCommand for Autoview {
     fn name(&self) -> &str {
         "autoview"
@@ -17,37 +20,49 @@ impl StaticCommand for Autoview {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        args.process(registry, autoview)?.run()
+        args.process_raw(registry, autoview)?.run()
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("autoview").sink()
+        Signature::build("autoview")
     }
 }
 
-pub fn autoview(args: (), context: RunnableContext) -> Result<OutputStream, ShellError> {
-    if args.input.len() > 0 {
-        if let Spanned {
-            item: Value::Binary(_),
-            ..
-        } = args.input[0]
-        {
-            args.ctx.get_sink("binaryview").run(args)?;
-        } else if is_single_text_value(&args.input) {
-            view_text_value(&args.input[0], &args.call_info.source_map);
-        } else if equal_shapes(&args.input) {
-            args.ctx.get_sink("table").run(args)?;
-        } else {
-            let mut host = args.ctx.host.lock().unwrap();
-            for i in args.input.iter() {
-                let view = GenericView::new(&i);
-                handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host));
-                host.stdout("");
+pub fn autoview(
+    AutoviewArgs {}: AutoviewArgs,
+    mut context: RunnableContext,
+    raw: RawCommandArgs,
+) -> Result<OutputStream, ShellError> {
+    let stream = async_stream_block! {
+        let input = context.input.drain_vec().await;
+
+        if input.len() > 0 {
+            if let Spanned {
+                item: Value::Binary(_),
+                ..
+            } = input[0]
+            {
+                let binary = context.expect_command("binaryview");
+                binary.run(raw.with_input(input), &context.commands).await;
+            } else if is_single_text_value(&input) {
+                view_text_value(&input[0], &raw.call_info.source_map);
+            } else if equal_shapes(&input) {
+                let table = context.expect_command("table");
+                table.run(raw.with_input(input), &context.commands).await;
+            } else {
+                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("");
+                // }
             }
         }
-    }
+    };
 
-    Ok(OutputStream::empty())
+    Ok(OutputStream::new(stream))
 }
 
 fn equal_shapes(input: &Vec<Spanned<Value>>) -> bool {

src/commands/clip.rs

@@ -24,7 +24,7 @@ impl StaticCommand for Clip {
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("clip").sink()
+        Signature::build("clip")
     }
 }
 

src/commands/command.rs

@@ -59,6 +59,25 @@ pub struct CommandArgs {
     pub input: InputStream,
 }
 
+#[derive(Getters)]
+#[get = "crate"]
+pub struct RawCommandArgs {
+    pub host: Arc<Mutex<dyn Host>>,
+    pub env: Arc<Mutex<Environment>>,
+    pub call_info: UnevaluatedCallInfo,
+}
+
+impl RawCommandArgs {
+    pub fn with_input(self, input: Vec<Spanned<Value>>) -> CommandArgs {
+        CommandArgs {
+            host: self.host,
+            env: self.env,
+            call_info: self.call_info,
+            input: input.into(),
+        }
+    }
+}
+
 impl ToDebug for CommandArgs {
     fn fmt_debug(&self, f: &mut fmt::Formatter, source: &str) -> fmt::Result {
         self.call_info.fmt_debug(f, source)
@@ -88,6 +107,7 @@ impl CommandArgs {
         callback: fn(T, RunnableContext) -> Result<OutputStream, ShellError>,
     ) -> Result<RunnableArgs<T>, ShellError> {
         let env = self.env.clone();
+        let host = self.host.clone();
         let args = self.evaluate_once(registry)?;
         let (input, args) = args.split();
         let name_span = args.call_info.name_span;
@@ -97,12 +117,46 @@ impl CommandArgs {
             args: T::deserialize(&mut deserializer)?,
             context: RunnableContext {
                 input: input,
+                commands: registry.clone(),
                 env,
                 name: name_span,
+                host,
             },
             callback,
         })
     }
+
+    pub fn process_raw<'de, T: Deserialize<'de>>(
+        self,
+        registry: &CommandRegistry,
+        callback: fn(T, RunnableContext, RawCommandArgs) -> Result<OutputStream, ShellError>,
+    ) -> Result<RunnableRawArgs<T>, ShellError> {
+        let raw_args = RawCommandArgs {
+            host: self.host.clone(),
+            env: self.env.clone(),
+            call_info: self.call_info.clone(),
+        };
+
+        let env = self.env.clone();
+        let host = self.host.clone();
+        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);
+
+        Ok(RunnableRawArgs {
+            args: T::deserialize(&mut deserializer)?,
+            context: RunnableContext {
+                input: input,
+                commands: registry.clone(),
+                env,
+                name: name_span,
+                host,
+            },
+            raw_args,
+            callback,
+        })
+    }
 }
 
 pub struct SinkContext {
@@ -120,6 +174,8 @@ pub struct SinkArgs<T> {
 pub struct RunnableContext {
     pub input: InputStream,
     pub env: Arc<Mutex<Environment>>,
+    pub host: Arc<Mutex<dyn Host>>,
+    pub commands: CommandRegistry,
     pub name: Option<Span>,
 }
 
@@ -130,6 +186,12 @@ impl RunnableContext {
 
         env.path.clone()
     }
+
+    pub fn expect_command(&self, name: &str) -> Arc<Command> {
+        self.commands
+            .get_command(name)
+            .expect(&format!("Expected command {}", name))
+    }
 }
 
 pub struct RunnableArgs<T> {
@@ -144,6 +206,19 @@ impl<T> RunnableArgs<T> {
     }
 }
 
+pub struct RunnableRawArgs<T> {
+    args: T,
+    raw_args: RawCommandArgs,
+    context: RunnableContext,
+    callback: fn(T, RunnableContext, RawCommandArgs) -> Result<OutputStream, ShellError>,
+}
+
+impl<T> RunnableRawArgs<T> {
+    pub fn run(self) -> Result<OutputStream, ShellError> {
+        (self.callback)(self.args, self.context, self.raw_args)
+    }
+}
+
 pub struct EvaluatedStaticCommandArgs {
     pub args: EvaluatedCommandArgs,
     pub input: InputStream,

src/commands/config.rs

@@ -30,7 +30,6 @@ impl StaticCommand for Config {
             .named("remove", SyntaxType::Any)
             .switch("clear")
             .switch("path")
-            .sink()
     }
 
     fn run(

src/commands/open.rs

@@ -28,7 +28,6 @@ impl StaticCommand for Open {
         Signature::build(self.name())
             .required("path", SyntaxType::Block)
             .switch("raw")
-            .sink()
     }
 
     fn run(

src/commands/rm.rs

@@ -23,7 +23,6 @@ impl StaticCommand for Remove {
         Signature::build("rm")
             .required("path", SyntaxType::Path)
             .switch("recursive")
-            .sink()
     }
 
     fn run(

src/commands/save.rs

@@ -7,13 +7,12 @@ use crate::errors::ShellError;
 use crate::object::{Primitive, Value};
 use crate::parser::Spanned;
 use crate::prelude::*;
-use futures_async_stream::async_stream_block;
 use std::path::{Path, PathBuf};
 
 pub struct Save;
 
 #[derive(Deserialize)]
-struct SaveArgs {
+pub struct SaveArgs {
     path: Spanned<PathBuf>,
     raw: bool,
 }
@@ -27,7 +26,6 @@ impl StaticCommand for Save {
         Signature::build("save")
             .required("path", SyntaxType::Path)
             .switch("raw")
-            .sink()
     }
 
     fn run(
@@ -55,7 +53,7 @@ pub fn save(
         let contents = match full_path.extension() {
             Some(x) if x == "csv" && !save_raw => {
                 if input.len() != 1 {
-                    return Err(ShellError::string(
+                    yield Err(ShellError::string(
                         "saving to csv requires a single object (or use --raw)",
                     ));
                 }
@@ -63,7 +61,7 @@ pub fn save(
             }
             Some(x) if x == "toml" && !save_raw => {
                 if input.len() != 1 {
-                    return Err(ShellError::string(
+                    yield Err(ShellError::string(
                         "saving to toml requires a single object (or use --raw)",
                     ));
                 }
@@ -71,7 +69,7 @@ pub fn save(
             }
             Some(x) if x == "json" && !save_raw => {
                 if input.len() != 1 {
-                    return Err(ShellError::string(
+                    yield Err(ShellError::string(
                         "saving to json requires a single object (or use --raw)",
                     ));
                 }
@@ -79,7 +77,7 @@ pub fn save(
             }
             Some(x) if x == "yml" && !save_raw => {
                 if input.len() != 1 {
-                    return Err(ShellError::string(
+                    yield Err(ShellError::string(
                         "saving to yml requires a single object (or use --raw)",
                     ));
                 }
@@ -87,7 +85,7 @@ pub fn save(
             }
             Some(x) if x == "yaml" && !save_raw => {
                 if input.len() != 1 {
-                    return Err(ShellError::string(
+                    yield Err(ShellError::string(
                         "saving to yaml requires a single object (or use --raw)",
                     ));
                 }
@@ -113,7 +111,5 @@ pub fn save(
         let _ = std::fs::write(full_path, contents);
     };
 
-    let stream: BoxStream<'static, ReturnValue> = stream.boxed();
-
-    Ok(OutputStream::from(stream))
+    Ok(OutputStream::new(stream))
 }

src/commands/table.rs

@@ -1,15 +1,42 @@
+use crate::commands::StaticCommand;
 use crate::errors::ShellError;
 use crate::format::TableView;
 use crate::prelude::*;
+use futures_async_stream::async_stream_block;
 
-pub fn table(args: CommandArgs, context: RunnableContext) -> Result<(), ShellError> {
-    if args.input.len() > 0 {
-        let mut host = args.ctx.host.lock().unwrap();
-        let view = TableView::from_list(&args.input);
-        if let Some(view) = view {
-            handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host));
-        }
+pub struct Table;
+
+#[derive(Deserialize)]
+pub struct TableArgs {}
+
+impl StaticCommand for Table {
+    fn name(&self) -> &str {
+        "table"
+    }
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, table)?.run()
+    }
+    fn signature(&self) -> Signature {
+        Signature::build("table")
     }
-
-    Ok(())
+}
+
+pub fn table(_args: TableArgs, context: RunnableContext) -> Result<OutputStream, ShellError> {
+    let stream = async_stream_block! {
+        let input: Vec<Spanned<Value>> = context.input.into_vec().await;
+        if input.len() > 0 {
+            let mut host = context.host.lock().unwrap();
+            let view = TableView::from_list(&input);
+            println!("{:#?}", view);
+            if let Some(view) = view {
+                handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host));
+            }
+        }
+    };
+
+    Ok(OutputStream::new(stream))
 }

src/commands/vtable.rs

@@ -1,15 +1,41 @@
+use crate::commands::StaticCommand;
 use crate::errors::ShellError;
 use crate::format::VTableView;
 use crate::prelude::*;
 
-pub fn vtable(args: CommandArgs, context: RunnableContext) -> Result<(), ShellError> {
-    if args.input.len() > 0 {
-        let mut host = args.ctx.host.lock().unwrap();
-        let view = VTableView::from_list(&args.input);
-        if let Some(view) = view {
-            handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host));
-        }
-    }
+pub struct VTable;
 
-    Ok(())
+#[derive(Deserialize)]
+pub struct VTableArgs {}
+
+impl StaticCommand for VTable {
+    fn name(&self) -> &str {
+        "vtable"
+    }
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, vtable)?.run()
+    }
+    fn signature(&self) -> Signature {
+        Signature::build("vtable")
+    }
+}
+
+pub fn vtable(args: VTableArgs, context: RunnableContext) -> Result<OutputStream, ShellError> {
+    let stream = async_stream_block! {
+        let input = context.input.into_vec().await;
+
+        if input.len() > 0 {
+            let mut host = context.host.lock().unwrap();
+            let view = VTableView::from_list(&input);
+            if let Some(view) = view {
+                handle_unexpected(&mut *host, |host| crate::format::print_view(&view, host));
+            }
+        }
+    };
+
+    Ok(OutputStream::new(stream))
 }

src/commands/where_.rs

@@ -21,9 +21,7 @@ impl StaticCommand for Where {
     }
 
     fn signature(&self) -> registry::Signature {
-        Signature::build("where")
-            .required("condition", SyntaxType::Block)
-            .sink()
+        Signature::build("where").required("condition", SyntaxType::Block)
     }
 
     fn run(

src/context.rs

@@ -47,7 +47,7 @@ impl CommandRegistry {
         }
     }
 
-    fn get_command(&self, name: &str) -> Option<Arc<Command>> {
+    crate fn get_command(&self, name: &str) -> Option<Arc<Command>> {
         let registry = self.registry.lock().unwrap();
 
         registry.get(name).map(|c| c.clone())

src/format/table.rs

@@ -7,7 +7,7 @@ use prettytable::format::{FormatBuilder, LinePosition, LineSeparator};
 
 use prettytable::{color, Attr, Cell, Row, Table};
 
-#[derive(new)]
+#[derive(Debug, new)]
 pub struct TableView {
     headers: Vec<String>,
     entries: Vec<Vec<String>>,

src/parser/registry.rs

@@ -125,11 +125,6 @@ impl Signature {
         self
     }
 
-    pub fn sink(mut self) -> Signature {
-        self.is_sink = true;
-        self
-    }
-
     pub fn filter(mut self) -> Signature {
         self.is_filter = true;
         self

src/plugins/add.rs

@@ -1,7 +1,7 @@
 use indexmap::IndexMap;
 use nu::{
-    serve_plugin, CallInfo, Signature, Plugin, PositionalType, Primitive, ReturnSuccess,
-    ReturnValue, ShellError, Spanned, Value,
+    serve_plugin, CallInfo, Plugin, PositionalType, Primitive, ReturnSuccess, ReturnValue,
+    ShellError, Signature, Spanned, Value,
 };
 
 struct Add {
@@ -48,7 +48,6 @@ impl Plugin for Add {
                 PositionalType::mandatory_any("Value"),
             ],
             is_filter: true,
-            is_sink: false,
             named: IndexMap::new(),
             rest_positional: true,
         })

src/plugins/binaryview.rs

@@ -2,8 +2,7 @@
 use crossterm::{cursor, terminal, Attribute, RawScreen};
 use indexmap::IndexMap;
 use nu::{
-    serve_plugin, CallInfo, Signature, NamedType, Plugin, ShellError, SpanSource, Spanned,
-    Value,
+    serve_plugin, CallInfo, NamedType, Plugin, ShellError, Signature, SpanSource, Spanned, Value,
 };
 use pretty_hex::*;
 
@@ -23,7 +22,6 @@ impl Plugin for BinaryView {
             name: "binaryview".to_string(),
             positional: vec![],
             is_filter: false,
-            is_sink: true,
             named,
             rest_positional: false,
         })

src/plugins/edit.rs

@@ -1,7 +1,7 @@
 use indexmap::IndexMap;
 use nu::{
-    serve_plugin, CallInfo, Signature, Plugin, PositionalType, Primitive, ReturnSuccess,
-    ReturnValue, ShellError, Spanned, Value,
+    serve_plugin, CallInfo, Plugin, PositionalType, Primitive, ReturnSuccess, ReturnValue,
+    ShellError, Signature, Spanned, Value,
 };
 
 struct Edit {
@@ -48,7 +48,6 @@ impl Plugin for Edit {
                 PositionalType::mandatory_any("Value"),
             ],
             is_filter: true,
-            is_sink: false,
             named: IndexMap::new(),
             rest_positional: true,
         })

src/plugins/inc.rs

@@ -1,7 +1,7 @@
 use indexmap::IndexMap;
 use nu::{
-    serve_plugin, CallInfo, Signature, NamedType, Plugin, PositionalType, Primitive,
-    ReturnSuccess, ReturnValue, ShellError, Spanned, SpannedItem, Value,
+    serve_plugin, CallInfo, NamedType, Plugin, PositionalType, Primitive, ReturnSuccess,
+    ReturnValue, ShellError, Signature, Spanned, SpannedItem, Value,
 };
 
 struct Inc {
@@ -94,7 +94,6 @@ impl Plugin for Inc {
             name: "inc".to_string(),
             positional: vec![PositionalType::optional_any("Field")],
             is_filter: true,
-            is_sink: false,
             named,
             rest_positional: true,
         })

src/plugins/skip.rs

@@ -1,7 +1,7 @@
 use indexmap::IndexMap;
 use nu::{
-    serve_plugin, CallInfo, Signature, Plugin, Primitive, ReturnSuccess, ReturnValue,
-    ShellError, Spanned, Value,
+    serve_plugin, CallInfo, Plugin, Primitive, ReturnSuccess, ReturnValue, ShellError, Signature,
+    Spanned, Value,
 };
 
 struct NewSkip {
@@ -19,7 +19,6 @@ impl Plugin for NewSkip {
             name: "skip".to_string(),
             positional: vec![],
             is_filter: true,
-            is_sink: false,
             named: IndexMap::new(),
             rest_positional: true,
         })

src/plugins/tree.rs

@@ -1,6 +1,6 @@
 use derive_new::new;
 use indexmap::IndexMap;
-use nu::{serve_plugin, CallInfo, Signature, Plugin, ShellError, Spanned, Value};
+use nu::{serve_plugin, CallInfo, Plugin, ShellError, Signature, Spanned, Value};
 use ptree::item::StringItem;
 use ptree::output::print_tree_with;
 use ptree::print_config::PrintConfig;
@@ -85,7 +85,6 @@ impl Plugin for TreeViewer {
             name: "tree".to_string(),
             positional: vec![],
             is_filter: false,
-            is_sink: true,
             named: IndexMap::new(),
             rest_positional: true,
         })

src/prelude.rs

@@ -50,6 +50,7 @@ crate use crate::traits::{HasSpan, ToDebug};
 crate use crate::Text;
 crate use futures::stream::BoxStream;
 crate use futures::{FutureExt, Stream, StreamExt};
+crate use futures_async_stream::async_stream_block;
 crate use serde::{Deserialize, Serialize};
 crate use std::collections::VecDeque;
 crate use std::future::Future;

src/stream.rs

@@ -9,6 +9,13 @@ impl InputStream {
         self.values.collect()
     }
 
+    pub fn drain_vec(&mut self) -> impl Future<Output = Vec<Spanned<Value>>> {
+        let mut values: BoxStream<'static, Spanned<Value>> = VecDeque::new().boxed();
+        std::mem::swap(&mut values, &mut self.values);
+
+        values.collect()
+    }
+
     pub fn from_stream(input: impl Stream<Item = Spanned<Value>> + Send + 'static) -> InputStream {
         InputStream {
             values: input.boxed(),
@@ -46,13 +53,19 @@ pub struct OutputStream {
 }
 
 impl OutputStream {
+    pub fn new(values: impl Stream<Item = ReturnValue> + Send + 'static) -> OutputStream {
+        OutputStream {
+            values: values.boxed(),
+        }
+    }
+
     pub fn empty() -> OutputStream {
         let v: VecDeque<ReturnValue> = VecDeque::new();
         v.into()
     }
 
     pub fn one(item: impl Into<ReturnValue>) -> OutputStream {
-        let v: VecDeque<ReturnValue> = VecDeque::new();
+        let mut v: VecDeque<ReturnValue> = VecDeque::new();
         v.push_back(item.into());
         v.into()
     }
@@ -64,6 +77,17 @@ impl OutputStream {
     }
 }
 
+impl Stream for OutputStream {
+    type Item = ReturnValue;
+
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> core::task::Poll<Option<Self::Item>> {
+        Stream::poll_next(std::pin::Pin::new(&mut self.values), cx)
+    }
+}
+
 impl From<InputStream> for OutputStream {
     fn from(input: InputStream) -> OutputStream {
         OutputStream {