tabnas_transduce/source/
mod.rs

1//! Sources: where `JsonEvents/1` come from.
2//!
3//! - [`ValueSource`] walks a parsed engine value. Always correct, retains
4//!   the whole value: the fallback for every grammar.
5//! - [`ParserSource`] drives a tabnas parse of one text and, in
6//!   [`SourceMode::Incremental`], turns its rule events into source events
7//!   as they happen, for the grammars the differential suite has verified
8//!   ([`capability::incremental`]); in [`SourceMode::Materialize`] it
9//!   parses and walks.
10//! - [`LinesSource`] reads JSON Lines or CSV from any `BufRead` a record
11//!   (or a chunk of records) at a time, bounding memory whatever the
12//!   file's size.
13//!
14//! Every source emits through [`Guarded`], which enforces the source
15//! limits (`max_depth`, `max_key_bytes`, `max_scalar_bytes`), polls the
16//! abort flag and counts the source metrics.
17
18pub mod capability;
19pub mod guard;
20pub mod lines;
21pub mod parser;
22pub(crate) mod rule_events;
23
24pub use guard::Guarded;
25pub use lines::{LineFormat, LinesSource, DEFAULT_CHUNK_BYTES};
26pub use parser::ParserSource;
27
28use crate::error::Fail;
29use crate::event::{JsonEvent, Number};
30use crate::limits::AbortFlag;
31use crate::selector::Selector;
32use crate::sink::{Flow, Sink};
33
34/// How [`ParserSource`] produces its events.
35#[derive(Clone, Debug, PartialEq, Eq)]
36pub enum SourceMode {
37    /// Parse the whole text, then walk the value. Sound for every grammar.
38    Materialize,
39    /// Emit from the engine's rule events as the parse proceeds. Sound for
40    /// the grammars [`capability::incremental`] lists.
41    Incremental { prune: Prune },
42}
43
44/// Which arrays the incremental source empties as it streams them.
45/// Pruning alters the value the engine returns, which the incremental
46/// source discards; it is never applied in `Materialize` mode.
47#[derive(Clone, Debug, PartialEq, Eq)]
48pub enum Prune {
49    Never,
50    /// The array whose elements the selector names (a trailing `[*]`
51    /// names the elements; without one the selector names the array).
52    Under(Selector),
53    AllArrays,
54}
55
56/// The engine's cancel code: what a parse guard that answered `false`
57/// reports, whether the guard is this crate's abort or the grammar's own.
58const CANCEL: &str = "cancel";
59
60/// Map an engine error to a failure. A cancel while the caller's flag is
61/// set is `ABORTED`. A cancel otherwise is a guard the GRAMMAR installed
62/// (tabnas-json refuses nesting deeper than 128, well below the default
63/// `Limits::max_depth`), so the message says so instead of "parse
64/// cancelled", which would read as the caller's doing; it stays
65/// `INPUT_INVALID`, because the document is what the grammar refused. Any
66/// other error is the input's, with the engine's code and position.
67pub(crate) fn engine_failure(error: &tabnas::TabnasError, abort: &AbortFlag) -> Fail {
68    if error.code != CANCEL {
69        return Fail::from_tabnas(error);
70    }
71    if abort.is_aborted() {
72        return Fail::aborted();
73    }
74    let mut fail = Fail::from_tabnas(error);
75    fail.message = format!(
76        "the grammar stopped the parse with a guard of its own ({}: {}); a grammar may refuse \
77         nesting or size below this crate's Limits",
78        error.code,
79        error.detail.trim_end()
80    );
81    fail
82}
83
84/// Something that can drive a sink with one document's events.
85pub trait Source {
86    /// Run to completion, or until the sink stops, or until a failure.
87    /// `Ok(Flow::Stop)` means the sink stopped it; the document was not
88    /// validated past that point.
89    fn run(self, sink: &mut dyn Sink) -> Result<Flow, Fail>;
90}
91
92/// Emit a parsed engine value as events, ending with [`JsonEvent::End`].
93pub struct ValueSource<'v>(pub &'v tabnas::Value);
94
95impl Source for ValueSource<'_> {
96    fn run(self, sink: &mut dyn Sink) -> Result<Flow, Fail> {
97        if walk_value(self.0, sink)? == Flow::Stop {
98            return Ok(Flow::Stop);
99        }
100        sink.event(JsonEvent::End)
101    }
102}
103
104/// Emit one engine value's events (without `End`).
105///
106/// `Undefined` is `null`, as the engine serializes it; the metadata
107/// wrappers (`Text`, `MapRef`, `ListRef`) unwrap to their plain forms.
108pub fn walk_value(value: &tabnas::Value, sink: &mut dyn Sink) -> Result<Flow, Fail> {
109    macro_rules! send {
110        ($ev:expr) => {
111            if sink.event($ev)? == Flow::Stop {
112                return Ok(Flow::Stop);
113            }
114        };
115    }
116    match value {
117        tabnas::Value::Undefined | tabnas::Value::Null => send!(JsonEvent::Null),
118        tabnas::Value::Bool(b) => send!(JsonEvent::Bool(*b)),
119        tabnas::Value::Number(n) => send!(JsonEvent::Number(Number::new(*n))),
120        tabnas::Value::String(s) => send!(JsonEvent::String(s)),
121        tabnas::Value::Text(t) => send!(JsonEvent::String(&t.string)),
122        tabnas::Value::Array(items) => {
123            send!(JsonEvent::ArrayStart);
124            for item in items.iter() {
125                if walk_value(item, sink)? == Flow::Stop {
126                    return Ok(Flow::Stop);
127                }
128            }
129            send!(JsonEvent::ArrayEnd);
130        }
131        tabnas::Value::ListRef(list) => {
132            send!(JsonEvent::ArrayStart);
133            for item in list.value.iter() {
134                if walk_value(item, sink)? == Flow::Stop {
135                    return Ok(Flow::Stop);
136                }
137            }
138            send!(JsonEvent::ArrayEnd);
139        }
140        tabnas::Value::Object(members) => {
141            send!(JsonEvent::ObjectStart);
142            for (k, v) in members.iter() {
143                send!(JsonEvent::Key(k));
144                if walk_value(v, sink)? == Flow::Stop {
145                    return Ok(Flow::Stop);
146                }
147            }
148            send!(JsonEvent::ObjectEnd);
149        }
150        tabnas::Value::MapRef(map) => {
151            send!(JsonEvent::ObjectStart);
152            for (k, v) in map.value.iter() {
153                send!(JsonEvent::Key(k));
154                if walk_value(v, sink)? == Flow::Stop {
155                    return Ok(Flow::Stop);
156                }
157            }
158            send!(JsonEvent::ObjectEnd);
159        }
160    }
161    Ok(Flow::Continue)
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167    use crate::event::OwnedJsonEvent;
168
169    #[test]
170    fn a_value_walks_in_document_order() {
171        let v = tabnas_json::parse(r#"{"a":[1,"x"],"b":null}"#).unwrap();
172        let mut rec: Vec<OwnedJsonEvent> = Vec::new();
173        ValueSource(&v).run(&mut rec).unwrap();
174        assert_eq!(
175            rec,
176            vec![
177                OwnedJsonEvent::ObjectStart,
178                OwnedJsonEvent::Key("a".into()),
179                OwnedJsonEvent::ArrayStart,
180                OwnedJsonEvent::Number {
181                    value: 1.0,
182                    lexeme: None
183                },
184                OwnedJsonEvent::String("x".into()),
185                OwnedJsonEvent::ArrayEnd,
186                OwnedJsonEvent::Key("b".into()),
187                OwnedJsonEvent::Null,
188                OwnedJsonEvent::ObjectEnd,
189                OwnedJsonEvent::End,
190            ]
191        );
192    }
193}
tabnas_transduce/source/mod.rs

tabnas_transduce/source/
mod.rs
