Merge branch 'master' into touchups

src/cli.rs

@@ -102,19 +102,16 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
         cc.store(true, Ordering::SeqCst);
     })
     .expect("Error setting Ctrl-C handler");
-
+    let mut ctrlcbreak = false;
     loop {
         if ctrl_c.load(Ordering::SeqCst) {
             ctrl_c.store(false, Ordering::SeqCst);
-            if let ShellError::String(s) = ShellError::string("CTRL-C") {
-                context.host.lock().unwrap().stdout(&format!("{:?}", s));
-            }
             continue;
         }
 
         let (obj, cwd) = {
             let env = context.env.lock().unwrap();
-            let last = env.last().unwrap();
+            let last = env.back().unwrap();
             (last.obj().clone(), last.path().display().to_string())
         };
         let readline = match obj {
@@ -134,6 +131,20 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
                 rl.add_history_entry(line.clone());
             }
 
+            LineResult::CtrlC => {
+                if ctrlcbreak {
+                    std::process::exit(0);
+                } else {
+                    context
+                        .host
+                        .lock()
+                        .unwrap()
+                        .stdout("CTRL-C pressed (again to quit)");
+                    ctrlcbreak = true;
+                    continue;
+                }
+            }
+
             LineResult::Error(mut line, err) => match err {
                 ShellError::Diagnostic(diag) => {
                     let host = context.host.lock().unwrap();
@@ -177,6 +188,7 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
                     .stdout(&format!("A surprising fatal error occurred.\n{:?}", err));
             }
         }
+        ctrlcbreak = false;
     }
     rl.save_history("history.txt").unwrap();
 
@@ -186,6 +198,7 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
 enum LineResult {
     Success(String),
     Error(String, ShellError),
+    CtrlC,
     Break,
 
     #[allow(unused)]
@@ -201,6 +214,7 @@ impl std::ops::Try for LineResult {
             LineResult::Success(s) => Ok(Some(s)),
             LineResult::Error(_, s) => Err(s),
             LineResult::Break => Ok(None),
+            LineResult::CtrlC => Ok(None),
             LineResult::FatalError(err) => Err(err),
         }
     }
