Add support for 'open' (#573)

crates/nu-protocol/src/pipeline_data.rs

@@ -87,10 +87,21 @@ impl PipelineData {
                     span, // FIXME?
                 }
             }
-            PipelineData::ByteStream(s, ..) => Value::Binary {
-                val: s.flatten().collect(),
-                span, // FIXME?
-            },
+            PipelineData::ByteStream(s, ..) => {
+                let mut output = vec![];
+
+                for item in s {
+                    match item {
+                        Ok(s) => output.extend(&s),
+                        Err(err) => return Value::Error { error: err },
+                    }
+                }
+
+                Value::Binary {
+                    val: output,
+                    span, // FIXME?
+                }
+            }
         }
     }
 
@@ -110,7 +121,7 @@ impl PipelineData {
             PipelineData::ListStream(s, ..) => Ok(s.into_string(separator, config)),
             PipelineData::StringStream(s, ..) => s.into_string(separator),
             PipelineData::ByteStream(s, ..) => {
-                Ok(String::from_utf8_lossy(&s.flatten().collect::<Vec<_>>()).to_string())
+                Ok(String::from_utf8_lossy(&s.into_vec()?).to_string())
             }
         }
     }
@@ -324,9 +335,12 @@ impl Iterator for PipelineIterator {
                 },
                 Err(err) => Value::Error { error: err },
             }),
-            PipelineData::ByteStream(stream, span, ..) => stream.next().map(|x| Value::Binary {
-                val: x,
-                span: *span,
+            PipelineData::ByteStream(stream, span, ..) => stream.next().map(|x| match x {
+                Ok(x) => Value::Binary {
+                    val: x,
+                    span: *span,
+                },
+                Err(err) => Value::Error { error: err },
             }),
         }
     }

crates/nu-protocol/src/value/stream.rs

@@ -10,12 +10,17 @@ use std::{
 /// A single buffer of binary data streamed over multiple parts. Optionally contains ctrl-c that can be used
 /// to break the stream.
 pub struct ByteStream {
-    pub stream: Box<dyn Iterator<Item = Vec<u8>> + Send + 'static>,
+    pub stream: Box<dyn Iterator<Item = Result<Vec<u8>, ShellError>> + Send + 'static>,
     pub ctrlc: Option<Arc<AtomicBool>>,
 }
 impl ByteStream {
-    pub fn into_vec(self) -> Vec<u8> {
-        self.flatten().collect::<Vec<u8>>()
+    pub fn into_vec(self) -> Result<Vec<u8>, ShellError> {
+        let mut output = vec![];
+        for item in self.stream {
+            output.append(&mut item?);
+        }
+
+        Ok(output)
     }
 }
 impl Debug for ByteStream {
@@ -25,7 +30,7 @@ impl Debug for ByteStream {
 }
 
 impl Iterator for ByteStream {
-    type Item = Vec<u8>;
+    type Item = Result<Vec<u8>, ShellError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         if let Some(ctrlc) = &self.ctrlc {