Merge branch 'master' into split-with-empty-cols

src/commands/classified.rs

@@ -122,10 +122,11 @@ impl InternalCommand {
             self.name_span.clone(),
             context.source_map.clone(),
             self.args,
-            source,
+            &source,
             objects,
         );
 
+        let result = trace_out_stream!(target: "nu::trace_stream::internal", source: &source, "output" = result);
         let mut result = result.values;
 
         let mut stream = VecDeque::new();

src/commands/command.rs

@@ -422,6 +422,25 @@ pub enum CommandAction {
     LeaveShell,
 }
 
+impl ToDebug for CommandAction {
+    fn fmt_debug(&self, f: &mut fmt::Formatter, _source: &str) -> fmt::Result {
+        match self {
+            CommandAction::ChangePath(s) => write!(f, "action:change-path={}", s),
+            CommandAction::AddSpanSource(u, source) => {
+                write!(f, "action:add-span-source={}@{:?}", u, source)
+            }
+            CommandAction::Exit => write!(f, "action:exit"),
+            CommandAction::EnterShell(s) => write!(f, "action:enter-shell={}", s),
+            CommandAction::EnterValueShell(t) => {
+                write!(f, "action:enter-value-shell={:?}", t.debug())
+            }
+            CommandAction::PreviousShell => write!(f, "action:previous-shell"),
+            CommandAction::NextShell => write!(f, "action:next-shell"),
+            CommandAction::LeaveShell => write!(f, "action:leave-shell"),
+        }
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub enum ReturnSuccess {
     Value(Tagged<Value>),
@@ -430,6 +449,16 @@ pub enum ReturnSuccess {
 
 pub type ReturnValue = Result<ReturnSuccess, ShellError>;
 
+impl ToDebug for ReturnValue {
+    fn fmt_debug(&self, f: &mut fmt::Formatter, source: &str) -> fmt::Result {
+        match self {
+            Err(err) => write!(f, "{}", err.debug(source)),
+            Ok(ReturnSuccess::Value(v)) => write!(f, "{:?}", v.debug()),
+            Ok(ReturnSuccess::Action(a)) => write!(f, "{}", a.debug(source)),
+        }
+    }
+}
+
 impl From<Tagged<Value>> for ReturnValue {
     fn from(input: Tagged<Value>) -> ReturnValue {
         Ok(ReturnSuccess::Value(input))
@@ -469,7 +498,7 @@ pub trait WholeStreamCommand: Send + Sync {
         Signature {
             name: self.name().to_string(),
             positional: vec![],
-            rest_positional: true,
+            rest_positional: None,
             named: indexmap::IndexMap::new(),
             is_filter: true,
         }
@@ -491,7 +520,7 @@ pub trait PerItemCommand: Send + Sync {
         Signature {
             name: self.name().to_string(),
             positional: vec![],
-            rest_positional: true,
+            rest_positional: None,
             named: indexmap::IndexMap::new(),
             is_filter: true,
         }

src/commands/from_bson.rs

@@ -2,7 +2,7 @@ use crate::commands::WholeStreamCommand;
 use crate::object::base::OF64;
 use crate::object::{Primitive, TaggedDictBuilder, Value};
 use crate::prelude::*;
-use bson::{decode_document, Bson, spec::BinarySubtype};
+use bson::{decode_document, spec::BinarySubtype, Bson};
 
 pub struct FromBSON;
 
@@ -47,71 +47,80 @@ fn convert_bson_value_to_nu_value(v: &Bson, tag: impl Into<Tag>) -> Tagged<Value
         Bson::Boolean(b) => Value::Primitive(Primitive::Boolean(*b)).tagged(tag),
         Bson::Null => Value::Primitive(Primitive::String(String::from(""))).tagged(tag),
         Bson::RegExp(r, opts) => {
-             let mut collected = TaggedDictBuilder::new(tag);
-             collected.insert_tagged(
-                 "$regex".to_string(),
-                 Value::Primitive(Primitive::String(String::from(r))).tagged(tag),
-             );
-             collected.insert_tagged(
-                 "$options".to_string(),
-                 Value::Primitive(Primitive::String(String::from(opts))).tagged(tag),
-             );
-             collected.into_tagged_value()
+            let mut collected = TaggedDictBuilder::new(tag);
+            collected.insert_tagged(
+                "$regex".to_string(),
+                Value::Primitive(Primitive::String(String::from(r))).tagged(tag),
+            );
+            collected.insert_tagged(
+                "$options".to_string(),
+                Value::Primitive(Primitive::String(String::from(opts))).tagged(tag),
+            );
+            collected.into_tagged_value()
         }
+        // TODO: Add Int32 to nushell?
         Bson::I32(n) => Value::Primitive(Primitive::Int(*n as i64)).tagged(tag),
         Bson::I64(n) => Value::Primitive(Primitive::Int(*n as i64)).tagged(tag),
         Bson::JavaScriptCode(js) => {
-             let mut collected = TaggedDictBuilder::new(tag);
-             collected.insert_tagged(
-                 "$javascript".to_string(),
-                 Value::Primitive(Primitive::String(String::from(js))).tagged(tag),
-             );
-             collected.into_tagged_value()
+            let mut collected = TaggedDictBuilder::new(tag);
+            collected.insert_tagged(
+                "$javascript".to_string(),
+                Value::Primitive(Primitive::String(String::from(js))).tagged(tag),
+            );
+            collected.into_tagged_value()
         }
         Bson::JavaScriptCodeWithScope(js, doc) => {
-             let mut collected = TaggedDictBuilder::new(tag);
-             collected.insert_tagged(
-                 "$javascript".to_string(),
-                 Value::Primitive(Primitive::String(String::from(js))).tagged(tag),
-             );
-             collected.insert_tagged(
-                 "$scope".to_string(),
-                 convert_bson_value_to_nu_value(&Bson::Document(doc.to_owned()), tag),
-             );
-             collected.into_tagged_value()
+            let mut collected = TaggedDictBuilder::new(tag);
+            collected.insert_tagged(
+                "$javascript".to_string(),
+                Value::Primitive(Primitive::String(String::from(js))).tagged(tag),
+            );
+            collected.insert_tagged(
+                "$scope".to_string(),
+                convert_bson_value_to_nu_value(&Bson::Document(doc.to_owned()), tag),
+            );
+            collected.into_tagged_value()
         }
         Bson::TimeStamp(ts) => {
-             let mut collected = TaggedDictBuilder::new(tag);
-             collected.insert_tagged(
-                 "$timestamp".to_string(),
-                 Value::Primitive(Primitive::Int(*ts as i64)).tagged(tag),
-             );
-             collected.into_tagged_value()
+            let mut collected = TaggedDictBuilder::new(tag);
+            collected.insert_tagged(
+                "$timestamp".to_string(),
+                Value::Primitive(Primitive::Int(*ts as i64)).tagged(tag),
+            );
+            collected.into_tagged_value()
         }
         Bson::Binary(bst, bytes) => {
-             let mut collected = TaggedDictBuilder::new(tag);
-             collected.insert_tagged(
-                 "$binary_subtype".to_string(),
-                 match bst {
-                     BinarySubtype::UserDefined(u) => Value::Primitive(Primitive::Int(*u as i64)),
-                     _ => Value::Primitive(Primitive::String(binary_subtype_to_string(*bst))),
-                 }.tagged(tag)
-             );
-             collected.insert_tagged(
-                 "$binary".to_string(),
-                 Value::Binary(bytes.to_owned()).tagged(tag),
-             );
-             collected.into_tagged_value()
+            let mut collected = TaggedDictBuilder::new(tag);
+            collected.insert_tagged(
+                "$binary_subtype".to_string(),
+                match bst {
+                    BinarySubtype::UserDefined(u) => Value::Primitive(Primitive::Int(*u as i64)),
+                    _ => Value::Primitive(Primitive::String(binary_subtype_to_string(*bst))),
+                }
+                .tagged(tag),
+            );
+            collected.insert_tagged(
+                "$binary".to_string(),
+                Value::Binary(bytes.to_owned()).tagged(tag),
+            );
+            collected.into_tagged_value()
+        }
+        Bson::ObjectId(obj_id) => {
+            let mut collected = TaggedDictBuilder::new(tag);
+            collected.insert_tagged(
+                "$object_id".to_string(),
+                Value::Primitive(Primitive::String(obj_id.to_hex())).tagged(tag),
+            );
+            collected.into_tagged_value()
         }
-        Bson::ObjectId(obj_id) => Value::Primitive(Primitive::String(obj_id.to_hex())).tagged(tag),
         Bson::UtcDatetime(dt) => Value::Primitive(Primitive::Date(*dt)).tagged(tag),
         Bson::Symbol(s) => {
-             let mut collected = TaggedDictBuilder::new(tag);
-             collected.insert_tagged(
-                 "$symbol".to_string(),
-                 Value::Primitive(Primitive::String(String::from(s))).tagged(tag),
-             );
-             collected.into_tagged_value()
+            let mut collected = TaggedDictBuilder::new(tag);
+            collected.insert_tagged(
+                "$symbol".to_string(),
+                Value::Primitive(Primitive::String(String::from(s))).tagged(tag),
+            );
+            collected.into_tagged_value()
         }
     }
 }
@@ -125,7 +134,8 @@ fn binary_subtype_to_string(bst: BinarySubtype) -> String {
         BinarySubtype::Uuid => "uuid",
         BinarySubtype::Md5 => "md5",
         _ => unreachable!(),
-    }.to_string()
+    }
+    .to_string()
 }
 
 #[derive(Debug)]

src/commands/get.rs

@@ -22,33 +22,41 @@ impl WholeStreamCommand for Get {
     ) -> Result<OutputStream, ShellError> {
         args.process(registry, get)?.run()
     }
+
     fn signature(&self) -> Signature {
-        Signature::build("get").rest()
+        Signature::build("get").rest(SyntaxType::Member)
     }
 }
 
 fn get_member(path: &Tagged<String>, obj: &Tagged<Value>) -> Result<Tagged<Value>, ShellError> {
-    let mut current = obj;
+    let mut current = Some(obj);
     for p in path.split(".") {
-        match current.get_data_by_key(p) {
-            Some(v) => current = v,
-            None => {
+        if let Some(obj) = current {
+            current = match obj.get_data_by_key(p) {
+                Some(v) => Some(v),
+                None =>
                 // Before we give up, see if they gave us a path that matches a field name by itself
-                match obj.get_data_by_key(&path.item) {
-                    Some(v) => return Ok(v.clone()),
-                    None => {
-                        return Err(ShellError::labeled_error(
-                            "Unknown column",
-                            "table missing column",
-                            path.span(),
-                        ));
+                {
+                    match obj.get_data_by_key(&path.item) {
+                        Some(v) => return Ok(v.clone()),
+                        None => {
+                            return Err(ShellError::labeled_error(
+                                "Unknown column",
+                                "table missing column",
+                                path.span(),
+                            ));
+                        }
                     }
                 }
             }
         }
     }
 
-    Ok(current.clone())
+    match current {
+        Some(v) => Ok(v.clone()),
+        None => Ok(Value::nothing().tagged(obj.tag)),
+    }
+    // Ok(current.clone())
 }
 
 pub fn get(

src/commands/last.rs

@@ -44,7 +44,7 @@ fn last(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, S
             "Value is too low",
             "expected a positive integer",
             args.expect_nth(0)?.span(),
-        ))
+        ));
     }
 
     let stream = async_stream_block! {

src/commands/mkdir.rs

@@ -27,7 +27,7 @@ impl PerItemCommand for Mkdir {
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("mkdir").rest()
+        Signature::build("mkdir").rest(SyntaxType::Path)
     }
 }
 

src/commands/open.rs

@@ -44,7 +44,8 @@ fn run(call_info: &CallInfo, shell_manager: &ShellManager) -> Result<OutputStrea
     {
         file => file,
     };
-    let path_str = path.as_string()?;
+    let path_buf = path.as_path()?;
+    let path_str = path_buf.display().to_string();
     let path_span = path.span();
     let name_span = call_info.name_span;
     let has_raw = call_info.args.has("raw");
@@ -426,14 +427,16 @@ pub fn parse_string_as_value(
     name_span: Span,
 ) -> Result<Tagged<Value>, ShellError> {
     match extension {
-        Some(x) if x == "csv" => crate::commands::from_csv::from_csv_string_to_value(
-            contents,
-            false,
-            contents_tag,
-        )
-        .map_err(move |_| {
-            ShellError::labeled_error("Could not open as CSV", "could not open as CSV", name_span)
-        }),
+        Some(x) if x == "csv" => {
+            crate::commands::from_csv::from_csv_string_to_value(contents, false, contents_tag)
+                .map_err(move |_| {
+                    ShellError::labeled_error(
+                        "Could not open as CSV",
+                        "could not open as CSV",
+                        name_span,
+                    )
+                })
+        }
         Some(x) if x == "toml" => {
             crate::commands::from_toml::from_toml_string_to_value(contents, contents_tag).map_err(
                 move |_| {
@@ -507,9 +510,9 @@ pub fn parse_binary_as_value(
             crate::commands::from_bson::from_bson_bytes_to_value(contents, contents_tag).map_err(
                 move |_| {
                     ShellError::labeled_error(
-                       "Could not open as BSON",
-                       "could not open as BSON",
-                       name_span,
+                        "Could not open as BSON",
+                        "could not open as BSON",
+                        name_span,
                     )
                 },
             )

src/commands/pick.rs

@@ -17,7 +17,7 @@ impl WholeStreamCommand for Pick {
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("pick").rest()
+        Signature::build("pick").rest(SyntaxType::Any)
     }
 
     fn run(

src/commands/plugin.rs

@@ -3,6 +3,7 @@ use crate::errors::ShellError;
 use crate::parser::registry;
 use crate::prelude::*;
 use derive_new::new;
+use log::trace;
 use serde::{self, Deserialize, Serialize};
 use std::io::prelude::*;
 use std::io::BufReader;
@@ -64,6 +65,8 @@ pub fn filter_plugin(
     args: CommandArgs,
     registry: &CommandRegistry,
 ) -> Result<OutputStream, ShellError> {
+    trace!("filter_plugin :: {}", path);
+
     let args = args.evaluate_once(registry)?;
 
     let mut child = std::process::Command::new(path)
@@ -80,6 +83,8 @@ pub fn filter_plugin(
 
     let call_info = args.call_info.clone();
 
+    trace!("filtering :: {:?}", call_info);
+
     let stream = bos
         .chain(args.input.values)
         .chain(eos)
@@ -95,7 +100,14 @@ pub fn filter_plugin(
 
                 let request = JsonRpc::new("begin_filter", call_info.clone());
                 let request_raw = serde_json::to_string(&request).unwrap();
-                let _ = stdin.write(format!("{}\n", request_raw).as_bytes()); // TODO: Handle error
+                match stdin.write(format!("{}\n", request_raw).as_bytes()) {
+                    Ok(_) => {}
+                    Err(err) => {
+                        let mut result = VecDeque::new();
+                        result.push_back(Err(ShellError::unexpected(format!("{}", err))));
+                        return result;
+                    }
+                }
 
                 let mut input = String::new();
                 match reader.read_line(&mut input) {
@@ -140,7 +152,15 @@ pub fn filter_plugin(
                 let mut reader = BufReader::new(stdout);
 
                 let request: JsonRpc<std::vec::Vec<Value>> = JsonRpc::new("end_filter", vec![]);
-                let request_raw = serde_json::to_string(&request).unwrap();
+                let request_raw = match serde_json::to_string(&request) {
+                    Ok(req) => req,
+                    Err(err) => {
+                        let mut result = VecDeque::new();
+                        result.push_back(Err(ShellError::unexpected(format!("{}", err))));
+                        return result;
+                    }
+                };
+
                 let _ = stdin.write(format!("{}\n", request_raw).as_bytes()); // TODO: Handle error
 
                 let mut input = String::new();

src/commands/reject.rs

@@ -24,7 +24,7 @@ impl WholeStreamCommand for Reject {
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("reject").rest()
+        Signature::build("reject").rest(SyntaxType::Member)
     }
 }
 

src/commands/save.rs

@@ -59,7 +59,7 @@ fn save(
             // If there is no filename, check the metadata for the origin filename
             if input.len() > 0 {
                 let origin = input[0].origin();
-                match origin.map(|x| source_map.get(&x)).flatten() {
+                match origin.and_then(|x| source_map.get(&x)) {
                     Some(path) => match path {
                         SpanSource::File(file) => {
                             full_path.push(Path::new(file));

src/commands/sort_by.rs

@@ -4,13 +4,19 @@ use crate::prelude::*;
 
 pub struct SortBy;
 
+#[derive(Deserialize)]
+pub struct SortByArgs {
+    rest: Vec<Tagged<String>>,
+    reverse: bool,
+}
+
 impl WholeStreamCommand for SortBy {
     fn run(
         &self,
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        sort_by(args, registry)
+        args.process(registry, sort_by)?.run()
     }
 
     fn name(&self) -> &str {
@@ -18,43 +24,32 @@ impl WholeStreamCommand for SortBy {
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("sort-by").switch("reverse")
+        Signature::build("sort-by")
+            .rest(SyntaxType::String)
+            .switch("reverse")
     }
 }
 
-fn sort_by(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
-    let (input, args) = args.parts();
+fn sort_by(
+    SortByArgs { reverse, rest }: SortByArgs,
+    mut context: RunnableContext,
+) -> Result<OutputStream, ShellError> {
+    Ok(OutputStream::new(async_stream_block! {
+        let mut vec = context.input.drain_vec().await;
 
-    let fields: Result<Vec<_>, _> = args
-        .positional
-        .iter()
-        .flatten()
-        .map(|a| a.as_string())
-        .collect();
-
-    let fields = fields?;
-
-    let output = input.values.collect::<Vec<_>>();
-
-    let reverse = args.has("reverse");
-    let output = output.map(move |mut vec| {
         let calc_key = |item: &Tagged<Value>| {
-            fields
-                .iter()
+            rest.iter()
                 .map(|f| item.get_data_by_key(f).map(|i| i.clone()))
                 .collect::<Vec<Option<Tagged<Value>>>>()
         };
         if reverse {
-            vec.sort_by_cached_key(|item| {
-                std::cmp::Reverse(calc_key(item))
-            });
+            vec.sort_by_cached_key(|item| std::cmp::Reverse(calc_key(item)));
         } else {
             vec.sort_by_cached_key(calc_key);
+        };
+
+        for item in vec {
+            yield item.into();
         }
-
-        vec.into_iter().collect::<VecDeque<_>>()
-    });
-
-    Ok(output.flatten_stream().from_input_stream())
+    }))
 }

src/commands/split_column.rs

@@ -31,7 +31,7 @@ impl WholeStreamCommand for SplitColumn {
         Signature::build("split-column")
             .required("separator", SyntaxType::Any)
             .switch("collapse-empty")
-            .rest()
+            .rest(SyntaxType::Member)
     }
 }
 

src/commands/tags.rs

@@ -38,7 +38,7 @@ fn tags(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream,
                 dict.insert("end", Value::int(span.end as i64));
                 tags.insert_tagged("span", dict.into_tagged_value());
 
-                match origin.map(|x| source_map.get(&x)).flatten() {
+                match origin.and_then(|x| source_map.get(&x)) {
                     Some(SpanSource::File(source)) => {
                         tags.insert("origin", Value::string(source));
                     }

src/commands/to_bson.rs

@@ -0,0 +1,231 @@
+use crate::commands::WholeStreamCommand;
+use crate::object::{Dictionary, Primitive, Value};
+use crate::prelude::*;
+use bson::{encode_document, oid::ObjectId, spec::BinarySubtype, Bson, Document};
+use std::convert::TryInto;
+
+pub struct ToBSON;
+
+impl WholeStreamCommand for ToBSON {
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        to_bson(args, registry)
+    }
+
+    fn name(&self) -> &str {
+        "to-bson"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("to-bson")
+    }
+}
+
+pub fn value_to_bson_value(v: &Value) -> Bson {
+    match v {
+        Value::Primitive(Primitive::Boolean(b)) => Bson::Boolean(*b),
+        Value::Primitive(Primitive::Bytes(b)) => Bson::I64(*b as i64),
+        Value::Primitive(Primitive::Date(d)) => Bson::UtcDatetime(*d),
+        Value::Primitive(Primitive::EndOfStream) => Bson::Null,
+        Value::Primitive(Primitive::BeginningOfStream) => Bson::Null,
+        Value::Primitive(Primitive::Float(f)) => Bson::FloatingPoint(f.into_inner()),
+        Value::Primitive(Primitive::Int(i)) => Bson::I64(*i),
+        Value::Primitive(Primitive::Nothing) => Bson::Null,
+        Value::Primitive(Primitive::String(s)) => Bson::String(s.clone()),
+        Value::Primitive(Primitive::Path(s)) => Bson::String(s.display().to_string()),
+        Value::List(l) => Bson::Array(l.iter().map(|x| value_to_bson_value(x)).collect()),
+        Value::Block(_) => Bson::Null,
+        Value::Binary(b) => Bson::Binary(BinarySubtype::Generic, b.clone()),
+        Value::Object(o) => object_value_to_bson(o),
+    }
+}
+
+// object_value_to_bson handles all Objects, even those that correspond to special
+// types (things like regex or javascript code).
+fn object_value_to_bson(o: &Dictionary) -> Bson {
+    let mut it = o.entries.iter();
+    if it.len() > 2 {
+        return generic_object_value_to_bson(o);
+    }
+    match it.next() {
+        Some((regex, tagged_regex_value)) if regex == "$regex" => match it.next() {
+            Some((options, tagged_opts_value)) if options == "$options" => {
+                let r: Result<String, _> = tagged_regex_value.try_into();
+                let opts: Result<String, _> = tagged_opts_value.try_into();
+                if r.is_err() || opts.is_err() {
+                    generic_object_value_to_bson(o)
+                } else {
+                    Bson::RegExp(r.unwrap(), opts.unwrap())
+                }
+            }
+            _ => generic_object_value_to_bson(o),
+        },
+        Some((javascript, tagged_javascript_value)) if javascript == "$javascript" => {
+            match it.next() {
+                Some((scope, tagged_scope_value)) if scope == "$scope" => {
+                    let js: Result<String, _> = tagged_javascript_value.try_into();
+                    let s: Result<&Dictionary, _> = tagged_scope_value.try_into();
+                    if js.is_err() || s.is_err() {
+                        generic_object_value_to_bson(o)
+                    } else {
+                        if let Bson::Document(doc) = object_value_to_bson(s.unwrap()) {
+                            Bson::JavaScriptCodeWithScope(js.unwrap(), doc)
+                        } else {
+                            generic_object_value_to_bson(o)
+                        }
+                    }
+                }
+                None => {
+                    let js: Result<String, _> = tagged_javascript_value.try_into();
+                    if js.is_err() {
+                        generic_object_value_to_bson(o)
+                    } else {
+                        Bson::JavaScriptCode(js.unwrap())
+                    }
+                }
+                _ => generic_object_value_to_bson(o),
+            }
+        }
+        Some((timestamp, tagged_timestamp_value)) if timestamp == "$timestamp" => {
+            let ts: Result<i64, _> = tagged_timestamp_value.try_into();
+            if ts.is_err() {
+                generic_object_value_to_bson(o)
+            } else {
+                Bson::TimeStamp(ts.unwrap())
+            }
+        }
+        Some((binary_subtype, tagged_binary_subtype_value))
+            if binary_subtype == "$binary_subtype" =>
+        {
+            match it.next() {
+                Some((binary, tagged_bin_value)) if binary == "$binary" => {
+                    let bst = get_binary_subtype(tagged_binary_subtype_value);
+                    let bin: Result<Vec<u8>, _> = tagged_bin_value.try_into();
+                    if bst.is_none() || bin.is_err() {
+                        generic_object_value_to_bson(o)
+                    } else {
+                        Bson::Binary(bst.unwrap(), bin.unwrap())
+                    }
+                }
+                _ => generic_object_value_to_bson(o),
+            }
+        }
+        Some((object_id, tagged_object_id_value)) if object_id == "$object_id" => {
+            let obj_id: Result<String, _> = tagged_object_id_value.try_into();
+            if obj_id.is_err() {
+                generic_object_value_to_bson(o)
+            } else {
+                let obj_id = ObjectId::with_string(&obj_id.unwrap());
+                if obj_id.is_err() {
+                    generic_object_value_to_bson(o)
+                } else {
+                    Bson::ObjectId(obj_id.unwrap())
+                }
+            }
+        }
+        Some((symbol, tagged_symbol_value)) if symbol == "$symbol" => {
+            let sym: Result<String, _> = tagged_symbol_value.try_into();
+            if sym.is_err() {
+                generic_object_value_to_bson(o)
+            } else {
+                Bson::Symbol(sym.unwrap())
+            }
+        }
+        _ => generic_object_value_to_bson(o),
+    }
+}
+
+fn get_binary_subtype<'a>(tagged_value: &'a Tagged<Value>) -> Option<BinarySubtype> {
+    match tagged_value.item() {
+        Value::Primitive(Primitive::String(s)) => Some(match s.as_ref() {
+            "generic" => BinarySubtype::Generic,
+            "function" => BinarySubtype::Function,
+            "binary_old" => BinarySubtype::BinaryOld,
+            "uuid_old" => BinarySubtype::UuidOld,
+            "uuid" => BinarySubtype::Uuid,
+            "md5" => BinarySubtype::Md5,
+            _ => unreachable!(),
+        }),
+        Value::Primitive(Primitive::Int(i)) => Some(BinarySubtype::UserDefined(*i as u8)),
+        _ => None,
+    }
+}
+
+// generic_object_value_bson handles any Object that does not
+// correspond to a special bson type (things like regex or javascript code).
+fn generic_object_value_to_bson(o: &Dictionary) -> Bson {
+    let mut doc = Document::new();
+    for (k, v) in o.entries.iter() {
+        doc.insert(k.clone(), value_to_bson_value(v));
+    }
+    Bson::Document(doc)
+}
+
+fn shell_encode_document(
+    writer: &mut Vec<u8>,
+    doc: Document,
+    span: Span,
+) -> Result<(), ShellError> {
+    match encode_document(writer, &doc) {
+        Err(e) => Err(ShellError::labeled_error(
+            format!("Failed to encode document due to: {:?}", e),
+            "requires BSON-compatible document",
+            span,
+        )),
+        _ => Ok(()),
+    }
+}
+
+fn bson_value_to_bytes(bson: Bson, span: Span) -> Result<Vec<u8>, ShellError> {
+    let mut out = Vec::new();
+    match bson {
+        Bson::Array(a) => {
+            for v in a.into_iter() {
+                match v {
+                    Bson::Document(d) => shell_encode_document(&mut out, d, span)?,
+                    _ => {
+                        return Err(ShellError::labeled_error(
+                            format!("All top level values must be Documents, got {:?}", v),
+                            "requires BSON-compatible document",
+                            span,
+                        ))
+                    }
+                }
+            }
+        }
+        Bson::Document(d) => shell_encode_document(&mut out, d, span)?,
+        _ => {
+            return Err(ShellError::labeled_error(
+                format!("All top level values must be Documents, got {:?}", bson),
+                "requires BSON-compatible document",
+                span,
+            ))
+        }
+    }
+    Ok(out)
+}
+
+fn to_bson(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
+    let args = args.evaluate_once(registry)?;
+    let name_span = args.name_span();
+    let out = args.input;
+
+    Ok(out
+        .values
+        .map(
+            move |a| match bson_value_to_bytes(value_to_bson_value(&a), name_span) {
+                Ok(x) => ReturnSuccess::value(Value::Binary(x).simple_spanned(name_span)),
+                _ => Err(ShellError::labeled_error_with_secondary(
+                    "Expected an object with BSON-compatible structure from pipeline",
+                    "requires BSON-compatible input: Must be Array or Object",
+                    name_span,
+                    format!("{} originates from here", a.item.type_name()),
+                    a.span(),
+                )),
+            },
+        )
+        .to_output_stream())
+}

src/commands/version.rs

@@ -5,8 +5,6 @@ use crate::parser::registry::Signature;
 use crate::prelude::*;
 use indexmap::IndexMap;
 
-const VERSION: &'static str = env!("CARGO_PKG_VERSION");
-
 pub struct Version;
 
 impl WholeStreamCommand for Version {
@@ -34,7 +32,7 @@ pub fn date(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStrea
     let mut indexmap = IndexMap::new();
     indexmap.insert(
         "version".to_string(),
-        Tagged::from_simple_spanned_item(Value::string(VERSION.to_string()), span),
+        Tagged::from_simple_spanned_item(Value::string(clap::crate_version!()), span),
     );
 
     let value = Tagged::from_simple_spanned_item(Value::Object(Dictionary::from(indexmap)), span);