Makes the serializers compatible with async I/O

Author: Tpt

README.md

@@ -29,16 +29,15 @@ assert_eq!(reader.read_next_event()?, JsonEvent::Eof);
 Writer example:
 
 ```rust
-use json_event_parser::{JsonWriter, JsonEvent};
+use json_event_parser::{ToWriteJsonWriter, JsonEvent};
 
-let mut buffer = Vec::new();
-let mut writer = JsonWriter::from_writer(&mut buffer);
+let mut writer = ToWriteJsonWriter::new(Vec::new());
 writer.write_event(JsonEvent::StartObject)?;
 writer.write_event(JsonEvent::ObjectKey("foo".into()))?;
 writer.write_event(JsonEvent::Number("1".into()))?;
 writer.write_event(JsonEvent::EndObject)?;
 
-assert_eq!(buffer.as_slice(), b"{\"foo\":1}");
+assert_eq!(writer.finish()?.as_slice(), b"{\"foo\":1}");
 # std::io::Result::Ok(())
 ```
 

fuzz/fuzz_targets/parse.rs

@@ -1,7 +1,7 @@
 #![no_main]
 
 use json_event_parser::{
-    JsonEvent, JsonWriter, LowLevelJsonReader, LowLevelJsonReaderResult, SyntaxError,
+    JsonEvent, LowLevelJsonReader, LowLevelJsonReaderResult, SyntaxError, ToWriteJsonWriter,
 };
 use libfuzzer_sys::fuzz_target;
 
