links_notation/

stream_parser.rs

//! Incremental parsing for long-lived Links Notation streams.

use crate::quotes::{DelimitedReferences, QUOTES};
use crate::reference_literal::reference_literal_end;
use crate::{
    LiNo, ParseError, ParserConfig, SyntaxReference, parse_lino_to_links_with_config,
    parse_lino_to_syntax_links_with_config,
};
use std::collections::VecDeque;
use std::fmt;

const DEFAULT_MAX_BUFFER_SIZE: usize = 10 * 1024 * 1024;
/// What ends a bare reference, so that a token starts after it.
const SEPARATORS: &[u8] = b" \t\n\r(:)";
const BEFORE_COMMENT: &[u8] = b" \t\n\r";

/// The point immediately after the last character written to a stream parser.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamPosition {
    /// One-based line number.
    pub line: usize,
    /// One-based character column.
    pub column: usize,
    /// Absolute byte offset from the start of the stream.
    pub offset: usize,
    /// Bytes held for the one unresolved top-level record.
    pub buffered: usize,
}

/// Backward-compatible name for stream position information.
pub type ErrorLocation = StreamPosition;

impl fmt::Display for StreamPosition {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            formatter,
            "line {}, column {} (offset {})",
            self.line, self.column, self.offset
        )
    }
}

/// A canonical parser or stream-limit error with absolute stream coordinates.
#[derive(Debug)]
pub struct StreamParseError {
    /// Human-readable description of the error.
    pub message: String,
    /// Position in the complete stream, when available.
    pub location: Option<StreamPosition>,
    /// The canonical parser error, when parsing a completed record failed.
    pub parse_error: Option<Box<ParseError>>,
}

impl fmt::Display for StreamParseError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match &self.location {
            Some(location) => write!(
                formatter,
                "stream parse error at {location}: {}",
                self.message
            ),
            None => write!(formatter, "stream parse error: {}", self.message),
        }
    }
}

impl std::error::Error for StreamParseError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        self.parse_error
            .as_deref()
            .map(|error| error as &(dyn std::error::Error + 'static))
    }
}

impl From<ParseError> for StreamParseError {
    fn from(error: ParseError) -> Self {
        Self {
            message: error.to_string(),
            location: None,
            parse_error: Some(Box::new(error)),
        }
    }
}

/// Callback invoked for every completed link.
pub type LinkCallback<T = String> = Box<dyn FnMut(&LiNo<T>) + Send>;

/// Callback invoked when a write or finish operation fails.
pub type ErrorCallback = Box<dyn FnMut(&StreamParseError) + Send>;

mod sealed {
    pub trait Sealed {}
    impl Sealed for String {}
    impl Sealed for crate::SyntaxReference {}
}

/// The reference type a [`StreamParser`] emits: `String` for the plain links
/// (the AST mode), or [`SyntaxReference`] for links that record how each ID
/// was written. The trait is sealed.
pub trait StreamReference: sealed::Sealed + Clone + Send + 'static {
    /// Parses one completed record.
    #[doc(hidden)]
    fn parse_record(document: &str, config: &ParserConfig) -> Result<Vec<LiNo<Self>>, ParseError>;
}

impl StreamReference for String {
    fn parse_record(document: &str, config: &ParserConfig) -> Result<Vec<LiNo<Self>>, ParseError> {
        parse_lino_to_links_with_config(document, config)
    }
}

impl StreamReference for SyntaxReference {
    fn parse_record(document: &str, config: &ParserConfig) -> Result<Vec<LiNo<Self>>, ParseError> {
        // Each record is parsed on its own, so raw IDs are sliced from the
        // record, which holds the exact text the stream received.
        parse_lino_to_syntax_links_with_config(document, config)
    }
}

