Repository: BurntSushi/rust-csv (language: rust)

Issue:
Do not flush underlying writer when buffer becomes full
#### What version of the `csv` crate are you using?

master: `fcdbea357739348b9eed2a0239e180213be78af9`

#### Briefly describe the question, bug or feature request.

As mentioned in [this issue of `xsv`](https://github.com/BurntSushi/xsv/issues/151#issuecomment-538790858), it seems that rust-csv is flushing the underlying writer when its internal buffer becomes full. This causes issues for `xsv` as the underlying writer is a [`TabWriter`](https://github.com/BurntSushi/tabwriter) and calling `flush` causes side-effects other than writing out the buffer (specifically, it resets the internal state, "breaking" alignment of subsequently written data)

#### Include a complete program demonstrating a problem.

The reproduction with `xsv` is attached on the linked issue - using a small `rust-csv` buffer size leads to broken tabular layout.

#### What is the expected or desired behavior of the code above?

When the internal buffer becomes full, its contents should be written to the underlying writer, but the underlying writer should not be `flush`ed. This seems to match with other buffering IO libraries, such as [`BufferedWriter` in Java](https://github.com/openjdk-mirror/jdk7u-jdk/blob/f4d80957e89a19a29bb9f9807d2a28351ed7f7df/src/share/classes/java/io/BufferedWriter.java#L119-L124) and [rust's stdlib](https://github.com/rust-lang/rust/blob/master/src/libstd/io/buffered.rs#L490)



Discussion / hints:

Fix patch:
diff --git a/src/writer.rs b/src/writer.rs
index f4080357..6ab442fa 100644
--- a/src/writer.rs
+++ b/src/writer.rs
@@ -1043,7 +1043,7 @@ impl<W: io::Write> Writer<W> {
                     self.state.fields_written += 1;
                     return Ok(());
                 }
-                WriteResult::OutputFull => self.flush()?,
+                WriteResult::OutputFull => self.flush_buf()?,
             }
         }
     }
@@ -1055,12 +1055,19 @@ impl<W: io::Write> Writer<W> {
     ///
     /// Note that this also flushes the underlying writer.
     pub fn flush(&mut self) -> io::Result<()> {
+        self.flush_buf()?;
+        self.wtr.as_mut().unwrap().flush()?;
+        Ok(())
+    }
+
+    /// Flush the contents of the internal buffer to the underlying writer,
+    /// without flushing the underlying writer.
+    fn flush_buf(&mut self) -> io::Result<()> {
         self.state.panicked = true;
         let result = self.wtr.as_mut().unwrap().write_all(self.buf.readable());
         self.state.panicked = false;
         result?;
         self.buf.clear();
-        self.wtr.as_mut().unwrap().flush()?;
         Ok(())
     }
 
@@ -1082,7 +1089,7 @@ impl<W: io::Write> Writer<W> {
             self.buf.written(nout);
             match res {
                 WriteResult::InputEmpty => return Ok(()),
-                WriteResult::OutputFull => self.flush()?,
+                WriteResult::OutputFull => self.flush_buf()?,
             }
         }
     }
@@ -1098,7 +1105,7 @@ impl<W: io::Write> Writer<W> {
                     self.state.fields_written = 0;
                     return Ok(());
                 }
-                WriteResult::OutputFull => self.flush()?,
+                WriteResult::OutputFull => self.flush_buf()?,
             }
         }
     }
@@ -1179,6 +1186,8 @@ impl Buffer {
 mod tests {
     use serde::{serde_if_integer128, Serialize};
 
+    use std::io::{self, Write};
+
     use crate::byte_record::ByteRecord;
     use crate::error::ErrorKind;
     use crate::string_record::StringRecord;
@@ -1306,6 +1315,47 @@ mod tests {
         assert_eq!(wtr_as_string(wtr), "a,b,c\na\n");
     }
 
+    #[test]
+    fn full_buffer_should_not_flush_underlying() {
+        struct MarkWriteAndFlush(Vec<u8>);
+
+        impl MarkWriteAndFlush {
+            fn to_str(self) -> String {
+                String::from_utf8(self.0).unwrap()
+            }
+        }
+
+        impl Write for MarkWriteAndFlush {
+            fn write(&mut self, data: &[u8]) -> io::Result<usize> {
+                self.0.write(b">")?;
+                let written = self.0.write(data)?;
+                self.0.write(b"<")?;
+
+                Ok(written)
+            }
+
+            fn flush(&mut self) -> io::Result<()> {
+                self.0.write(b"!")?;
+                Ok(())
+            }
+        }
+
+        let underlying = MarkWriteAndFlush(vec![]);
+        let mut wtr =
+            WriterBuilder::new().buffer_capacity(4).from_writer(underlying);
+
+        wtr.write_byte_record(&ByteRecord::from(vec!["a", "b"])).unwrap();
+        wtr.write_byte_record(&ByteRecord::from(vec!["c", "d"])).unwrap();
+        wtr.flush().unwrap();
+        wtr.write_byte_record(&ByteRecord::from(vec!["e", "f"])).unwrap();
+
+        let got = wtr.into_inner().unwrap().to_str();
+
+        // As the buffer size is 4 we should write each record separately, and
+        // flush when explicitly called and implictly in into_inner.
+        assert_eq!(got, ">a,b\n<>c,d\n<!>e,f\n<!");
+    }
+
     #[test]
     fn serialize_with_headers() {
         #[derive(Serialize)]


Test patch: