Make Serde document iteration incremental

docs/EVENT_BACKED_SERDE.md

@@ -8,8 +8,9 @@ semantics.
 
 - `from_str`, `from_slice`, and `from_documents_str` parse into spanful `Node`
   trees before handing values to Serde.
-- `Deserializer::from_str` is document-iterating, but it still owns parsed
-  `Node` documents internally.
+- `Deserializer::from_str` is document-iterating and consumes the public
+  `DocumentStream` one parsed `Node` at a time; reader-backed entrypoints still
+  buffer source bytes before iteration.
 - `DocumentStream` bounds retained parsed documents, but reader-backed Serde
   entrypoints still read all input bytes first.
 - `EventStream` exposes useful parser events, spans, tags, anchors, and document

src/de.rs

@@ -18,7 +18,7 @@
 //! # Ok::<(), saneyaml::Error>(())
 //! ```
 
-use crate::parse::parse_document_results_with_options;
+use crate::parse::{DocumentStream, parse_document_results_with_options};
 use crate::{
     Error, ErrorPathSegment, Mapping, Node, NodeValue, Number, Span, Tag, TaggedValue, Value,
     ast::MergePolicy,
@@ -32,7 +32,7 @@ use serde::de::{
     Visitor,
 };
 use serde::forward_to_deserialize_any;
-use std::{collections::HashSet, io::Read};
+use std::{collections::HashSet, fmt, io::Read};
 
 /// Deserializes a single YAML document from a string.
 pub fn from_str<'de, T>(input: &'de str) -> crate::Result<T>
@@ -229,9 +229,8 @@ fn read_error(err: std::io::Error) -> Error {
 }
 
 /// Streaming Serde deserializer over one or more YAML documents.
-#[derive(Debug)]
 pub struct Deserializer<'de> {