/// Incrementally emits complete top-level Links Notation records.
///
/// A record can acquire indented children, so a terminating newline is not by
/// itself a safe boundary. A record is committed when the next non-indented
/// content line begins, and the canonical parser validates every committed
/// segment. Set collection off for bounded-memory callback or iterator use.
///
/// `StreamParser::new()` emits `LiNo<String>`, like
/// [`parse_lino_to_links`](crate::parse_lino_to_links).
/// [`StreamParser::syntax`] emits `LiNo<SyntaxReference>`, like
/// [`parse_lino_to_syntax_links`](crate::parse_lino_to_syntax_links), so every
/// link remembers how its ID was written:
///
/// ```
/// use links_notation::StreamParser;
///
/// let mut stream = StreamParser::syntax();
/// stream.write("(t 1 '1')\n").unwrap();
/// let links = stream.finish().unwrap();
/// assert_eq!(links[0].to_string(), "(t 1 1)");
/// let links_notation::LiNo::Link { values, .. } = &links[0] else { unreachable!() };
/// assert_eq!(values[2].raw_id(), Some("'1'"));
/// assert_eq!(values[2].quote(), Some('\''));
/// ```
pub struct StreamParser<T: StreamReference = String> {
    config: ParserConfig,
    buffer: String,
    current_line: String,
    base_indentation: Option<usize>,
    line_classified: bool,
    link_callback: Option<LinkCallback<T>>,
    error_callback: Option<ErrorCallback>,
    collect: bool,
    max_buffer_size: usize,
    links: Vec<LiNo<T>>,
    offset: usize,
    line: usize,
    column: usize,
    segment_offset: usize,
    segment_line: usize,
    ended: bool,
}

impl Default for StreamParser {
    fn default() -> Self {
        Self::new()
    }
}

impl StreamParser {
    /// Create a parser with the canonical default options.
    pub fn new() -> Self {
        Self::with_config(ParserConfig::default())
    }

    /// Create a parser with explicit canonical parser options.
    pub fn with_config(config: ParserConfig) -> Self {
        Self::create(config)
    }

    /// Lazily parse any iterator of string-like chunks.
    pub fn parse_chunks<I, S>(chunks: I) -> StreamIterator<I::IntoIter>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        StreamIterator::new(chunks.into_iter(), Self::new())
    }
}

impl StreamParser<SyntaxReference> {
    /// Create a parser whose links record how each ID was written.
    pub fn syntax() -> Self {
        Self::syntax_with_config(ParserConfig::default())
    }

    /// Create a parser with explicit options whose links record how each ID
    /// was written.
    pub fn syntax_with_config(config: ParserConfig) -> Self {
        Self::create(config)
    }

    /// Lazily parse any iterator of string-like chunks into links that record
    /// how each ID was written.
    pub fn parse_syntax_chunks<I, S>(chunks: I) -> StreamIterator<I::IntoIter, SyntaxReference>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        StreamIterator::new(chunks.into_iter(), Self::syntax())
    }
}

impl<T: StreamReference> StreamParser<T> {
    fn create(config: ParserConfig) -> Self {
        Self {
            config,
            buffer: String::new(),
            current_line: String::new(),
            base_indentation: None,
            line_classified: false,
            link_callback: None,
            error_callback: None,
            collect: true,
            max_buffer_size: DEFAULT_MAX_BUFFER_SIZE,
            links: Vec::new(),
            offset: 0,
            line: 1,
            column: 1,
            segment_offset: 0,
            segment_line: 1,
            ended: false,
        }
    }

    /// Register a callback invoked once for every completed link.
    pub fn on_link<F>(&mut self, callback: F) -> &mut Self
    where
        F: FnMut(&LiNo<T>) + Send + 'static,
    {
        self.link_callback = Some(Box::new(callback));
        self
    }

    /// Register a callback invoked when parsing fails.
    pub fn on_error<F>(&mut self, callback: F) -> &mut Self
    where
        F: FnMut(&StreamParseError) + Send + 'static,
    {
        self.error_callback = Some(Box::new(callback));
        self
    }

    /// Enable or disable retaining links for [`finish`](Self::finish) and [`drain`](Self::drain).
    pub fn set_collect(&mut self, collect: bool) -> &mut Self {
        self.collect = collect;
        self
    }

    /// Set the largest unresolved record accepted by the stream.
    pub fn set_max_buffer_size(
        &mut self,
        max_buffer_size: usize,
    ) -> Result<&mut Self, StreamParseError> {
        if max_buffer_size == 0 {
            return Err(StreamParseError {
                message: "maximum buffer size must be positive".to_string(),
                location: Some(self.position()),
                parse_error: None,
            });
        }
        self.max_buffer_size = max_buffer_size;
        Ok(self)
    }

