links_notation/
stream_parser.rs

1//! Incremental parsing for long-lived Links Notation streams.
2
3use crate::quotes::DelimitedReferences;
4use crate::{
5    LiNo, ParseError, ParserConfig, SyntaxReference, parse_lino_to_links_with_config,
6    parse_lino_to_syntax_links_with_config,
7};
8use std::collections::VecDeque;
9use std::fmt;
10
11const DEFAULT_MAX_BUFFER_SIZE: usize = 10 * 1024 * 1024;
12const BEFORE_REFERENCE: &[u8] = b" \t\n\r(:";
13const BEFORE_COMMENT: &[u8] = b" \t\n\r";
14
15/// The point immediately after the last character written to a stream parser.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct StreamPosition {
18    /// One-based line number.
19    pub line: usize,
20    /// One-based character column.
21    pub column: usize,
22    /// Absolute byte offset from the start of the stream.
23    pub offset: usize,
24    /// Bytes held for the one unresolved top-level record.
25    pub buffered: usize,
26}
27
28/// Backward-compatible name for stream position information.
29pub type ErrorLocation = StreamPosition;
30
31impl fmt::Display for StreamPosition {
32    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
33        write!(
34            formatter,
35            "line {}, column {} (offset {})",
36            self.line, self.column, self.offset
37        )
38    }
39}
40
41/// A canonical parser or stream-limit error with absolute stream coordinates.
42#[derive(Debug)]
43pub struct StreamParseError {
44    /// Human-readable description of the error.
45    pub message: String,
46    /// Position in the complete stream, when available.
47    pub location: Option<StreamPosition>,
48    /// The canonical parser error, when parsing a completed record failed.
49    pub parse_error: Option<Box<ParseError>>,
50}
51
52impl fmt::Display for StreamParseError {
53    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
54        match &self.location {
55            Some(location) => write!(
56                formatter,
57                "stream parse error at {location}: {}",
58                self.message
59            ),
60            None => write!(formatter, "stream parse error: {}", self.message),
61        }
62    }
63}
64
65impl std::error::Error for StreamParseError {
66    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
67        self.parse_error
68            .as_deref()
69            .map(|error| error as &(dyn std::error::Error + 'static))
70    }
71}
72
73impl From<ParseError> for StreamParseError {
74    fn from(error: ParseError) -> Self {
75        Self {
76            message: error.to_string(),
77            location: None,
78            parse_error: Some(Box::new(error)),
79        }
80    }
81}
82
83/// Callback invoked for every completed link.
84pub type LinkCallback<T = String> = Box<dyn FnMut(&LiNo<T>) + Send>;
85
86/// Callback invoked when a write or finish operation fails.
87pub type ErrorCallback = Box<dyn FnMut(&StreamParseError) + Send>;
88
89mod sealed {
90    pub trait Sealed {}
91    impl Sealed for String {}
92    impl Sealed for crate::SyntaxReference {}
93}
94
95/// The reference type a [`StreamParser`] emits: `String` for the plain links
96/// (the AST mode), or [`SyntaxReference`] for links that record how each ID
97/// was written. The trait is sealed.
98pub trait StreamReference: sealed::Sealed + Clone + Send + 'static {
99    /// Parses one completed record.
100    #[doc(hidden)]
101    fn parse_record(document: &str, config: &ParserConfig) -> Result<Vec<LiNo<Self>>, ParseError>;
102}
103
104impl StreamReference for String {
105    fn parse_record(document: &str, config: &ParserConfig) -> Result<Vec<LiNo<Self>>, ParseError> {
106        parse_lino_to_links_with_config(document, config)
107    }
108}
109
110impl StreamReference for SyntaxReference {
111    fn parse_record(document: &str, config: &ParserConfig) -> Result<Vec<LiNo<Self>>, ParseError> {
112        // Each record is parsed on its own, so raw IDs are sliced from the
113        // record, which holds the exact text the stream received.
114        parse_lino_to_syntax_links_with_config(document, config)
115    }
116}
117
118/// Incrementally emits complete top-level Links Notation records.
119///
120/// A record can acquire indented children, so a terminating newline is not by
121/// itself a safe boundary. A record is committed when the next non-indented
122/// content line begins, and the canonical parser validates every committed
123/// segment. Set collection off for bounded-memory callback or iterator use.
124///
125/// `StreamParser::new()` emits `LiNo<String>`, like
126/// [`parse_lino_to_links`](crate::parse_lino_to_links).
127/// [`StreamParser::syntax`] emits `LiNo<SyntaxReference>`, like
128/// [`parse_lino_to_syntax_links`](crate::parse_lino_to_syntax_links), so every
129/// link remembers how its ID was written:
130///
131/// ```
132/// use links_notation::StreamParser;
133///
134/// let mut stream = StreamParser::syntax();
135/// stream.write("(t 1 '1')\n").unwrap();
136/// let links = stream.finish().unwrap();
137/// assert_eq!(links[0].to_string(), "(t 1 1)");
138/// let links_notation::LiNo::Link { values, .. } = &links[0] else { unreachable!() };
139/// assert_eq!(values[2].raw_id(), Some("'1'"));
140/// assert_eq!(values[2].quote(), Some('\''));
141/// ```
142pub struct StreamParser<T: StreamReference = String> {
143    config: ParserConfig,
144    buffer: String,
145    current_line: String,
146    base_indentation: Option<usize>,
147    line_classified: bool,
148    link_callback: Option<LinkCallback<T>>,
149    error_callback: Option<ErrorCallback>,
150    collect: bool,
151    max_buffer_size: usize,
152    links: Vec<LiNo<T>>,
153    offset: usize,
154    line: usize,
155    column: usize,
156    segment_offset: usize,
157    segment_line: usize,
158    ended: bool,
159}
160
161impl Default for StreamParser {
162    fn default() -> Self {
163        Self::new()
164    }
165}
166
167impl StreamParser {
168    /// Create a parser with the canonical default options.
169    pub fn new() -> Self {
170        Self::with_config(ParserConfig::default())
171    }
172
173    /// Create a parser with explicit canonical parser options.
174    pub fn with_config(config: ParserConfig) -> Self {
175        Self::create(config)
176    }
177
178    /// Lazily parse any iterator of string-like chunks.
179    pub fn parse_chunks<I, S>(chunks: I) -> StreamIterator<I::IntoIter>
180    where
181        I: IntoIterator<Item = S>,
182        S: AsRef<str>,
183    {
184        StreamIterator::new(chunks.into_iter(), Self::new())
185    }
186}
187
188impl StreamParser<SyntaxReference> {
189    /// Create a parser whose links record how each ID was written.
190    pub fn syntax() -> Self {
191        Self::syntax_with_config(ParserConfig::default())
192    }
193
194    /// Create a parser with explicit options whose links record how each ID
195    /// was written.
196    pub fn syntax_with_config(config: ParserConfig) -> Self {
197        Self::create(config)
198    }
199
200    /// Lazily parse any iterator of string-like chunks into links that record
201    /// how each ID was written.
202    pub fn parse_syntax_chunks<I, S>(chunks: I) -> StreamIterator<I::IntoIter, SyntaxReference>
203    where
204        I: IntoIterator<Item = S>,
205        S: AsRef<str>,
206    {
207        StreamIterator::new(chunks.into_iter(), Self::syntax())
208    }
209}
210
211impl<T: StreamReference> StreamParser<T> {
212    fn create(config: ParserConfig) -> Self {
213        Self {
214            config,
215            buffer: String::new(),
216            current_line: String::new(),
217            base_indentation: None,
218            line_classified: false,
219            link_callback: None,
220            error_callback: None,
221            collect: true,
222            max_buffer_size: DEFAULT_MAX_BUFFER_SIZE,
223            links: Vec::new(),
224            offset: 0,
225            line: 1,
226            column: 1,
227            segment_offset: 0,
228            segment_line: 1,
229            ended: false,
230        }
231    }
232
233    /// Register a callback invoked once for every completed link.
234    pub fn on_link<F>(&mut self, callback: F) -> &mut Self
235    where
236        F: FnMut(&LiNo<T>) + Send + 'static,
237    {
238        self.link_callback = Some(Box::new(callback));
239        self
240    }
241
242    /// Register a callback invoked when parsing fails.
243    pub fn on_error<F>(&mut self, callback: F) -> &mut Self
244    where
245        F: FnMut(&StreamParseError) + Send + 'static,
246    {
247        self.error_callback = Some(Box::new(callback));
248        self
249    }
250
251    /// Enable or disable retaining links for [`finish`](Self::finish) and [`drain`](Self::drain).
252    pub fn set_collect(&mut self, collect: bool) -> &mut Self {
253        self.collect = collect;
254        self
255    }
256
257    /// Set the largest unresolved record accepted by the stream.
258    pub fn set_max_buffer_size(
259        &mut self,
260        max_buffer_size: usize,
261    ) -> Result<&mut Self, StreamParseError> {
262        if max_buffer_size == 0 {
263            return Err(StreamParseError {
264                message: "maximum buffer size must be positive".to_string(),
265                location: Some(self.position()),
266                parse_error: None,
267            });
268        }
269        self.max_buffer_size = max_buffer_size;
270        Ok(self)
271    }
272
273    /// Consume a string chunk and return links made complete by it.
274    pub fn write(&mut self, chunk: &str) -> Result<Vec<LiNo<T>>, StreamParseError> {
275        if self.ended {
276            return self.fail(StreamParseError {
277                message: "cannot write after finish()".to_string(),
278                location: Some(self.position()),
279                parse_error: None,
280            });
281        }
282
283        let mut emitted = Vec::new();
284        for character in chunk.chars() {
285            self.current_line.push(character);
286            self.offset += character.len_utf8();
287
288            if character == '\n' {
289                self.buffer.push_str(&self.current_line);
290                self.current_line.clear();
291                self.line_classified = false;
292                self.line += 1;
293                self.column = 1;
294            } else {
295                if !self.line_classified && !matches!(character, ' ' | '\t' | '\r') {
296                    self.line_classified = true;
297                    if !(self.config.comments && character == '#') {
298                        let indentation = leading_spaces(&self.current_line);
299                        self.start_content_line(indentation, &mut emitted);
300                    }
301                }
302                self.column += 1;
303            }
304
305            if self.buffer.len() + self.current_line.len() > self.max_buffer_size {
306                return self.fail(StreamParseError {
307                    message: format!(
308                        "buffered record exceeds maximum size of {} bytes",
309                        self.max_buffer_size
310                    ),
311                    location: Some(self.position()),
312                    parse_error: None,
313                });
314            }
315        }
316        Ok(emitted)
317    }
318
319    /// Finish the stream and return all undrained or newly emitted links.
320    pub fn finish(&mut self) -> Result<Vec<LiNo<T>>, StreamParseError> {
321        if self.ended {
322            return Ok(if self.collect {
323                self.links.clone()
324            } else {
325                Vec::new()
326            });
327        }
328
329        let mut emitted = Vec::new();
330        let document = format!("{}{}", self.buffer, self.current_line);
331        if !document.is_empty() {
332            let parsed = match T::parse_record(&document, &self.config) {
333                Ok(links) => links,
334                Err(error) => {
335                    let stream_error = self.stream_error(error);
336                    return self.fail(stream_error);
337                }
338            };
339            self.publish(parsed, &mut emitted);
340            self.advance_segment(&document);
341        }
342
343        self.buffer.clear();
344        self.current_line.clear();
345        self.base_indentation = None;
346        self.ended = true;
347        Ok(if self.collect {
348            self.links.clone()
349        } else {
350            emitted
351        })
352    }
353
354    /// Return and forget retained links.
355    pub fn drain(&mut self) -> Vec<LiNo<T>> {
356        std::mem::take(&mut self.links)
357    }
358
359    /// Reuse this parser while preserving its configuration and callbacks.
360    pub fn reset(&mut self) -> &mut Self {
361        self.buffer.clear();
362        self.current_line.clear();
363        self.base_indentation = None;
364        self.line_classified = false;
365        self.links.clear();
366        self.offset = 0;
367        self.line = 1;
368        self.column = 1;
369        self.segment_offset = 0;
370        self.segment_line = 1;
371        self.ended = false;
372        self
373    }
374
375    /// Return the absolute stream position and unresolved buffer size.
376    pub fn position(&self) -> StreamPosition {
377        StreamPosition {
378            line: self.line,
379            column: self.column,
380            offset: self.offset,
381            buffered: self.buffer.len() + self.current_line.len(),
382        }
383    }
384
385    fn start_content_line(&mut self, indentation: usize, emitted: &mut Vec<LiNo<T>>) {
386        if !self.buffer.is_empty()
387            && self
388                .base_indentation
389                .is_some_and(|base| indentation <= base)
390            && structurally_complete(&self.buffer, self.config.comments)
391        {
392            if let Ok(parsed) = T::parse_record(&self.buffer, &self.config) {
393                let document = std::mem::take(&mut self.buffer);
394                self.publish(parsed, emitted);
395                self.advance_segment(&document);
396                self.base_indentation = None;
397            }
398        }
399        self.base_indentation.get_or_insert(indentation);
400    }
401
402    fn publish(&mut self, parsed: Vec<LiNo<T>>, emitted: &mut Vec<LiNo<T>>) {
403        for link in parsed {
404            emitted.push(link.clone());
405            if self.collect {
406                self.links.push(link.clone());
407            }
408            if let Some(callback) = &mut self.link_callback {
409                callback(&link);
410            }
411        }
412    }
413
414    fn advance_segment(&mut self, document: &str) {
415        self.segment_offset += document.len();
416        self.segment_line += document.matches('\n').count();
417    }
418
419    fn stream_error(&self, error: ParseError) -> StreamParseError {
420        let location = match &error {
421            ParseError::SyntaxError(syntax) => StreamPosition {
422                line: self.segment_line + syntax.line - 1,
423                column: syntax.column,
424                offset: self.segment_offset + syntax.offset,
425                buffered: self.buffer.len() + self.current_line.len(),
426            },
427            ParseError::NestingTooDeep(nesting) => StreamPosition {
428                line: self.segment_line + nesting.line - 1,
429                column: nesting.column,
430                offset: self.segment_offset + nesting.offset,
431                buffered: self.buffer.len() + self.current_line.len(),
432            },
433            _ => StreamPosition {
434                line: self.segment_line,
435                column: 1,
436                offset: self.segment_offset,
437                buffered: self.buffer.len() + self.current_line.len(),
438            },
439        };
440        StreamParseError {
441            message: error.to_string(),
442            location: Some(location),
443            parse_error: Some(Box::new(error)),
444        }
445    }
446
447    fn fail<R>(&mut self, error: StreamParseError) -> Result<R, StreamParseError> {
448        if let Some(callback) = &mut self.error_callback {
449            callback(&error);
450        }
451        Err(error)
452    }
453}
454
455/// Lazy iterator returned by [`StreamParser::parse_chunks`] and
456/// [`StreamParser::parse_syntax_chunks`].
457pub struct StreamIterator<I, T: StreamReference = String> {
458    chunks: I,
459    parser: StreamParser<T>,
460    ready: VecDeque<LiNo<T>>,
461    finished: bool,
462}
463
464impl<I, T: StreamReference> StreamIterator<I, T> {
465    fn new(chunks: I, mut parser: StreamParser<T>) -> Self {
466        parser.set_collect(false);
467        Self {
468            chunks,
469            parser,
470            ready: VecDeque::new(),
471            finished: false,
472        }
473    }
474}
475
476impl<I, S, T> Iterator for StreamIterator<I, T>
477where
478    I: Iterator<Item = S>,
479    S: AsRef<str>,
480    T: StreamReference,
481{
482    type Item = Result<LiNo<T>, StreamParseError>;
483
484    fn next(&mut self) -> Option<Self::Item> {
485        loop {
486            if let Some(link) = self.ready.pop_front() {
487                return Some(Ok(link));
488            }
489            if self.finished {
490                return None;
491            }
492            match self.chunks.next() {
493                Some(chunk) => match self.parser.write(chunk.as_ref()) {
494                    Ok(links) => self.ready.extend(links),
495                    Err(error) => {
496                        self.finished = true;
497                        return Some(Err(error));
498                    }
499                },
500                None => {
501                    self.finished = true;
502                    match self.parser.finish() {
503                        Ok(links) => self.ready.extend(links),
504                        Err(error) => return Some(Err(error)),
505                    }
506                }
507            }
508        }
509    }
510}
511
512fn leading_spaces(line: &str) -> usize {
513    line.bytes().take_while(|byte| *byte == b' ').count()
514}
515
516fn follows(document: &[u8], position: usize, allowed: &[u8]) -> bool {
517    position == 0 || allowed.contains(&document[position - 1])
518}
519
520fn structurally_complete(document: &str, comments: bool) -> bool {
521    let bytes = document.as_bytes();
522    let references = DelimitedReferences::new(document);
523    let mut position = 0;
524    let mut depth = 0_isize;
525
526    while position < bytes.len() {
527        let character = document[position..]
528            .chars()
529            .next()
530            .expect("position is a character boundary");
531        if matches!(character, '"' | '\'' | '`') && follows(bytes, position, BEFORE_REFERENCE) {
532            let Some(end) = references.end_at(document, position) else {
533                return false;
534            };
535            position = end;
536            continue;
537        }
538        if comments && character == '#' && follows(bytes, position, BEFORE_COMMENT) {
539            match document[position..].find('\n') {
540                Some(newline) => position += newline + 1,
541                None => break,
542            }
543            continue;
544        }
545        if character == '(' {
546            depth += 1;
547        } else if character == ')' {
548            depth -= 1;
549        }
550        position += character.len_utf8();
551    }
552    depth == 0
553}
links_notation/stream_parser.rs

links_notation/
stream_parser.rs