-    documents: std::vec::IntoIter<Document<'de>>,
+    documents: DocumentSource<'de>,
 }
 
 impl<'de> Deserializer<'de> {
@@ -243,8 +242,8 @@ impl<'de> Deserializer<'de> {
 
     /// Creates a streaming deserializer from a YAML string using load options.
     pub fn from_str_with_options(input: &'de str, options: LoadOptions) -> Self {
-        Self::from_document_results(
-            parse_document_results_with_options(input, options),
+        Self::from_document_stream(
+            DocumentStream::from_str_with_options(input, options),
             Some(input),
         )
     }
@@ -257,14 +256,14 @@ impl<'de> Deserializer<'de> {
     /// Creates a streaming deserializer from UTF-8 YAML bytes using load options.
     pub fn from_slice_with_options(input: &'de [u8], options: LoadOptions) -> Self {
         if let Err(error) = options.check_input_len(input.len()) {
-            return Self::from_parse_result(Err(error));
+            return Self::from_error(error);
         }
         match std::str::from_utf8(input) {
             Ok(input) => Self::from_str_with_options(input, options),
-            Err(err) => Self::from_parse_result(Err(Error::encoding(
+            Err(err) => Self::from_error(Error::encoding(
                 "input is not valid UTF-8",
                 utf8_error_span(input, err),
-            ))),
+            )),
         }
     }
 
@@ -282,65 +281,40 @@ impl<'de> Deserializer<'de> {
         R: Read,
     {
         match read_to_end_with_options(reader, options) {
-            Ok(input) => match std::str::from_utf8(&input) {
-                Ok(input) => Self::from_document_results(
-                    parse_document_results_with_options(input, options),
-                    None,
-                ),
-                Err(err) => Self::from_parse_result(Err(Error::encoding(
+            Ok(input) => match String::from_utf8(input) {
+                Ok(input) => Self::from_owned_input(input, options),
+                Err(err) => Self::from_error(Error::encoding(
                     "input is not valid UTF-8",
-                    utf8_error_span(&input, err),
-                ))),
+                    utf8_error_span(err.as_bytes(), err.utf8_error()),
+                )),
             },
-            Err(error) => Self::from_parse_result(Err(error)),
+            Err(error) => Self::from_error(error),
         }
     }
 
-    fn from_parse_result(result: crate::Result<Vec<Node>>) -> Self {
+    fn from_owned_input(input: String, options: LoadOptions) -> Self {
+        Self::from_document_stream(DocumentStream::from_str_with_options(&input, options), None)
+    }
+
+    fn from_document_stream(
+        result: crate::Result<DocumentStream>,
+        input: Option<&'de str>,
+    ) -> Self {
         let documents = match result {
-            Ok(documents) => documents
-                .into_iter()
-                .enumerate()
-                .map(|(index, node)| Document {
-                    node: Ok(node),
-                    input: None,
-                    index,
-                })
-                .collect(),
-            Err(error) => vec![Document {
-                node: Err(error.with_document_index(0)),
-                input: None,
-                index: 0,
-            }],
+            Ok(stream) => DocumentSource::Stream {
+                stream,
+                input,
+                next_index: 0,
+                yielded_document: false,
+            },
+            Err(error) => DocumentSource::PendingError(Some(error)),
         };
-        Self {
-            documents: documents.into_iter(),
-        }
+        Self { documents }
     }
 
-    fn from_document_results(results: Vec<crate::Result<Node>>, input: Option<&'de str>) -> Self {
-        if results.is_empty() {
-            return Self {
-                documents: vec![Document {
-                    node: Ok(Node::null(Span::point(0, 1, 1))),
-                    input,
-                    index: 0,
-                }]
-                .into_iter(),
-            };
-        }
-
+    fn from_error(error: Error) -> Self {
         Self {
-            documents: results
-                .into_iter()
-                .enumerate()
-                .map(|(index, node)| Document {
-                    node: node.map_err(|error| error.with_document_index(index)),
-                    input,
-                    index,
-                })
-                .collect::<Vec<_>>()
-                .into_iter(),
+            documents: DocumentSource::PendingError(Some(error)),
         }
     }
 
@@ -368,12 +342,74 @@ impl<'de> Deserializer<'de> {
     }
 }
 
+impl fmt::Debug for Deserializer<'_> {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("Deserializer")
+            .finish_non_exhaustive()
+    }
+}
+
+enum DocumentSource<'de> {
+    Stream {
+        stream: DocumentStream,
+        input: Option<&'de str>,
+        next_index: usize,
+        yielded_document: bool,
+    },
+    PendingError(Option<Error>),
+    Single(Option<Document<'de>>),
+}
+
+impl<'de> DocumentSource<'de> {
+    fn next(&mut self) -> Option<Document<'de>> {
+        match self {
+            Self::Stream {
+                stream,
+                input,
+                next_index,
+                yielded_document,
+            } => {
+                let result = stream.next();
+                let index = *next_index;
+                match result {
+                    Some(result) => {
+                        *next_index += 1;
+                        *yielded_document = true;
+                        Some(Document {
+                            node: result.map_err(|error| error.with_document_index(index)),
+                            input: *input,
+                            index,
+                        })
+                    }
+                    None if !*yielded_document => {
+                        *yielded_document = true;
+                        *next_index = 1;
+                        Some(Document {
+                            node: Ok(Node::null(Span::point(0, 1, 1))),
+                            input: *input,
+                            index: 0,
+                        })
+                    }
+                    None => None,
+                }
+            }
+            Self::PendingError(error) => error.take().map(|error| Document {
+                node: Err(error.with_document_index(0)),
+                input: None,
+                index: 0,
+            }),
+            Self::Single(document) => document.take(),
+        }
+    }
+}
+
 impl<'de> Iterator for Deserializer<'de> {
     type Item = Deserializer<'de>;
 
     fn next(&mut self) -> Option<Self::Item> {
         self.documents.next().map(|document| Deserializer {
-            documents: vec![document].into_iter(),
+            documents: DocumentSource::Single(Some(document)),
         })
     }
 }
@@ -4234,6 +4270,53 @@ impl<'de> IntoDeserializer<'de, Error> for Value {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use super::{Deserializer, DocumentSource};
+
+    #[test]
+    fn deserializer_parses_documents_when_they_are_requested() {
+        let input = "---\nname: first\n---\nbad: *missing\n";
+        let mut stream = Deserializer::from_str(input);
+
+        let DocumentSource::Stream {
+            next_index,
+            yielded_document,
+            ..
+        } = &stream.documents
+        else {
+            panic!("valid input should initialize a document stream");
+        };
+        assert_eq!(*next_index, 0);
+        assert!(!*yielded_document);
+
+        let first = stream.next().expect("first document");
+        let DocumentSource::Single(Some(document)) = first.documents else {
+            panic!("iterator items should contain one document");
+        };
+        assert!(document.node.is_ok());
+
+        let DocumentSource::Stream {
+            next_index,
+            yielded_document,
+            ..
+        } = &stream.documents
+        else {
+            panic!("stream should remain available after the first document");
+        };
+        assert_eq!(*next_index, 1);
+        assert!(*yielded_document);
+
+        let second = stream.next().expect("later parse error");
+        let DocumentSource::Single(Some(document)) = second.documents else {
+            panic!("iterator items should contain one document");
+        };
+        let error = document.node.expect_err("later document should fail");
+        assert_eq!(error.document_index(), Some(1));
+        assert!(stream.next().is_none());
+    }
+}
+
 impl<'de> de::Deserializer<'de> for TaggedValue {
     type Error = Error;