    /// Consume a string chunk and return links made complete by it.
    pub fn write(&mut self, chunk: &str) -> Result<Vec<LiNo<T>>, StreamParseError> {
        if self.ended {
            return self.fail(StreamParseError {
                message: "cannot write after finish()".to_string(),
                location: Some(self.position()),
                parse_error: None,
            });
        }

        let mut emitted = Vec::new();
        for character in chunk.chars() {
            self.current_line.push(character);
            self.offset += character.len_utf8();

            if character == '\n' {
                self.buffer.push_str(&self.current_line);
                self.current_line.clear();
                self.line_classified = false;
                self.line += 1;
                self.column = 1;
            } else {
                if !self.line_classified && !matches!(character, ' ' | '\t' | '\r') {
                    self.line_classified = true;
                    if !(self.config.comments && character == '#') {
                        let indentation = leading_spaces(&self.current_line);
                        self.start_content_line(indentation, &mut emitted);
                    }
                }
                self.column += 1;
            }

            if self.buffer.len() + self.current_line.len() > self.max_buffer_size {
                return self.fail(StreamParseError {
                    message: format!(
                        "buffered record exceeds maximum size of {} bytes",
                        self.max_buffer_size
                    ),
                    location: Some(self.position()),
                    parse_error: None,
                });
            }
        }
        Ok(emitted)
    }

    /// Finish the stream and return all undrained or newly emitted links.
    pub fn finish(&mut self) -> Result<Vec<LiNo<T>>, StreamParseError> {
        if self.ended {
            return Ok(if self.collect {
                self.links.clone()
            } else {
                Vec::new()
            });
        }

        let mut emitted = Vec::new();
        let document = format!("{}{}", self.buffer, self.current_line);
        if !document.is_empty() {
            let parsed = match T::parse_record(&document, &self.config) {
                Ok(links) => links,
                Err(error) => {
                    let stream_error = self.stream_error(error);
                    return self.fail(stream_error);
                }
            };
            self.publish(parsed, &mut emitted);
            self.advance_segment(&document);
        }

        self.buffer.clear();
        self.current_line.clear();
        self.base_indentation = None;
        self.ended = true;
        Ok(if self.collect {
            self.links.clone()
        } else {
            emitted
        })
    }

    /// Return and forget retained links.
    pub fn drain(&mut self) -> Vec<LiNo<T>> {
        std::mem::take(&mut self.links)
    }

    /// Reuse this parser while preserving its configuration and callbacks.
    pub fn reset(&mut self) -> &mut Self {
        self.buffer.clear();
        self.current_line.clear();
        self.base_indentation = None;
        self.line_classified = false;
        self.links.clear();
        self.offset = 0;
        self.line = 1;
        self.column = 1;
        self.segment_offset = 0;
        self.segment_line = 1;
        self.ended = false;
        self
    }

    /// Return the absolute stream position and unresolved buffer size.
    pub fn position(&self) -> StreamPosition {
        StreamPosition {
            line: self.line,
            column: self.column,
            offset: self.offset,
            buffered: self.buffer.len() + self.current_line.len(),
        }
    }

    fn start_content_line(&mut self, indentation: usize, emitted: &mut Vec<LiNo<T>>) {
        if !self.buffer.is_empty()
            && self
                .base_indentation
                .is_some_and(|base| indentation <= base)
            && structurally_complete(&self.buffer, self.config.comments)
        {
            if let Ok(parsed) = T::parse_record(&self.buffer, &self.config) {
                let document = std::mem::take(&mut self.buffer);
                self.publish(parsed, emitted);
                self.advance_segment(&document);
                self.base_indentation = None;
            }
        }
        self.base_indentation.get_or_insert(indentation);
    }

    fn publish(&mut self, parsed: Vec<LiNo<T>>, emitted: &mut Vec<LiNo<T>>) {
        for link in parsed {
            emitted.push(link.clone());
            if self.collect {
                self.links.push(link.clone());
            }
            if let Some(callback) = &mut self.link_callback {
                callback(&link);
            }
        }
    }

    fn advance_segment(&mut self, document: &str) {
        self.segment_offset += document.len();
        self.segment_line += document.matches('\n').count();
    }

