"Fix" preserves-tool binary output by adding a streaming write mode

implementations/rust/preserves-tools/src/bin/preserves-tool.rs

@@ -417,6 +417,7 @@ fn convert(c: Convert) -> io::Result<()> {
         }
         OutputFormat::Binary => {
             let mut p = PackedWriter::new(io::stdout());
+            p.start_sequence()?;
             Box::new(move |v| {
                 if write_ann {
                     p.write(&mut IOValueDomainCodec, v)?;
@@ -467,7 +468,7 @@ impl Quote {
 fn output_one(q: &Quote, v: &IOValue) -> io::Result<()> {
     match q.output_format {
         OutputFormat::Binary =>
-            PackedWriter::new(io::stdout()).write(&mut IOValueDomainCodec, v),
+            PackedWriter::new(io::stdout()).streaming().write(&mut IOValueDomainCodec, v),
         OutputFormat::Text => {
             TextWriter::new(io::stdout())
                 .set_escape_spaces(q.escape_spaces())
@@ -483,6 +484,9 @@ fn output_one(q: &Quote, v: &IOValue) -> io::Result<()> {
 }
 
 fn quote(q: Quote) -> io::Result<()> {
+    if matches!(q.output_format, OutputFormat::Text) {
+        PackedWriter::new(io::stdout()).start_sequence()?;
+    }
     match &q.output {
         QuotationOutput::ByteString => {
             let mut buf = Vec::new();

implementations/rust/preserves/src/value/packed/writer.rs

@@ -16,6 +16,7 @@ pub struct PackedWriter<W: io::Write> {
     w: W,
     buffer: IOList,
     items: Vec<Vec<IOList>>,
+    streaming: bool,
 }
 
 impl PackedWriter<&mut Vec<u8>> {
@@ -55,6 +56,7 @@ impl<W: io::Write> PackedWriter<W> {
             w: write,
             buffer: IOList::new(),
             items: Vec::new(),
+            streaming: false,
         }
     }
 
@@ -86,15 +88,29 @@ impl<W: io::Write> PackedWriter<W> {
     }
 
     pub fn finish_item(&mut self) -> io::Result<()> {
-        let buffer = std::mem::replace(&mut self.buffer, IOList::new());
-        match self.items.last_mut() {
-            Some(iols) => Ok(iols.push(buffer)),
-            None => buffer.write_to(&mut self.w),
+        if self.buffer.len() > 0 {
+            let mut buffer = std::mem::replace(&mut self.buffer, IOList::new());
+            match self.items.last_mut() {
+                Some(iols) => iols.push(buffer),
+                None => {
+                    if self.streaming {
+                        let mut len = IOList::new();
+                        varint(&mut len, buffer.len());
+                        len.write_to(&mut self.w)?;
+                    }
+                    buffer.write_to(&mut self.w)?;
+                }
+            }
         }
+        Ok(())
+    }
+
+    fn at_toplevel(&self) -> bool {
+        self.items.is_empty()
     }
 
     pub fn finish_item_if_toplevel(&mut self) -> io::Result<()> {
-        if self.items.is_empty() {
+        if self.at_toplevel() {
             self.finish_item()?;
         }
         Ok(())
@@ -117,6 +133,14 @@ impl<W: io::Write> PackedWriter<W> {
         }
         self.finish_item_if_toplevel()
     }
+
+    pub fn streaming(mut self) -> Self {
+        if !self.at_toplevel() {
+            panic!("Attempt to enable streaming mode mid-way through serialization");
+        }
+        self.streaming = true;
+        self
+    }
 }
 
 macro_rules! fits_in_bytes {
@@ -338,12 +362,24 @@ impl<W: io::Write> Writer for PackedWriter<W>
 
     #[inline(always)]
     fn start_sequence(&mut self) -> io::Result<()> {
-        self.start_seq()
+        if self.at_toplevel() && !self.streaming {
+            self.write_tag(Tag::Sequence);
+            self.finish_item()?;
+            self.streaming = true;
+            Ok(())
+        } else {
+            self.start_seq()
+        }
     }
 
     #[inline(always)]
     fn end_sequence(&mut self) -> io::Result<()> {
-        self.finish_seq(Tag::Sequence, false)
+        if self.at_toplevel() && self.streaming {
+            self.streaming = false;
+            Ok(())
+        } else {
+            self.finish_seq(Tag::Sequence, false)
+        }
     }
 
     #[inline(always)]