Merge pull request #319 from jonathandturner/implement-whole-stream-command-for-all-commands

Implement whole stream command for all commands

src/cli.rs

@@ -145,38 +145,37 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
         use crate::commands::*;
 
         context.add_commands(vec![
-            command("first", Box::new(first::first)),
-            command("pick", Box::new(pick::pick)),
-            command("from-array", Box::new(from_array::from_array)),
-            command("from-ini", Box::new(from_ini::from_ini)),
-            command("from-csv", Box::new(from_csv::from_csv)),
-            command("from-json", Box::new(from_json::from_json)),
-            command("from-toml", Box::new(from_toml::from_toml)),
-            command("from-xml", Box::new(from_xml::from_xml)),
-            command("ps", Box::new(ps::ps)),
-            command("ls", Box::new(ls::ls)),
-            command("cd", Box::new(cd::cd)),
-            command("size", Box::new(size::size)),
-            command("from-yaml", Box::new(from_yaml::from_yaml)),
-            command("nth", Box::new(nth::nth)),
-            command("n", Box::new(next::next)),
-            command("p", Box::new(prev::prev)),
-            command("debug", Box::new(debug::debug)),
-            command("lines", Box::new(lines::lines)),
-            command("pick", Box::new(pick::pick)),
-            command("shells", Box::new(shells::shells)),
-            command("split-column", Box::new(split_column::split_column)),
-            command("split-row", Box::new(split_row::split_row)),
-            command("lines", Box::new(lines::lines)),
-            command("reject", Box::new(reject::reject)),
-            command("trim", Box::new(trim::trim)),
-            command("to-array", Box::new(to_array::to_array)),
-            command("to-csv", Box::new(to_csv::to_csv)),
-            command("to-json", Box::new(to_json::to_json)),
-            command("to-toml", Box::new(to_toml::to_toml)),
-            command("to-yaml", Box::new(to_yaml::to_yaml)),
-            command("sort-by", Box::new(sort_by::sort_by)),
-            command("tags", Box::new(tags::tags)),
+            whole_stream_command(PS),
+            whole_stream_command(LS),
+            whole_stream_command(CD),
+            whole_stream_command(Size),
+            whole_stream_command(Nth),
+            whole_stream_command(Next),
+            whole_stream_command(Previous),
+            whole_stream_command(Debug),
+            whole_stream_command(Lines),
+            whole_stream_command(Shells),
+            whole_stream_command(SplitColumn),
+            whole_stream_command(SplitRow),
+            whole_stream_command(Lines),
+            whole_stream_command(Reject),
+            whole_stream_command(Trim),
+            whole_stream_command(ToArray),
+            whole_stream_command(ToCSV),
+            whole_stream_command(ToJSON),
+            whole_stream_command(ToTOML),
+            whole_stream_command(ToYAML),
+            whole_stream_command(SortBy),
+            whole_stream_command(Tags),
+            whole_stream_command(First),
+            whole_stream_command(FromArray),
+            whole_stream_command(FromCSV),
+            whole_stream_command(FromINI),
+            whole_stream_command(FromJSON),
+            whole_stream_command(FromTOML),
+            whole_stream_command(FromXML),
+            whole_stream_command(FromYAML),
+            whole_stream_command(Pick),
             whole_stream_command(Get),
             per_item_command(Remove),
             per_item_command(Open),

src/commands.rs

@@ -56,25 +56,54 @@ crate mod where_;
 crate mod which_;
 
 crate use autoview::Autoview;
-//crate use cd::Cd;
+crate use cd::CD;
 crate use clip::Clip;
 crate use command::{
-    command, per_item_command, whole_stream_command, Command, CommandArgs, PerItemCommand,
-    RawCommandArgs, UnevaluatedCallInfo, WholeStreamCommand,
+    per_item_command, whole_stream_command, Command, CommandArgs, PerItemCommand, RawCommandArgs,
+    UnevaluatedCallInfo, WholeStreamCommand,
 };
 crate use config::Config;
 crate use cp::Cpy;
 crate use date::Date;
+crate use debug::Debug;
 crate use enter::Enter;
 crate use exit::Exit;
+crate use first::First;
+crate use from_array::FromArray;
+crate use from_csv::FromCSV;
+crate use from_ini::FromINI;
+crate use from_json::FromJSON;
+crate use from_toml::FromTOML;
+crate use from_xml::FromXML;
+crate use from_yaml::FromYAML;
 crate use get::Get;
+crate use lines::Lines;
+crate use ls::LS;
 crate use mkdir::Mkdir;
 crate use mv::Move;
+crate use next::Next;
+crate use nth::Nth;
 crate use open::Open;
+crate use pick::Pick;
+crate use prev::Previous;
+crate use ps::PS;
+crate use reject::Reject;
 crate use rm::Remove;
 crate use save::Save;
+crate use shells::Shells;
+crate use size::Size;
 crate use skip_while::SkipWhile;
+crate use sort_by::SortBy;
+crate use split_column::SplitColumn;
+crate use split_row::SplitRow;
 crate use table::Table;
+crate use tags::Tags;
+crate use to_array::ToArray;
+crate use to_csv::ToCSV;
+crate use to_json::ToJSON;
+crate use to_toml::ToTOML;
+crate use to_yaml::ToYAML;
+crate use trim::Trim;
 crate use version::Version;
 crate use vtable::VTable;
 crate use where_::Where;

src/commands/cd.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::prelude::*;
 
-pub fn cd(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct CD;
+
+impl WholeStreamCommand for CD {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        cd(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "cd"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("cd").optional("directory", SyntaxType::Path)
+    }
+}
+
+fn cd(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let shell_manager = args.shell_manager.clone();
     let args = args.evaluate_once(registry)?;
     shell_manager.cd(args)

src/commands/command.rs

@@ -635,43 +635,6 @@ impl WholeStreamCommand for FnFilterCommand {
     }
 }
 
-pub struct FnRawCommand {
-    name: String,
-    func: Box<
-        dyn Fn(CommandArgs, &registry::CommandRegistry) -> Result<OutputStream, ShellError>
-            + Send
-            + Sync,
-    >,
-}
-
-impl WholeStreamCommand for FnRawCommand {
-    fn name(&self) -> &str {
-        &self.name
-    }
-
-    fn run(
-        &self,
-        args: CommandArgs,
-        registry: &registry::CommandRegistry,
-    ) -> Result<OutputStream, ShellError> {
-        (self.func)(args, registry)
-    }
-}
-
-pub fn command(
-    name: &str,
-    func: Box<
-        dyn Fn(CommandArgs, &registry::CommandRegistry) -> Result<OutputStream, ShellError>
-            + Send
-            + Sync,
-    >,
-) -> Arc<Command> {
-    Arc::new(Command::WholeStream(Arc::new(FnRawCommand {
-        name: name.to_string(),
-        func,
-    })))
-}
-
 pub fn whole_stream_command(command: impl WholeStreamCommand + 'static) -> Arc<Command> {
     Arc::new(Command::WholeStream(Arc::new(command)))
 }

src/commands/debug.rs

@@ -1,6 +1,27 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::prelude::*;
 
+pub struct Debug;
+
+impl WholeStreamCommand for Debug {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        debug(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "debug"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("debug")
+    }
+}
+
 pub fn debug(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let input = args.input;
 

src/commands/first.rs

@@ -1,18 +1,31 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::parser::CommandRegistry;
 use crate::prelude::*;
 
-pub fn first(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
+pub struct First;
 
-    if args.len() == 0 {
-        return Err(ShellError::labeled_error(
-            "First requires an amount",
-            "needs parameter",
-            args.name_span(),
-        ));
+impl WholeStreamCommand for First {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        first(args, registry)
     }
 
+    fn name(&self) -> &str {
+        "first"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("first").required("amount", SyntaxType::Literal)
+    }
+}
+
+fn first(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+    let args = args.evaluate_once(registry)?;
+
     let amount = args.expect_nth(0)?.as_i64();
 
     let amount = match amount {

src/commands/from_array.rs

@@ -1,10 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::Value;
 use crate::prelude::*;
 
-pub fn from_array(
-    args: CommandArgs,
-    _registry: &CommandRegistry,
-) -> Result<OutputStream, ShellError> {
+pub struct FromArray;
+
+impl WholeStreamCommand for FromArray {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        from_array(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "from-array"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("from-array")
+    }
+}
+
+fn from_array(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let stream = args
         .input
         .values

src/commands/from_csv.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
 use csv::ReaderBuilder;
 
+pub struct FromCSV;
+
+impl WholeStreamCommand for FromCSV {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        from_csv(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "from-csv"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("from-csv")
+    }
+}
+
 pub fn from_csv_string_to_value(
     s: String,
     tag: impl Into<Tag>,
@@ -45,7 +66,7 @@ pub fn from_csv_string_to_value(
     Ok(Tagged::from_item(Value::List(rows), tag))
 }
 
-pub fn from_csv(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn from_csv(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let span = args.name_span();
     let out = args.input;

src/commands/from_ini.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
 use std::collections::HashMap;
 
+pub struct FromINI;
+
+impl WholeStreamCommand for FromINI {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        from_ini(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "from-ini"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("from-ini")
+    }
+}
+
 fn convert_ini_second_to_nu_value(
     v: &HashMap<String, String>,
     tag: impl Into<Tag>,
@@ -37,7 +58,7 @@ pub fn from_ini_string_to_value(
     Ok(convert_ini_top_to_nu_value(&v, tag))
 }
 
-pub fn from_ini(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn from_ini(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let span = args.name_span();
     let out = args.input;

src/commands/from_json.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::base::OF64;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
 
+pub struct FromJSON;
+
+impl WholeStreamCommand for FromJSON {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        from_json(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "from-json"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("from-json")
+    }
+}
+
 fn convert_json_value_to_nu_value(v: &serde_hjson::Value, tag: impl Into<Tag>) -> Tagged<Value> {
     let tag = tag.into();
 
@@ -43,10 +64,7 @@ pub fn from_json_string_to_value(
     Ok(convert_json_value_to_nu_value(&v, tag))
 }
 
-pub fn from_json(
-    args: CommandArgs,
-    registry: &CommandRegistry,
-) -> Result<OutputStream, ShellError> {
+fn from_json(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let span = args.name_span();
     let out = args.input;

src/commands/from_toml.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::base::OF64;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
 
+pub struct FromTOML;
+
+impl WholeStreamCommand for FromTOML {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        from_toml(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "from-toml"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("from-toml")
+    }
+}
+
 pub fn convert_toml_value_to_nu_value(v: &toml::Value, tag: impl Into<Tag>) -> Tagged<Value> {
     let tag = tag.into();
 

src/commands/from_xml.rs

@@ -1,6 +1,27 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
 
+pub struct FromXML;
+
+impl WholeStreamCommand for FromXML {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        from_xml(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "from-xml"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("from-xml")
+    }
+}
+
 fn from_node_to_value<'a, 'd>(n: &roxmltree::Node<'a, 'd>, tag: impl Into<Tag>) -> Tagged<Value> {
     let tag = tag.into();
 
@@ -56,7 +77,7 @@ pub fn from_xml_string_to_value(
     Ok(from_document_to_value(&parsed, tag))
 }
 
-pub fn from_xml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn from_xml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let span = args.name_span();
     let out = args.input;

src/commands/from_yaml.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::base::OF64;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
 
+pub struct FromYAML;
+
+impl WholeStreamCommand for FromYAML {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        from_yaml(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "from-yaml"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("from-yaml")
+    }
+}
+
 fn convert_yaml_value_to_nu_value(v: &serde_yaml::Value, tag: impl Into<Tag>) -> Tagged<Value> {
     let tag = tag.into();
 
@@ -47,10 +68,7 @@ pub fn from_yaml_string_to_value(
     Ok(convert_yaml_value_to_nu_value(&v, tag))
 }
 
-pub fn from_yaml(
-    args: CommandArgs,
-    _registry: &CommandRegistry,
-) -> Result<OutputStream, ShellError> {
+fn from_yaml(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let span = args.name_span();
     let out = args.input;
 

src/commands/lines.rs

@@ -1,11 +1,32 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::{Primitive, Value};
 use crate::prelude::*;
 use log::trace;
 
+pub struct Lines;
+
+impl WholeStreamCommand for Lines {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        lines(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "lines"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("lines")
+    }
+}
+
 // TODO: "Amount remaining" wrapper
 
-pub fn lines(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn lines(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let span = args.name_span();
     let input = args.input;

src/commands/ls.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::prelude::*;
 
-pub fn ls(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct LS;
+
+impl WholeStreamCommand for LS {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        ls(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "ls"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("ls").optional("path", SyntaxType::Path)
+    }
+}
+
+fn ls(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let shell_manager = args.shell_manager.clone();
     let args = args.evaluate_once(registry)?;
     shell_manager.ls(args)

src/commands/next.rs

@@ -1,7 +1,28 @@
 use crate::commands::command::CommandAction;
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::prelude::*;
 
-pub fn next(_args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct Next;
+
+impl WholeStreamCommand for Next {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        next(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "n"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("n")
+    }
+}
+
+fn next(_args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     Ok(vec![Ok(ReturnSuccess::Action(CommandAction::NextShell))].into())
 }

src/commands/nth.rs

@@ -1,32 +1,38 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::parser::CommandRegistry;
 use crate::prelude::*;
 
-pub fn nth(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
+#[derive(Deserialize)]
+struct NthArgs {
+    amount: Tagged<i64>,
+}
 
-    if args.len() == 0 {
-        return Err(ShellError::labeled_error(
-            "Nth requires an amount",
-            "needs amount",
-            args.name_span(),
-        ));
+pub struct Nth;
+
+impl WholeStreamCommand for Nth {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, nth)?.run()
     }
 
-    let amount = args.expect_nth(0)?.as_i64();
+    fn name(&self) -> &str {
+        "nth"
+    }
 
-    let amount = match amount {
-        Ok(o) => o,
-        Err(_) => {
-            return Err(ShellError::labeled_error(
-                "Value is not a number",
-                "expected integer",
-                args.expect_nth(0)?.span(),
-            ))
-        }
-    };
+    fn signature(&self) -> Signature {
+        Signature::build("nth").required("amount", SyntaxType::Any)
+    }
+}
 
+fn nth(
+    NthArgs { amount }: NthArgs,
+    RunnableContext { input, .. }: RunnableContext,
+) -> Result<OutputStream, ShellError> {
     Ok(OutputStream::from_input(
-        args.input.values.skip(amount as u64).take(1),
+        input.values.skip(amount.item as u64).take(1),
     ))
 }

src/commands/pick.rs

@@ -1,30 +1,47 @@
+use crate::commands::WholeStreamCommand;
 use crate::context::CommandRegistry;
 use crate::errors::ShellError;
 use crate::object::base::select_fields;
 use crate::prelude::*;
 
-pub fn pick(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
-    let len = args.len();
-    let span = args.name_span();
-    let (input, args) = args.parts();
+#[derive(Deserialize)]
+struct PickArgs {
+    rest: Vec<Tagged<String>>,
+}
 
-    if len == 0 {
+pub struct Pick;
+
+impl WholeStreamCommand for Pick {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, pick)?.run()
+    }
+
+    fn name(&self) -> &str {
+        "pick"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("pick").rest()
+    }
+}
+
+fn pick(
+    PickArgs { rest: fields }: PickArgs,
+    RunnableContext { input, name, .. }: RunnableContext,
+) -> Result<OutputStream, ShellError> {
+    if fields.len() == 0 {
         return Err(ShellError::labeled_error(
             "Pick requires fields",
             "needs parameter",
-            span,
+            name,
         ));
     }
 
-    let fields: Result<Vec<String>, _> = args
-        .positional
-        .iter()
-        .flatten()
-        .map(|a| a.as_string())
-        .collect();
-
-    let fields = fields?;
+    let fields: Vec<_> = fields.iter().map(|f| f.item.clone()).collect();
 
     let objects = input
         .values

src/commands/prev.rs

@@ -2,6 +2,28 @@ use crate::commands::command::CommandAction;
 use crate::errors::ShellError;
 use crate::prelude::*;
 
-pub fn prev(_args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+use crate::commands::WholeStreamCommand;
+
+pub struct Previous;
+
+impl WholeStreamCommand for Previous {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        previous(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "p"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("p")
+    }
+}
+
+fn previous(_args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     Ok(vec![Ok(ReturnSuccess::Action(CommandAction::PreviousShell))].into())
 }

src/commands/ps.rs

@@ -1,9 +1,30 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::process::process_dict;
 use crate::prelude::*;
 use sysinfo::{RefreshKind, SystemExt};
 
-pub fn ps(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct PS;
+
+impl WholeStreamCommand for PS {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        ps(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "ps"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("ps")
+    }
+}
+
+fn ps(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let mut system = sysinfo::System::new_with_specifics(RefreshKind::new().with_processes());
     system.refresh_processes();
     let list = system.get_process_list();

src/commands/reject.rs

@@ -1,29 +1,46 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::base::reject_fields;
 use crate::prelude::*;
 
-pub fn reject(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
-    let len = args.len();
-    let span = args.name_span();
-    let (input, args) = args.parts();
+#[derive(Deserialize)]
+pub struct RejectArgs {
+    rest: Vec<Tagged<String>>,
+}
 
-    if len == 0 {
+pub struct Reject;
+
+impl WholeStreamCommand for Reject {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, reject)?.run()
+    }
+
+    fn name(&self) -> &str {
+        "reject"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("reject").rest()
+    }
+}
+
+fn reject(
+    RejectArgs { rest: fields }: RejectArgs,
+    RunnableContext { input, name, .. }: RunnableContext,
+) -> Result<OutputStream, ShellError> {
+    if fields.len() == 0 {
         return Err(ShellError::labeled_error(
             "Reject requires fields",
             "needs parameter",
-            span,
+            name,
         ));
     }
 
-    let fields: Result<Vec<String>, _> = args
-        .positional
-        .iter()
-        .flatten()
-        .map(|a| a.as_string())
-        .collect();
-
-    let fields = fields?;
+    let fields: Vec<_> = fields.iter().map(|f| f.item.clone()).collect();
 
     let stream = input
         .values

src/commands/shells.rs

@@ -1,8 +1,29 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::TaggedDictBuilder;
 use crate::prelude::*;
 
-pub fn shells(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct Shells;
+
+impl WholeStreamCommand for Shells {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        shells(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "shells"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("shells")
+    }
+}
+
+fn shells(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let mut shells_out = VecDeque::new();
     let span = args.call_info.name_span;
 

src/commands/size.rs

@@ -1,8 +1,29 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::{TaggedDictBuilder, Value};
 use crate::prelude::*;
 
-pub fn size(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct Size;
+
+impl WholeStreamCommand for Size {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        size(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "size"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("size")
+    }
+}
+
+fn size(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let input = args.input;
     let span = args.call_info.name_span;
     Ok(input

src/commands/sort_by.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::prelude::*;
 
-pub fn sort_by(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct SortBy;
+
+impl WholeStreamCommand for SortBy {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        sort_by(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "sort-by"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("sort-by")
+    }
+}
+
+fn sort_by(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let (input, args) = args.parts();
 

src/commands/split_column.rs

@@ -1,38 +1,55 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
 use log::trace;
 
-pub fn split_column(
-    args: CommandArgs,
-    registry: &CommandRegistry,
-) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
-    let span = args.name_span();
-    let (input, args) = args.parts();
+#[derive(Deserialize)]
+struct SplitColumnArgs {
+    separator: Tagged<String>,
+    rest: Vec<Tagged<String>>,
+}
 
-    let positional: Vec<_> = args.positional.iter().flatten().cloned().collect();
+pub struct SplitColumn;
 
-    if positional.len() == 0 {
-        return Err(ShellError::labeled_error(
-            "Split-column needs more information",
-            "needs parameter (eg split-column \",\")",
-            span,
-        ));
+impl WholeStreamCommand for SplitColumn {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, split_column)?.run()
     }
 
+    fn name(&self) -> &str {
+        "split-column"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("split-column")
+            .required("separator", SyntaxType::Any)
+            .rest()
+    }
+}
+
+fn split_column(
+    SplitColumnArgs { separator, rest }: SplitColumnArgs,
+    RunnableContext { input, name, .. }: RunnableContext,
+) -> Result<OutputStream, ShellError> {
     Ok(input
         .values
         .map(move |v| match v.item {
             Value::Primitive(Primitive::String(ref s)) => {
-                let splitter = positional[0].as_string().unwrap().replace("\\n", "\n");
+                let splitter = separator.replace("\\n", "\n");
                 trace!("splitting with {:?}", splitter);
-                let split_result: Vec<_> = s.split(&splitter).filter(|s| s.trim() != "").collect();
 
+                let split_result: Vec<_> = s.split(&splitter).filter(|s| s.trim() != "").collect();
                 trace!("split result = {:?}", split_result);
 
+                let positional: Vec<_> = rest.iter().map(|f| f.item.clone()).collect();
+
                 // If they didn't provide column names, make up our own
-                if (positional.len() - 1) == 0 {
+                if positional.len() == 0 {
                     let mut gen_columns = vec![];
                     for i in 0..split_result.len() {
                         gen_columns.push(format!("Column{}", i + 1));
@@ -44,19 +61,16 @@ pub fn split_column(
                     }
 
                     ReturnSuccess::value(dict.into_tagged_value())
-                } else if split_result.len() == (positional.len() - 1) {
+                } else if split_result.len() == positional.len() {
                     let mut dict = TaggedDictBuilder::new(v.tag());
-                    for (&k, v) in split_result.iter().zip(positional.iter().skip(1)) {
-                        dict.insert(
-                            v.as_string().unwrap(),
-                            Value::Primitive(Primitive::String(k.into())),
-                        );
+                    for (&k, v) in split_result.iter().zip(positional.iter()) {
+                        dict.insert(v, Value::Primitive(Primitive::String(k.into())));
                     }
                     ReturnSuccess::value(dict.into_tagged_value())
                 } else {
                     let mut dict = TaggedDictBuilder::new(v.tag());
-                    for k in positional.iter().skip(1) {
-                        dict.insert(k.as_string().unwrap().trim(), Primitive::String("".into()));
+                    for (&k, v) in split_result.iter().zip(positional.iter()) {
+                        dict.insert(v, Value::Primitive(Primitive::String(k.into())));
                     }
                     ReturnSuccess::value(dict.into_tagged_value())
                 }
@@ -64,7 +78,7 @@ pub fn split_column(
             _ => Err(ShellError::labeled_error_with_secondary(
                 "Expected a string from pipeline",
                 "requires string input",
-                span,
+                name,
                 "value originates from here",
                 v.span(),
             )),

src/commands/split_row.rs

@@ -1,32 +1,43 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::{Primitive, Value};
 use crate::prelude::*;
 use log::trace;
 
-pub fn split_row(
-    args: CommandArgs,
-    registry: &CommandRegistry,
-) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
-    let span = args.name_span();
-    let len = args.len();
-    let (input, args) = args.parts();
+#[derive(Deserialize)]
+struct SplitRowArgs {
+    separator: Tagged<String>,
+}
 
-    let positional: Vec<Tagged<Value>> = args.positional.iter().flatten().cloned().collect();
+pub struct SplitRow;
 
-    if len == 0 {
-        return Err(ShellError::labeled_error(
-            "Split-row needs more information",
-            "needs parameter (eg split-row \"\\n\")",
-            span,
-        ));
+impl WholeStreamCommand for SplitRow {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, split_row)?.run()
     }
 
+    fn name(&self) -> &str {
+        "split-row"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("split-row").required("separator", SyntaxType::Any)
+    }
+}
+
+fn split_row(
+    SplitRowArgs { separator }: SplitRowArgs,
+    RunnableContext { input, name, .. }: RunnableContext,
+) -> Result<OutputStream, ShellError> {
     let stream = input
         .values
         .map(move |v| match v.item {
             Value::Primitive(Primitive::String(ref s)) => {
-                let splitter = positional[0].as_string().unwrap().replace("\\n", "\n");
+                let splitter = separator.item.replace("\\n", "\n");
                 trace!("splitting with {:?}", splitter);
                 let split_result: Vec<_> = s.split(&splitter).filter(|s| s.trim() != "").collect();
 
@@ -45,7 +56,7 @@ pub fn split_row(
                 result.push_back(Err(ShellError::labeled_error_with_secondary(
                     "Expected a string from pipeline",
                     "requires string input",
-                    span,
+                    name,
                     "value originates from here",
                     v.span(),
                 )));

src/commands/tags.rs

@@ -1,8 +1,29 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::{TaggedDictBuilder, Value};
 use crate::prelude::*;
 
-pub fn tags(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct Tags;
+
+impl WholeStreamCommand for Tags {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        tags(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "tags"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("tags")
+    }
+}
+
+fn tags(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let source_map = args.call_info.source_map.clone();
     Ok(args
         .input

src/commands/to_array.rs

@@ -1,14 +1,34 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::Value;
 use crate::prelude::*;
 
-pub fn to_array(
-    args: CommandArgs,
-    _registry: &CommandRegistry,
-) -> Result<OutputStream, ShellError> {
+pub struct ToArray;
+
+impl WholeStreamCommand for ToArray {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        to_array(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "to-array"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("to-array")
+    }
+}
+
+fn to_array(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+    let args = args.evaluate_once(registry)?;
+    let span = args.call_info.name_span;
     let out = args.input.values.collect();
 
     Ok(out
-        .map(|vec: Vec<_>| stream![Value::List(vec).tagged_unknown()]) // TODO: args.input should have a span
+        .map(move |vec: Vec<_>| stream![Value::List(vec).simple_spanned(span)])
         .flatten_stream()
         .from_input_stream())
 }

src/commands/to_csv.rs

@@ -1,7 +1,28 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::{Primitive, Value};
 use crate::prelude::*;
 use csv::WriterBuilder;
 
+pub struct ToCSV;
+
+impl WholeStreamCommand for ToCSV {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        to_csv(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "to-csv"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("to-csv")
+    }
+}
+
 pub fn value_to_csv_value(v: &Value) -> Value {
     match v {
         Value::Primitive(Primitive::String(s)) => Value::Primitive(Primitive::String(s.clone())),
@@ -36,7 +57,7 @@ pub fn to_string(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
     }
 }
 
-pub fn to_csv(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn to_csv(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
     let out = args.input;

src/commands/to_json.rs

@@ -1,6 +1,27 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::{Primitive, Value};
 use crate::prelude::*;
 
+pub struct ToJSON;
+
+impl WholeStreamCommand for ToJSON {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        to_json(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "to-json"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("to-json")
+    }
+}
+
 pub fn value_to_json_value(v: &Value) -> serde_json::Value {
     match v {
         Value::Primitive(Primitive::Boolean(b)) => serde_json::Value::Bool(*b),
@@ -41,7 +62,7 @@ pub fn value_to_json_value(v: &Value) -> serde_json::Value {
     }
 }
 
-pub fn to_json(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn to_json(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
     let out = args.input;

src/commands/to_toml.rs

@@ -1,6 +1,27 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::{Primitive, Value};
 use crate::prelude::*;
 
+pub struct ToTOML;
+
+impl WholeStreamCommand for ToTOML {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        to_toml(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "to-toml"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("to-toml")
+    }
+}
+
 pub fn value_to_toml_value(v: &Value) -> toml::Value {
     match v {
         Value::Primitive(Primitive::Boolean(b)) => toml::Value::Boolean(*b),
@@ -33,7 +54,7 @@ pub fn value_to_toml_value(v: &Value) -> toml::Value {
     }
 }
 
-pub fn to_toml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn to_toml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
     let out = args.input;

src/commands/to_yaml.rs

@@ -1,6 +1,27 @@
+use crate::commands::WholeStreamCommand;
 use crate::object::{Primitive, Value};
 use crate::prelude::*;
 
+pub struct ToYAML;
+
+impl WholeStreamCommand for ToYAML {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        to_yaml(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "to-yaml"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("to-yaml")
+    }
+}
+
 pub fn value_to_yaml_value(v: &Value) -> serde_yaml::Value {
     match v {
         Value::Primitive(Primitive::Boolean(b)) => serde_yaml::Value::Bool(*b),
@@ -39,7 +60,7 @@ pub fn value_to_yaml_value(v: &Value) -> serde_yaml::Value {
     }
 }
 
-pub fn to_yaml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+fn to_yaml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
     let out = args.input;

src/commands/trim.rs

@@ -1,8 +1,29 @@
+use crate::commands::WholeStreamCommand;
 use crate::errors::ShellError;
 use crate::object::Value;
 use crate::prelude::*;
 
-pub fn trim(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+pub struct Trim;
+
+impl WholeStreamCommand for Trim {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        trim(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "trim"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("trim")
+    }
+}
+
+fn trim(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let input = args.input;
 
     Ok(input

src/errors.rs

@@ -186,13 +186,14 @@ impl ShellError {
                 ArgumentError::MissingMandatoryPositional(name) => Diagnostic::new(
                     Severity::Error,
                     format!(
-                        "{} requires {}",
+                        "{} requires {} parameter",
                         Color::Cyan.paint(command),
-                        Color::Green.bold().paint(name)
+                        Color::Green.bold().paint(name.clone())
                     ),
                 )
-                .with_label(Label::new_primary(span)),
-
+                .with_label(
+                    Label::new_primary(span).with_message(format!("requires {} parameter", name)),
+                ),
                 ArgumentError::MissingValueForName(name) => Diagnostic::new(
                     Severity::Error,
                     format!(