Get stream errors working

src/cli.rs

@@ -113,7 +113,7 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
 
         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 {

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 => {

src/commands/classified.rs

@@ -4,6 +4,7 @@ use crate::parser::lexer::Span;
 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,15 +110,50 @@ 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 {
+            match item {
+                ReturnValue::Value(Value::Error(err)) => {
+                    return Err(*err);
+                }
+                ReturnValue::Action(action) => match action {
+                    CommandAction::ChangePath(path) => {
+                        context.env.lock().unwrap().back_mut().map(|x| {
+                            x.path = path;
+                            x
+                        });
+                    }
+                    CommandAction::Enter(obj) => {
+                        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);
+                }
+            }
+        }
+        /*
         let stream = result.filter_map(move |v| match v {
+            //ReturnValue::Value(Value::Error(err)) => futures::future::err(Err(err)),
             ReturnValue::Action(action) => match action {
                 CommandAction::ChangePath(path) => {
                     env.lock().unwrap().last_mut().map(|x| {
@@ -146,7 +182,7 @@ impl InternalCommand {
 
             ReturnValue::Value(v) => futures::future::ready(Some(v)),
         });
-
+        */
         Ok(stream.boxed() as InputStream)
     }
 }
@@ -229,7 +265,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

@@ -14,7 +14,7 @@ pub fn enter(args: CommandArgs) -> Result<OutputStream, ShellError> {
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/exit.rs

@@ -1,11 +1,8 @@
 use crate::commands::command::CommandAction;
 use crate::errors::ShellError;
-use crate::object::{Primitive, Value};
-use crate::parser::lexer::Spanned;
 use crate::prelude::*;
-use std::path::{Path, PathBuf};
 
-pub fn exit(args: CommandArgs) -> Result<OutputStream, ShellError> {
+pub fn exit(_args: CommandArgs) -> Result<OutputStream, ShellError> {
     let mut stream = VecDeque::new();
     stream.push_back(ReturnValue::Action(CommandAction::Exit));
     Ok(stream.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
+                    "unknown field",
-                )))))
+                    span,
+                ))));
             }
         }
     }
@@ -44,7 +46,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 +58,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 {

src/commands/open.rs

@@ -13,7 +13,7 @@ pub fn open(args: CommandArgs) -> Result<OutputStream, ShellError> {
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/save.rs

@@ -14,7 +14,7 @@ pub fn save(args: SinkCommandArgs) -> Result<(), ShellError> {
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/size.rs

@@ -13,7 +13,7 @@ pub fn size(args: CommandArgs) -> Result<OutputStream, ShellError> {
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

src/commands/view.rs

@@ -34,7 +34,7 @@ pub fn view(args: CommandArgs) -> Result<OutputStream, ShellError> {
         .env
         .lock()
         .unwrap()
-        .first()
+        .front()
         .unwrap()
         .path()
         .to_path_buf();

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)),
         })
     }