@@ -10,7 +10,7 @@ fn parse_chunks(chunks: &[&[u8]]) -> (String, Option<SyntaxError>) {
     let mut input_cursor = 0;
     let mut output_buffer = Vec::new();
     let mut reader = LowLevelJsonReader::new();
-    let mut writer = JsonWriter::from_writer(&mut output_buffer);
+    let mut writer = ToWriteJsonWriter::new(&mut output_buffer);
     for (i, chunk) in chunks.iter().enumerate() {
         input_buffer.extend_from_slice(chunk);
         loop {
@@ -21,7 +21,8 @@ fn parse_chunks(chunks: &[&[u8]]) -> (String, Option<SyntaxError>) {
             input_cursor += consumed_bytes;
             match event {
                 Some(Ok(JsonEvent::Eof)) => {
-                    return (String::from_utf8(output_buffer).unwrap(), None)
+                    writer.finish().unwrap();
+                    return (String::from_utf8(output_buffer).unwrap(), None);
                 }
                 Some(Ok(event)) => writer.write_event(event).unwrap(),
                 Some(Err(e)) => return (String::from_utf8(output_buffer).unwrap(), Some(e)),

src/lib.rs

@@ -19,7 +19,9 @@ pub use crate::read::{
     FromBufferJsonReader, FromReadJsonReader, LowLevelJsonReader, LowLevelJsonReaderResult,
     ParseError, SyntaxError, TextPosition,
 };
-pub use crate::write::JsonWriter;
+#[cfg(feature = "async-tokio")]
+pub use crate::write::ToTokioAsyncWriteJsonWriter;
+pub use crate::write::{LowLevelJsonWriter, ToWriteJsonWriter};
 use std::borrow::Cow;
 
 /// Possible events during JSON parsing.

src/read.rs

@@ -37,9 +37,9 @@ pub struct FromReadJsonReader<R: Read> {
 }
 
 impl<R: Read> FromReadJsonReader<R> {
-    pub fn new(read: R) -> Self {
+    pub const fn new(read: R) -> Self {
         Self {
-            input_buffer: Vec::with_capacity(MIN_BUFFER_SIZE),
+            input_buffer: Vec::new(),
             input_buffer_start: 0,
             input_buffer_end: 0,
             max_buffer_size: MAX_BUFFER_SIZE,
@@ -140,9 +140,9 @@ pub struct FromTokioAsyncReadJsonReader<R: AsyncRead + Unpin> {
 
 #[cfg(feature = "async-tokio")]
 impl<R: AsyncRead + Unpin> FromTokioAsyncReadJsonReader<R> {
-    pub fn new(read: R) -> Self {
+    pub const fn new(read: R) -> Self {
         Self {
-            input_buffer: Vec::with_capacity(MIN_BUFFER_SIZE),
+            input_buffer: Vec::new(),
             input_buffer_start: 0,
             input_buffer_end: 0,
             max_buffer_size: MAX_BUFFER_SIZE,
@@ -234,7 +234,7 @@ pub struct FromBufferJsonReader<'a> {
 }
 
 impl<'a> FromBufferJsonReader<'a> {
-    pub fn new(buffer: &'a [u8]) -> Self {
+    pub const fn new(buffer: &'a [u8]) -> Self {
         Self {
             input_buffer: buffer,
             parser: LowLevelJsonReader::new(),

src/write.rs

@@ -1,71 +1,149 @@
 use crate::JsonEvent;
 use std::io::{Error, ErrorKind, Result, Write};
+#[cfg(feature = "async-tokio")]
+use tokio::io::{AsyncWrite, AsyncWriteExt};
 
-/// A JSON streaming writer.
+/// A JSON streaming writer writing to a [`Write`] implementation.
 ///
 /// ```
-/// use json_event_parser::{JsonWriter, JsonEvent};
+/// use json_event_parser::{ToWriteJsonWriter, JsonEvent};
 ///
-/// let mut buffer = Vec::new();
-/// let mut writer = JsonWriter::from_writer(&mut buffer);
+/// let mut writer = ToWriteJsonWriter::new(Vec::new());
 /// writer.write_event(JsonEvent::StartObject)?;
 /// writer.write_event(JsonEvent::ObjectKey("foo".into()))?;
 /// writer.write_event(JsonEvent::Number("1".into()))?;
 /// writer.write_event(JsonEvent::EndObject)?;
 ///
-/// assert_eq!(buffer.as_slice(), b"{\"foo\":1}");
+/// assert_eq!(writer.finish()?.as_slice(), b"{\"foo\":1}");
+/// # std::io::Result::Ok(())
+/// ```
+pub struct ToWriteJsonWriter<W: Write> {
+    write: W,
+    writer: LowLevelJsonWriter,
+}
+
+impl<W: Write> ToWriteJsonWriter<W> {
+    pub const fn new(write: W) -> Self {
+        Self {
+            write,
+            writer: LowLevelJsonWriter::new(),
+        }
+    }
+
+    pub fn write_event(&mut self, event: JsonEvent<'_>) -> Result<()> {
+        self.writer.write_event(event, &mut self.write)
+    }
+
+    pub fn finish(self) -> Result<W> {
+        self.writer.validate_eof()?;
+        Ok(self.write)
+    }
+}
+
+/// A JSON streaming writer writing to an [`AsyncWrite`] implementation.
+///
+/// ```
+/// use json_event_parser::{ToTokioAsyncWriteJsonWriter, JsonEvent};
 ///
+/// # #[tokio::main(flavor = "current_thread")]
+/// # async fn main() -> ::std::io::Result<()> {
+/// let mut writer = ToTokioAsyncWriteJsonWriter::new(Vec::new());
+/// writer.write_event(JsonEvent::StartObject).await?;
+/// writer.write_event(JsonEvent::ObjectKey("foo".into())).await?;
+/// writer.write_event(JsonEvent::Number("1".into())).await?;
+/// writer.write_event(JsonEvent::EndObject).await?;
+/// assert_eq!(writer.finish()?.as_slice(), b"{\"foo\":1}");
+/// # Ok(())
+/// # }
+/// ```
+#[cfg(feature = "async-tokio")]
+pub struct ToTokioAsyncWriteJsonWriter<W: AsyncWrite + Unpin> {
+    write: W,
+    writer: LowLevelJsonWriter,
+    buffer: Vec<u8>,
+}
+
+#[cfg(feature = "async-tokio")]
+impl<W: AsyncWrite + Unpin> ToTokioAsyncWriteJsonWriter<W> {
+    pub const fn new(write: W) -> Self {
+        Self {
+            write,
+            writer: LowLevelJsonWriter::new(),
+            buffer: Vec::new(),
+        }
+    }
+
+    pub async fn write_event(&mut self, event: JsonEvent<'_>) -> Result<()> {
+        self.writer.write_event(event, &mut self.buffer)?;
+        self.write.write_all(&self.buffer).await?;
+        self.buffer.clear();
+        Ok(())
+    }
+
+    pub fn finish(self) -> Result<W> {
+        self.writer.validate_eof()?;
+        Ok(self.write)
+    }
+}
+
+/// A low-level JSON streaming writer writing to a [`Write`] implementation.
+///
+/// YOu probably want to use [`ToWriteJsonWriter`] instead.
+///
+/// ```
+/// use json_event_parser::{JsonEvent, LowLevelJsonWriter};
+///
+/// let mut writer = LowLevelJsonWriter::new();
+/// let mut output = Vec::new();
+/// writer.write_event(JsonEvent::StartObject, &mut output)?;
+/// writer.write_event(JsonEvent::ObjectKey("foo".into()), &mut output)?;
+/// writer.write_event(JsonEvent::Number("1".into()), &mut output)?;
+/// writer.write_event(JsonEvent::EndObject, &mut output)?;
+///
+/// assert_eq!(output.as_slice(), b"{\"foo\":1}");
 /// # std::io::Result::Ok(())
 /// ```
-pub struct JsonWriter<W: Write> {
-    writer: W,
+
+#[derive(Default)]
+pub struct LowLevelJsonWriter {
     state_stack: Vec<JsonState>,
     element_written: bool,
 }
 
-impl<W: Write> JsonWriter<W> {
-    pub fn from_writer(writer: W) -> Self {
+impl LowLevelJsonWriter {
+    pub const fn new() -> Self {
         Self {
-            writer,
             state_stack: Vec::new(),
             element_written: false,
         }
     }
 
-    pub fn into_inner(self) -> W {
-        self.writer
-    }
-
-    pub fn inner(&mut self) -> &mut W {
-        &mut self.writer
-    }
-
-    pub fn write_event(&mut self, event: JsonEvent<'_>) -> Result<()> {
+    pub fn write_event(&mut self, event: JsonEvent<'_>, mut write: impl Write) -> Result<()> {
         match event {
             JsonEvent::String(s) => {
-                self.before_value()?;
-                write_escaped_json_string(&s, &mut self.writer)
+                self.before_value(&mut write)?;
+                write_escaped_json_string(&s, write)
             }
             JsonEvent::Number(number) => {
-                self.before_value()?;
-                self.writer.write_all(number.as_bytes())
+                self.before_value(&mut write)?;
+                write.write_all(number.as_bytes())
             }
             JsonEvent::Boolean(b) => {
-                self.before_value()?;
-                self.writer.write_all(if b { b"true" } else { b"false" })
+                self.before_value(&mut write)?;
+                write.write_all(if b { b"true" } else { b"false" })
             }
             JsonEvent::Null => {
-                self.before_value()?;
-                self.writer.write_all(b"null")
+                self.before_value(&mut write)?;
+                write.write_all(b"null")
             }
             JsonEvent::StartArray => {
-                self.before_value()?;
+                self.before_value(&mut write)?;
                 self.state_stack.push(JsonState::OpenArray);
-                self.writer.write_all(b"[")
+                write.write_all(b"[")
             }
             JsonEvent::EndArray => match self.state_stack.pop() {
                 Some(JsonState::OpenArray) | Some(JsonState::ContinuationArray) => {
-                    self.writer.write_all(b"]")
+                    write.write_all(b"]")
                 }
                 Some(s) => {
                     self.state_stack.push(s);
@@ -80,13 +158,13 @@ impl<W: Write> JsonWriter<W> {
                 )),
             },
             JsonEvent::StartObject => {
-                self.before_value()?;
+                self.before_value(&mut write)?;
                 self.state_stack.push(JsonState::OpenObject);
-                self.writer.write_all(b"{")
+                write.write_all(b"{")
             }
             JsonEvent::EndObject => match self.state_stack.pop() {
                 Some(JsonState::OpenObject) | Some(JsonState::ContinuationObject) => {
-                    self.writer.write_all(b"}")
+                    write.write_all(b"}")
                 }
                 Some(s) => {
                     self.state_stack.push(s);
@@ -103,7 +181,7 @@ impl<W: Write> JsonWriter<W> {
             JsonEvent::ObjectKey(key) => {
                 match self.state_stack.pop() {
                     Some(JsonState::OpenObject) => (),
-                    Some(JsonState::ContinuationObject) => self.writer.write_all(b",")?,
+                    Some(JsonState::ContinuationObject) => write.write_all(b",")?,
                     _ => {
                         return Err(Error::new(
                             ErrorKind::InvalidInput,
@@ -113,8 +191,8 @@ impl<W: Write> JsonWriter<W> {
                 }
                 self.state_stack.push(JsonState::ContinuationObject);
                 self.state_stack.push(JsonState::ObjectValue);
-                write_escaped_json_string(&key, &mut self.writer)?;
-                self.writer.write_all(b":")
+                write_escaped_json_string(&key, &mut write)?;
+                write.write_all(b":")
             }
             JsonEvent::Eof => Err(Error::new(
                 ErrorKind::InvalidInput,
@@ -123,15 +201,15 @@ impl<W: Write> JsonWriter<W> {
         }
     }
 
-    fn before_value(&mut self) -> Result<()> {
+    fn before_value(&mut self, mut write: impl Write) -> Result<()> {
         match self.state_stack.pop() {
             Some(JsonState::OpenArray) => {
                 self.state_stack.push(JsonState::ContinuationArray);
                 Ok(())
             }
             Some(JsonState::ContinuationArray) => {
                 self.state_stack.push(JsonState::ContinuationArray);
-                self.writer.write_all(b",")?;
+                write.write_all(b",")?;
                 Ok(())
             }
             Some(last_state @ JsonState::OpenObject)
@@ -156,6 +234,22 @@ impl<W: Write> JsonWriter<W> {
             }
         }
     }
+
+    fn validate_eof(&self) -> Result<()> {
+        if !self.state_stack.is_empty() {
+            return Err(Error::new(
+                ErrorKind::InvalidInput,
+                "The written JSON is not balanced: an object or an array has not been closed",
+            ));
+        }
+        if !self.element_written {
+            return Err(Error::new(
+                ErrorKind::InvalidInput,
+                "A JSON file can't be empty",
+            ));
+        }
+        Ok(())
+    }
 }
 
 enum JsonState {
@@ -166,37 +260,37 @@ enum JsonState {
     ObjectValue,
 }
 
-fn write_escaped_json_string(s: &str, sink: &mut impl Write) -> Result<()> {
-    sink.write_all(b"\"")?;
+fn write_escaped_json_string(s: &str, mut write: impl Write) -> Result<()> {
+    write.write_all(b"\"")?;
     let mut buffer = [b'\\', b'u', 0, 0, 0, 0];
     for c in s.chars() {
         match c {
-            '\\' => sink.write_all(b"\\\\"),
-            '"' => sink.write_all(b"\\\""),
+            '\\' => write.write_all(b"\\\\"),
+            '"' => write.write_all(b"\\\""),
             c => {
                 if c < char::from(32) {
                     match c {
-                        '\u{08}' => sink.write_all(b"\\b"),
-                        '\u{0C}' => sink.write_all(b"\\f"),
-                        '\n' => sink.write_all(b"\\n"),
-                        '\r' => sink.write_all(b"\\r"),
-                        '\t' => sink.write_all(b"\\t"),
+                        '\u{08}' => write.write_all(b"\\b"),
+                        '\u{0C}' => write.write_all(b"\\f"),
+                        '\n' => write.write_all(b"\\n"),
+                        '\r' => write.write_all(b"\\r"),
+                        '\t' => write.write_all(b"\\t"),
                         c => {
                             let mut c = c as u8;
                             for i in (2..6).rev() {
                                 let ch = c % 16;
                                 buffer[i] = if ch < 10 { b'0' + ch } else { b'A' + ch - 10 };
                                 c /= 16;
                             }
-                            sink.write_all(&buffer)
+                            write.write_all(&buffer)
                         }
                     }
                 } else {
-                    sink.write_all(c.encode_utf8(&mut buffer[2..]).as_bytes())
+                    write.write_all(c.encode_utf8(&mut buffer[2..]).as_bytes())
                 }
             }
         }?;
     }
-    sink.write_all(b"\"")?;
+    write.write_all(b"\"")?;
     Ok(())
 }