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;
const SEPARATORS: &[u8] = b" \t\n\r(:)";
const BEFORE_COMMENT: &[u8] = b" \t\n\r";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamPosition {
pub line: usize,
pub column: usize,
pub offset: usize,
pub buffered: usize,
}
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
)
}
}
#[derive(Debug)]
pub struct StreamParseError {
pub message: String,
pub location: Option<StreamPosition>,
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)),
}
}
}
pub type LinkCallback<T = String> = Box<dyn FnMut(&LiNo<T>) + Send>;
pub type ErrorCallback = Box<dyn FnMut(&StreamParseError) + Send>;
mod sealed {
pub trait Sealed {}
impl Sealed for String {}
impl Sealed for crate::SyntaxReference {}
}
pub trait StreamReference: sealed::Sealed + Clone + Send + 'static {
#[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> {
parse_lino_to_syntax_links_with_config(document, config)
}
}
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 {
pub fn new() -> Self {
Self::with_config(ParserConfig::default())
}
pub fn with_config(config: ParserConfig) -> Self {
Self::create(config)
}
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> {
pub fn syntax() -> Self {
Self::syntax_with_config(ParserConfig::default())
}
pub fn syntax_with_config(config: ParserConfig) -> Self {
Self::create(config)
}
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,
}
}
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
}
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
}
pub fn set_collect(&mut self, collect: bool) -> &mut Self {
self.collect = collect;
self
}
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)
}
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)
}
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
})
}
pub fn drain(&mut self) -> Vec<LiNo<T>> {
std::mem::take(&mut self.links)
}
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
}
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)
}
}
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 {
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) {
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
}