fixup drain_to functions to respect a 0 write as an temporary end of …

…target but not a hard error

decodecorpus_files/abc.txt

@@ -0,0 +1 @@
+abcdefghijklmnopqrstuvwxy

src/decoding/decodebuffer.rs

@@ -231,10 +231,7 @@ impl DecodeBuffer {
     pub fn drain_to_window_size_writer(&mut self, mut sink: impl Write) -> Result<usize, Error> {
         match self.can_drain_to_window_size() {
             None => Ok(0),
-            Some(can_drain) => {
-                self.drain_to(can_drain, |buf| write_all_bytes(&mut sink, buf))?;
-                Ok(can_drain)
-            }
+            Some(can_drain) => self.drain_to(can_drain, |buf| write_all_bytes(&mut sink, buf)),
         }
     }
 
@@ -255,10 +252,8 @@ impl DecodeBuffer {
     }
 
     pub fn drain_to_writer(&mut self, mut sink: impl Write) -> Result<usize, Error> {
-        let len = self.buffer.len();
-        self.drain_to(len, |buf| write_all_bytes(&mut sink, buf))?;
-
-        Ok(len)
+        let write_limit = self.buffer.len();
+        self.drain_to(write_limit, |buf| write_all_bytes(&mut sink, buf))
     }
 
     pub fn read_all(&mut self, target: &mut [u8]) -> Result<usize, Error> {
@@ -280,9 +275,9 @@ impl DecodeBuffer {
         &mut self,
         amount: usize,
         mut write_bytes: impl FnMut(&[u8]) -> (usize, Result<(), Error>),
-    ) -> Result<(), Error> {
+    ) -> Result<usize, Error> {
         if amount == 0 {
-            return Ok(());
+            return Ok(0);
         }
 
         struct DrainGuard<'a> {
@@ -329,10 +324,11 @@ impl DecodeBuffer {
             }
         }
 
+        let amount_written = drain_guard.amount;
         // Make sure we don't accidentally drop `DrainGuard` earlier.
         drop(drain_guard);
 
-        Ok(())
+        Ok(amount_written)
     }
 }
 
@@ -341,6 +337,7 @@ fn write_all_bytes(mut sink: impl Write, buf: &[u8]) -> (usize, Result<(), Error
     let mut written = 0;
     while written < buf.len() {
         match sink.write(&buf[written..]) {
+            Ok(0) => return (written, Ok(())),
             Ok(w) => written += w,
             Err(e) => return (written, Err(e)),
         }

src/io_nostd.rs

@@ -153,3 +153,18 @@ where
         (*self).flush()
     }
 }
+
+impl Write for &mut [u8] {
+    #[inline]
+    fn write(&mut self, data: &[u8]) -> Result<usize, Error> {
+        let amt = core::cmp::min(data.len(), self.len());
+        let (a, b) = core::mem::take(self).split_at_mut(amt);
+        a.copy_from_slice(&data[..amt]);
+        *self = b;
+        Ok(amt)
+    }
+
+    fn flush(&mut self) -> Result<(), Error> {
+        Ok(())
+    }
+}

src/tests/mod.rs

@@ -373,6 +373,30 @@ fn test_streaming() {
     }
 }
 
+#[test]
+fn test_incremental_read() {
+    use crate::frame_decoder::FrameDecoder;
+
+    let mut unread_compressed_content =
+        include_bytes!("../../decodecorpus_files/abc.txt.zst").as_slice();
+
+    let mut frame_dec = FrameDecoder::new();
+    frame_dec.reset(&mut unread_compressed_content).unwrap();
+
+    let mut output = [0u8; 3];
+    let (_, written) = frame_dec
+        .decode_from_to(&unread_compressed_content, &mut output)
+        .unwrap();
+
+    assert_eq!(written, 3);
+    assert_eq!(output.map(char::from), ['a', 'b', 'c']);
+
+    assert!(frame_dec.is_finished());
+    let written = frame_dec.collect_to_writer(&mut &mut output[..]).unwrap();
+    assert_eq!(written, 3);
+    assert_eq!(output.map(char::from), ['d', 'e', 'f']);
+}
+
 #[test]
 #[cfg(not(feature = "std"))]
 fn test_streaming_no_std() {