links_notation/
stream_parser.rs

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

links_notation/
stream_parser.rs