@@ -259,27 +273,26 @@ async fn process_line(readline: Result<String, ReadlineError>, ctx: &mut Context
                     (None, _) => break,
 
                     (Some(ClassifiedCommand::Expr(_)), _) => {
-                        return LineResult::Error(line.clone(), ShellError::unimplemented(
+                        return LineResult::Error(
-                            "Expression-only commands",
+                            line.clone(),
-                        ))
+                            ShellError::unimplemented("Expression-only commands"),
+                        )
                     }
 
                     (_, Some(ClassifiedCommand::Expr(_))) => {
-                        return LineResult::Error(line.clone(), ShellError::unimplemented(
+                        return LineResult::Error(
-                            "Expression-only commands",
+                            line.clone(),
-                        ))
+                            ShellError::unimplemented("Expression-only commands"),
+                        )
                     }
 
-                    (Some(ClassifiedCommand::Sink(_)), Some(_)) => {
+                    (Some(ClassifiedCommand::Sink(SinkCommand { name_span, .. })), Some(_)) => {
-                        return LineResult::Error(line.clone(), ShellError::string("Commands like table, save, and autoview must come last in the pipeline"))
+                        return LineResult::Error(line.clone(), ShellError::maybe_labeled_error("Commands like table, save, and autoview must come last in the pipeline", "must come last", name_span));
                     }
 
                     (Some(ClassifiedCommand::Sink(left)), None) => {
                         let input_vec: Vec<Value> = input.objects.collect().await;
-                        left.run(
+                        left.run(ctx, input_vec)?;
-                            ctx,
-                            input_vec,
-                        )?;
                         break;
                     }
 
@@ -291,13 +304,12 @@ async fn process_line(readline: Result<String, ReadlineError>, ctx: &mut Context
                         Err(err) => return LineResult::Error(line.clone(), err),
                     },
 
-                    (
+                    (Some(ClassifiedCommand::Internal(left)), Some(_)) => {
-                        Some(ClassifiedCommand::Internal(left)),
+                        match left.run(ctx, input).await {
-                        Some(_),
+                            Ok(val) => ClassifiedInputStream::from_input_stream(val),
-                    ) => match left.run(ctx, input).await {
+                            Err(err) => return LineResult::Error(line.clone(), err),
-                        Ok(val) => ClassifiedInputStream::from_input_stream(val),
+                        }
-                        Err(err) => return LineResult::Error(line.clone(), err),
+                    }
-                    },
 
                     (Some(ClassifiedCommand::Internal(left)), None) => {
                         match left.run(ctx, input).await {
@@ -314,13 +326,12 @@ async fn process_line(readline: Result<String, ReadlineError>, ctx: &mut Context
                         Err(err) => return LineResult::Error(line.clone(), err),
                     },
 
-                    (
+                    (Some(ClassifiedCommand::External(left)), Some(_)) => {
-                        Some(ClassifiedCommand::External(left)),
+                        match left.run(ctx, input, StreamNext::Internal).await {
-                        Some(_),
+                            Ok(val) => val,
-                    ) => match left.run(ctx, input, StreamNext::Internal).await {
+                            Err(err) => return LineResult::Error(line.clone(), err),
-                        Ok(val) => val,
+                        }
-                        Err(err) => return LineResult::Error(line.clone(), err),
+                    }
-                    },
 
                     (Some(ClassifiedCommand::External(left)), None) => {
                         match left.run(ctx, input, StreamNext::Last).await {
@@ -333,9 +344,7 @@ async fn process_line(readline: Result<String, ReadlineError>, ctx: &mut Context
 
             LineResult::Success(line.clone())
         }
-        Err(ReadlineError::Interrupted) => {
+        Err(ReadlineError::Interrupted) => LineResult::CtrlC,
-            LineResult::Error("".to_string(), ShellError::string("CTRL-C"))
-        }
         Err(ReadlineError::Eof) => {
             println!("CTRL-D");
             LineResult::Break

src/commands/cd.rs

@@ -5,7 +5,7 @@ use std::path::PathBuf;
 
 pub fn cd(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let env = args.env.lock().unwrap();
-    let latest = env.last().unwrap();
+    let latest = env.back().unwrap();
 
     match latest.obj {
         Value::Filesystem => {
@@ -13,17 +13,23 @@ pub fn cd(args: CommandArgs) -> Result<OutputStream, ShellError> {
             let path = match args.positional.first() {
                 None => match dirs::home_dir() {
                     Some(o) => o,
-                    _ => return Err(ShellError::string("Can not change to home directory")),
+                    _ => {
+                        return Err(ShellError::maybe_labeled_error(
+                            "Can not change to home directory",
+                            "can not go to home",
+                            args.name_span,
+                        ))
+                    }
                 },
                 Some(v) => {
                     let target = v.as_string()?.clone();
                     match dunce::canonicalize(cwd.join(&target).as_path()) {
                         Ok(p) => p,
                         Err(_) => {
-                            return Err(ShellError::labeled_error(
+                            return Err(ShellError::maybe_labeled_error(
                                 "Can not change to directory",
                                 "directory not found",
-                                args.positional[0].span.clone(),
+                                Some(args.positional[0].span.clone()),
                             ));
                         }
                     }
@@ -35,10 +41,10 @@ pub fn cd(args: CommandArgs) -> Result<OutputStream, ShellError> {
                 Ok(_) => {}
                 Err(_) => {
                     if args.positional.len() > 0 {
-                        return Err(ShellError::labeled_error(
+                        return Err(ShellError::maybe_labeled_error(
                             "Can not change to directory",
                             "directory not found",
-                            args.positional[0].span.clone(),
+                            Some(args.positional[0].span.clone()),
                         ));
                     } else {
                         return Err(ShellError::string("Can not change to directory"));

src/commands/classified.rs

@@ -4,6 +4,7 @@ use crate::parser::lexer::{Span, Spanned};
 use crate::parser::registry::Args;
 use crate::prelude::*;
 use bytes::{BufMut, BytesMut};
+use futures::stream::StreamExt;
 use futures_codec::{Decoder, Encoder, Framed};
 use std::io::{Error, ErrorKind};
 use std::path::PathBuf;
@@ -109,44 +110,47 @@ impl InternalCommand {
         context: &mut Context,
         input: ClassifiedInputStream,
     ) -> Result<InputStream, ShellError> {
-        let result = context.run_command(
+        let mut result = context.run_command(
             self.command,
             self.name_span.clone(),
             self.args,
             input.objects,
         )?;
-        let env = context.env.clone();
+        let mut stream = VecDeque::new();
-
+        while let Some(item) = result.next().await {
-        let stream = result.filter_map(move |v| match v {
+            match item {
-            ReturnValue::Action(action) => match action {
+                ReturnValue::Value(Value::Error(err)) => {
-                CommandAction::ChangePath(path) => {
+                    return Err(*err);
-                    env.lock().unwrap().last_mut().map(|x| {
-                        x.path = path;
-                        x
-                    });
-                    futures::future::ready(None)
                 }
-                CommandAction::Enter(obj) => {
+                ReturnValue::Action(action) => match action {
-                    let new_env = Environment {
+                    CommandAction::ChangePath(path) => {
-                        obj: obj,
+                        context.env.lock().unwrap().back_mut().map(|x| {
-                        path: PathBuf::from("/"),
+                            x.path = path;
-                    };
+                            x
-                    env.lock().unwrap().push(new_env);
+                        });
-                    futures::future::ready(None)
-                }
-                CommandAction::Exit => {
-                    let mut v = env.lock().unwrap();
-                    if v.len() == 1 {
-                        std::process::exit(0);
                     }
-                    v.pop();
+                    CommandAction::Enter(obj) => {
-                    futures::future::ready(None)
+                        let new_env = Environment {
+                            obj: obj,
+                            path: PathBuf::from("/"),
+                        };
+                        context.env.lock().unwrap().push_back(new_env);
+                    }
+                    CommandAction::Exit => match context.env.lock().unwrap().pop_back() {
+                        Some(Environment {
+                            obj: Value::Filesystem,
+                            ..
+                        }) => std::process::exit(0),
+                        None => std::process::exit(-1),
+                        _ => {}
+                    },
+                },
+
+                ReturnValue::Value(v) => {
+                    stream.push_back(v);
                 }
-            },
+            }
-
+        }
-            ReturnValue::Value(v) => futures::future::ready(Some(v)),
-        });
-
         Ok(stream.boxed() as InputStream)
     }
 }
@@ -265,7 +269,7 @@ impl ExternalCommand {
             }
             process = Exec::shell(new_arg_string);
         }
-        process = process.cwd(context.env.lock().unwrap().first().unwrap().path());
+        process = process.cwd(context.env.lock().unwrap().front().unwrap().path());
 
         let mut process = match stream_next {
             StreamNext::Last => process,

src/commands/command.rs

@@ -8,7 +8,7 @@ use std::path::PathBuf;
 
 pub struct CommandArgs {
     pub host: Arc<Mutex<dyn Host + Send>>,
-    pub env: Arc<Mutex<Vec<Environment>>>,
+    pub env: Arc<Mutex<VecDeque<Environment>>>,
     pub name_span: Option<Span>,
     pub positional: Vec<Spanned<Value>>,
     pub named: indexmap::IndexMap<String, Value>,

src/commands/enter.rs

@@ -7,14 +7,18 @@ use std::path::{Path, PathBuf};
 
 pub fn enter(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        return Err(ShellError::string("open requires a filepath or url"));
+        return Err(ShellError::maybe_labeled_error(
+            "open requires a path or url",
+            "missing path",
+            args.name_span,
+        ));
     }
 
     let cwd = args
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/first.rs

@@ -5,15 +5,11 @@ use crate::prelude::*;
 
 pub fn first(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        if let Some(span) = args.name_span {
+        return Err(ShellError::maybe_labeled_error(
-            return Err(ShellError::labeled_error(
+            "First requires an amount",
-                "First requires an amount",
+            "needs parameter",
-                "needs parameter",
+            args.name_span,
-                span,
+        ));
-            ));
-        } else {
-            return Err(ShellError::string("first requires an amount."));
-        }
     }
 
     let amount = args.positional[0].as_i64();

src/commands/from_xml.rs

@@ -15,7 +15,7 @@ fn from_node_to_value<'a, 'd>(n: &roxmltree::Node<'a, 'd>) -> Value {
             .filter(|x| match x {
                 Value::Primitive(Primitive::String(f)) => {
                     if f.trim() == "" {
-                        false 
+                        false
                     } else {
                         true
                     }
@@ -57,11 +57,16 @@ pub fn from_xml_string_to_value(s: String) -> Value {
 
 pub fn from_xml(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let out = args.input;
+    let span = args.name_span;
     Ok(out
-        .map(|a| match a {
+        .map(move |a| match a {
-            Value::Primitive(Primitive::String(s)) => ReturnValue::Value(from_xml_string_to_value(s)),
+            Value::Primitive(Primitive::String(s)) => {
-            _ => ReturnValue::Value(Value::Error(Box::new(ShellError::string(
+                ReturnValue::Value(from_xml_string_to_value(s))
+            }
+            _ => ReturnValue::Value(Value::Error(Box::new(ShellError::maybe_labeled_error(
                 "Trying to convert XML from non-string".to_string(),
+                "given non-string",
+                span,
             )))),
         })
         .boxed())

src/commands/get.rs

@@ -1,17 +1,19 @@
 use crate::errors::ShellError;
 use crate::object::Value;
+use crate::parser::lexer::Span;
 use crate::prelude::*;
 
-fn get_member(path: &str, obj: &Value) -> Option<Value> {
+fn get_member(path: &str, span: Span, obj: &Value) -> Option<Value> {
     let mut current = obj;
     for p in path.split(".") {
         match current.get_data_by_key(p) {
             Some(v) => current = v,
             None => {
-                return Some(Value::Error(Box::new(ShellError::string(format!(
+                return Some(Value::Error(Box::new(ShellError::labeled_error(
-                    "Object field name not found: {}",
+                    "Unknown field",
-                    p
+                    "object missing field",
-                )))))
+                    span,
+                ))));
             }
         }
     }
@@ -21,15 +23,11 @@ fn get_member(path: &str, obj: &Value) -> Option<Value> {
 
 pub fn get(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        if let Some(span) = args.name_span {
+        return Err(ShellError::maybe_labeled_error(
-            return Err(ShellError::labeled_error(
+            "Get requires a field or field path",
-                "Get requires a field or field path",
+            "needs parameter",
-                "needs parameter",
+            args.name_span,
-                span,
+        ));
-            ));
-        } else {
-            return Err(ShellError::string("get requires fields."));
-        }
     }
 
     let amount = args.positional[0].as_i64();
@@ -44,7 +42,11 @@ pub fn get(args: CommandArgs) -> Result<OutputStream, ShellError> {
             .boxed());
     }
 
-    let fields: Result<Vec<String>, _> = args.positional.iter().map(|a| a.as_string()).collect();
+    let fields: Result<Vec<(String, Span)>, _> = args
+        .positional
+        .iter()
+        .map(|a| (a.as_string().map(|x| (x, a.span))))
+        .collect();
     let fields = fields?;
 
     let stream = args
@@ -52,7 +54,7 @@ pub fn get(args: CommandArgs) -> Result<OutputStream, ShellError> {
         .map(move |item| {
             let mut result = VecDeque::new();
             for field in &fields {
-                match get_member(field, &item) {
+                match get_member(&field.0, field.1, &item) {
                     Some(Value::List(l)) => {
                         for item in l {
                             result.push_back(ReturnValue::Value(item.copy()));

src/commands/ls.rs

@@ -7,8 +7,8 @@ use std::path::{Path, PathBuf};
 
 pub fn ls(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let env = args.env.lock().unwrap();
-    let path = env.last().unwrap().path.to_path_buf();
+    let path = env.back().unwrap().path.to_path_buf();
-    let obj = &env.last().unwrap().obj;
+    let obj = &env.back().unwrap().obj;
     let mut full_path = PathBuf::from(path);
     match &args.positional.get(0) {
         Some(Spanned {
@@ -31,7 +31,11 @@ pub fn ls(args: CommandArgs) -> Result<OutputStream, ShellError> {
                             s.span,
                         ));
                     } else {
-                        return Err(ShellError::string(e.to_string()));
+                        return Err(ShellError::maybe_labeled_error(
+                            e.to_string(),
+                            e.to_string(),
+                            args.name_span,
+                        ));
                     }
                 }
                 Ok(o) => o,

src/commands/open.rs

@@ -6,14 +6,18 @@ use std::path::{Path, PathBuf};
 
 pub fn open(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        return Err(ShellError::string("open requires a filepath or url"));
+        return Err(ShellError::maybe_labeled_error(
+            "Open requires a path or url",
+            "needs path or url",
+            args.name_span,
+        ));
     }
 
     let cwd = args
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/pick.rs

@@ -5,15 +5,11 @@ use crate::prelude::*;
 
 pub fn pick(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        if let Some(span) = args.name_span {
+        return Err(ShellError::maybe_labeled_error(
-            return Err(ShellError::labeled_error(
+            "Pick requires fields",
-                "Pick requires fields",
+            "needs parameter",
-                "needs parameter",
+            args.name_span,
-                span,
+        ));
-            ));
-        } else {
-            return Err(ShellError::string("pick requires fields."));
-        }
     }
 
     let fields: Result<Vec<String>, _> = args.positional.iter().map(|a| a.as_string()).collect();

src/commands/reject.rs

@@ -5,15 +5,11 @@ use crate::prelude::*;
 
 pub fn reject(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        if let Some(span) = args.name_span {
+        return Err(ShellError::maybe_labeled_error(
-            return Err(ShellError::labeled_error(
+            "Reject requires fields",
-                "Reject requires fields",
+            "needs parameter",
-                "needs parameter",
+            args.name_span,
-                span,
+        ));
-            ));
-        } else {
-            return Err(ShellError::string("reject requires fields."));
-        }
     }
 
     let fields: Result<Vec<String>, _> = args.positional.iter().map(|a| a.as_string()).collect();

src/commands/save.rs

@@ -6,7 +6,11 @@ use std::path::{Path, PathBuf};
 
 pub fn save(args: SinkCommandArgs) -> Result<(), ShellError> {
     if args.positional.len() == 0 {
-        return Err(ShellError::string("save requires a filepath"));
+        return Err(ShellError::maybe_labeled_error(
+            "Save requires a filepath",
+            "needs path",
+            args.name_span,
+        ));
     }
 
     let cwd = args
@@ -14,7 +18,7 @@ pub fn save(args: SinkCommandArgs) -> Result<(), ShellError> {
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/size.rs

@@ -6,14 +6,18 @@ use std::fs::File;
 use std::io::prelude::*;
 
 pub fn size(args: CommandArgs) -> Result<OutputStream, ShellError> {
-    if args.positional.is_empty() {
+    if args.positional.len() == 0 {
-        return Err(ShellError::string("size requires at least one file"));
+        return Err(ShellError::maybe_labeled_error(
+            "Size requires a filepath",
+            "needs path",
+            args.name_span,
+        ));
     }
     let cwd = args
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/skip.rs

@@ -3,15 +3,11 @@ use crate::prelude::*;
 
 pub fn skip(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        if let Some(span) = args.name_span {
+        return Err(ShellError::maybe_labeled_error(
-            return Err(ShellError::labeled_error(
+            "Skip requires an amount",
-                "Skip requires an amount",
+            "needs parameter",
-                "needs parameter",
+            args.name_span,
-                span,
+        ));
-            ));
-        } else {
-            return Err(ShellError::string("skip requires an amount."));
-        }
     }
 
     let amount = args.positional[0].as_i64();

src/commands/view.rs

@@ -4,15 +4,11 @@ use prettyprint::PrettyPrinter;
 
 pub fn view(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.positional.len() == 0 {
-        if let Some(span) = args.name_span {
+        return Err(ShellError::maybe_labeled_error(
-            return Err(ShellError::labeled_error(
+            "View requires a filename",
-                "View requires a filename",
+            "needs parameter",
-                "needs parameter",
+            args.name_span,
-                span,
+        ));
-            ));
-        } else {
-            return Err(ShellError::string("view requires a filename."));
-        }
     }
 
     let target = match args.positional[0].as_string() {
@@ -34,7 +30,7 @@ pub fn view(args: CommandArgs) -> Result<OutputStream, ShellError> {
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/where_.rs

@@ -25,8 +25,12 @@ impl Command for Where {
 }
 
 pub fn r#where(args: CommandArgs) -> Result<OutputStream, ShellError> {
-    if args.positional.is_empty() {
+    if args.positional.len() == 0 {
-        return Err(ShellError::string("select requires a field"));
+        return Err(ShellError::maybe_labeled_error(
+            "Where requires a condition",
+            "needs condition",
+            args.name_span,
+        ));
     }
 
     let block = args.positional[0].as_block()?;

src/context.rs

@@ -13,16 +13,18 @@ pub struct Context {
     commands: IndexMap<String, Arc<dyn Command>>,
     sinks: IndexMap<String, Arc<dyn Sink>>,
     crate host: Arc<Mutex<dyn Host + Send>>,
-    crate env: Arc<Mutex<Vec<Environment>>>,
+    crate env: Arc<Mutex<VecDeque<Environment>>>,
 }
 
 impl Context {
     crate fn basic() -> Result<Context, Box<dyn Error>> {
+        let mut env = VecDeque::new();
+        env.push_back(Environment::basic()?);
         Ok(Context {
             commands: indexmap::IndexMap::new(),
             sinks: indexmap::IndexMap::new(),
             host: Arc::new(Mutex::new(crate::env::host::BasicHost)),
-            env: Arc::new(Mutex::new(vec![Environment::basic()?])),
+            env: Arc::new(Mutex::new(env)),
         })
     }
 

src/errors.rs

@@ -53,6 +53,20 @@ impl ShellError {
         )
     }
 
+    crate fn maybe_labeled_error(
+        msg: impl Into<String>,
+        label: impl Into<String>,
+        span: Option<Span>,
+    ) -> ShellError {
+        match span {
+            Some(span) => ShellError::diagnostic(
+                Diagnostic::new(Severity::Error, msg.into())
+                    .with_label(Label::new_primary(span).with_message(label.into())),
+            ),
+            None => ShellError::string(msg),
+        }
+    }
+
     crate fn string(title: impl Into<String>) -> ShellError {
         ShellError::String(StringError::new(title.into(), Value::nothing()))
     }