    fn stream_error(&self, error: ParseError) -> StreamParseError {
        let location = match &error {
            ParseError::SyntaxError(syntax) => StreamPosition {
                line: self.segment_line + syntax.line - 1,
                column: syntax.column,
                offset: self.segment_offset + syntax.offset,
                buffered: self.buffer.len() + self.current_line.len(),
            },
            ParseError::NestingTooDeep(nesting) => StreamPosition {
                line: self.segment_line + nesting.line - 1,
                column: nesting.column,
                offset: self.segment_offset + nesting.offset,
                buffered: self.buffer.len() + self.current_line.len(),
            },
            _ => StreamPosition {
                line: self.segment_line,
                column: 1,
                offset: self.segment_offset,
                buffered: self.buffer.len() + self.current_line.len(),
            },
        };
        StreamParseError {
            message: error.to_string(),
            location: Some(location),
            parse_error: Some(Box::new(error)),
        }
    }

    fn fail<R>(&mut self, error: StreamParseError) -> Result<R, StreamParseError> {
        if let Some(callback) = &mut self.error_callback {
            callback(&error);
        }
        Err(error)
    }
}

/// Lazy iterator returned by [`StreamParser::parse_chunks`] and
/// [`StreamParser::parse_syntax_chunks`].
pub struct StreamIterator<I, T: StreamReference = String> {
    chunks: I,
    parser: StreamParser<T>,
    ready: VecDeque<LiNo<T>>,
    finished: bool,
}

impl<I, T: StreamReference> StreamIterator<I, T> {
    fn new(chunks: I, mut parser: StreamParser<T>) -> Self {
        parser.set_collect(false);
        Self {
            chunks,
            parser,
            ready: VecDeque::new(),
            finished: false,
        }
    }
}

impl<I, S, T> Iterator for StreamIterator<I, T>
where
    I: Iterator<Item = S>,
    S: AsRef<str>,
    T: StreamReference,
{
    type Item = Result<LiNo<T>, StreamParseError>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if let Some(link) = self.ready.pop_front() {
                return Some(Ok(link));
            }
            if self.finished {
                return None;
            }
            match self.chunks.next() {
                Some(chunk) => match self.parser.write(chunk.as_ref()) {
                    Ok(links) => self.ready.extend(links),
                    Err(error) => {
                        self.finished = true;
                        return Some(Err(error));
                    }
                },
                None => {
                    self.finished = true;
                    match self.parser.finish() {
                        Ok(links) => self.ready.extend(links),
                        Err(error) => return Some(Err(error)),
                    }
                }
            }
        }
    }
}

fn leading_spaces(line: &str) -> usize {
    line.bytes().take_while(|byte| *byte == b' ').count()
}

fn follows(document: &[u8], position: usize, allowed: &[u8]) -> bool {
    position == 0 || allowed.contains(&document[position - 1])
}

fn structurally_complete(document: &str, comments: bool) -> bool {
    // Tokens are read as the comment reader and the parser read them: a
    // reference may start after a separator, and right after a delimited
    // reference or a literal (issue #377).
    let bytes = document.as_bytes();
    let references = DelimitedReferences::new(document);
    let mut position = 0;
    let mut depth = 0_isize;
    let mut token_start = true;

    while position < bytes.len() {
        let byte = bytes[position];
        if comments && byte == b'#' && follows(bytes, position, BEFORE_COMMENT) {
            match document[position..].find('\n') {
                Some(newline) => position += newline,
                None => break,
            }
            continue;
        }
        if SEPARATORS.contains(&byte) {
            match byte {
                b'(' => depth += 1,
                b')' => depth -= 1,
                _ => {}
            }
            token_start = true;
            position += 1;
            continue;
        }
        if token_start && QUOTES.contains(&byte) {
            // A reference nothing closes yet may be closed by a later line,
            // which changes what it holds: an even run is the empty reference
            // only when no run in the whole document closes it.
            match references.read(&document[position..]) {
                Some(reading) if reading.closed => position += reading.length,
                _ => return false,
            }
            continue;
        }
        if token_start {
            if let Some(end) = reference_literal_end(document, position) {
                position = end;
                continue;
            }
        }
        token_start = false;
        position += 1;
    }
    depth == 0
}