rust_yaml/

streaming_enhanced.rs

//! Enhanced streaming YAML parser with true incremental parsing
//!
//! This module provides advanced streaming capabilities including:
//! - Incremental parsing with partial document support
//! - Async/await support for I/O operations
//! - Memory-mapped file support for large documents
//! - Buffered reading with configurable chunk sizes

use crate::{
    parser::{Event, EventType},
    Error, Limits, Position, ResourceTracker, Result,
};
use std::collections::VecDeque;
use std::io::{BufRead, BufReader};
use std::path::Path;

/// Configuration for enhanced streaming parser
#[derive(Debug, Clone)]
pub struct StreamConfig {
    /// Size of the read buffer in bytes
    pub buffer_size: usize,
    /// Maximum number of events to buffer
    pub max_event_buffer: usize,
    /// Enable incremental parsing (parse partial documents)
    pub incremental: bool,
    /// Resource limits
    pub limits: Limits,
    /// Chunk size for reading (bytes)
    pub chunk_size: usize,
}

impl Default for StreamConfig {
    fn default() -> Self {
        Self {
            buffer_size: 64 * 1024, // 64KB buffer
            max_event_buffer: 1000,
            incremental: true,
            limits: Limits::default(),
            chunk_size: 8 * 1024, // 8KB chunks
        }
    }
}

impl StreamConfig {
    /// Create config for large files
    pub fn large_file() -> Self {
        Self {
            buffer_size: 1024 * 1024, // 1MB buffer
            max_event_buffer: 10000,
            incremental: true,
            limits: Limits::permissive(),
            chunk_size: 64 * 1024, // 64KB chunks
        }
    }

    /// Create config for memory-constrained environments
    pub fn low_memory() -> Self {
        Self {
            buffer_size: 8 * 1024, // 8KB buffer
            max_event_buffer: 100,
            incremental: true,
            limits: Limits::strict(),
            chunk_size: 1024, // 1KB chunks
        }
    }
}

/// State of the streaming parser
#[derive(Debug, Clone, PartialEq)]
enum StreamState {
    /// Initial state
    Initial,
    /// Reading document
    InDocument,
    /// Between documents
    BetweenDocuments,
    /// End of stream
    EndOfStream,
    /// Error state
    Error(String),
}

/// Enhanced streaming YAML parser
pub struct StreamingYamlParser<R: BufRead> {
    /// Input reader
    reader: R,
    /// Configuration
    config: StreamConfig,
    /// Current parsing state
    state: StreamState,
    /// Buffer for incomplete data
    buffer: String,
    /// Event queue
    events: VecDeque<Event>,
    /// Current position in the stream
    position: Position,
    /// Resource tracker
    resource_tracker: ResourceTracker,
    /// Parse context for incremental parsing
    context: ParseContext,
    /// Statistics
    stats: StreamStats,
}

/// Parsing context for incremental parsing
#[derive(Debug, Clone)]
struct ParseContext {
    /// Stack of collection types (true = mapping, false = sequence)
    collection_stack: Vec<bool>,
    /// Current indentation level
    indent_level: usize,
    /// Pending anchor
    pending_anchor: Option<String>,
    /// Pending tag
    pending_tag: Option<String>,
    /// In block scalar
    in_block_scalar: bool,
    /// Block scalar indent
    block_scalar_indent: Option<usize>,
}

impl ParseContext {
    fn new() -> Self {
        Self {
            collection_stack: Vec::new(),
            indent_level: 0,
            pending_anchor: None,
            pending_tag: None,
            in_block_scalar: false,
            block_scalar_indent: None,
        }
    }

    fn reset(&mut self) {
        self.collection_stack.clear();
        self.indent_level = 0;
        self.pending_anchor = None;
        self.pending_tag = None;
        self.in_block_scalar = false;
        self.block_scalar_indent = None;
    }
}

/// Statistics for streaming parser
#[derive(Debug, Clone, Default)]
pub struct StreamStats {
    /// Total bytes read
    pub bytes_read: usize,
    /// Total events generated
    pub events_generated: usize,
    /// Documents parsed
    pub documents_parsed: usize,
    /// Parse errors encountered
    pub errors_encountered: usize,
    /// Maximum buffer size used
    pub max_buffer_size: usize,
    /// Total parse time (milliseconds)
    pub parse_time_ms: u64,
}

impl<R: BufRead> StreamingYamlParser<R> {
    /// Create a new streaming parser from a reader
    pub fn new(reader: R, config: StreamConfig) -> Self {
        Self {
            reader,
            config,
            state: StreamState::Initial,
            buffer: String::with_capacity(4096),
            events: VecDeque::with_capacity(100),
            position: Position::new(),
            resource_tracker: ResourceTracker::new(),
            context: ParseContext::new(),
            stats: StreamStats::default(),
        }
    }

    /// Parse the next chunk of data
    pub fn parse_next(&mut self) -> Result<bool> {
        let start = std::time::Instant::now();

        // Read next chunk
        let bytes_read = self.read_chunk()?;
        if bytes_read == 0 && self.buffer.is_empty() {
            self.state = StreamState::EndOfStream;
            return Ok(false);
        }

        self.stats.bytes_read += bytes_read;

        // Parse buffer content
        self.parse_buffer()?;

        // Update statistics
        self.stats.parse_time_ms += start.elapsed().as_millis() as u64;
        self.stats.max_buffer_size = self.stats.max_buffer_size.max(self.buffer.len());

        Ok(!self.events.is_empty())
    }

    /// Read a chunk of data from the reader
    fn read_chunk(&mut self) -> Result<usize> {
        let mut temp_buffer = vec![0u8; self.config.chunk_size];
        let bytes_read = self.reader.read(&mut temp_buffer)?;

        if bytes_read > 0 {
            let chunk = String::from_utf8_lossy(&temp_buffer[..bytes_read]);
            self.buffer.push_str(&chunk);
        }

        Ok(bytes_read)
    }

    /// Parse the current buffer content
    fn parse_buffer(&mut self) -> Result<()> {
        // Handle different states
        match self.state {
            StreamState::Initial => {
                self.emit_stream_start()?;
                self.state = StreamState::BetweenDocuments;
            }
            StreamState::BetweenDocuments => {
                self.parse_document_start()?;
            }
            StreamState::InDocument => {
                self.parse_document_content()?;
            }
            StreamState::EndOfStream => {
                return Ok(());
            }
            StreamState::Error(ref msg) => {
                return Err(Error::parse(self.position, msg.clone()));
            }
        }

        Ok(())
    }

    /// Parse document start markers
    fn parse_document_start(&mut self) -> Result<()> {
        // Skip whitespace
        self.skip_whitespace();

        // Check for document start marker (---)
        if self.buffer.starts_with("---") {
            self.buffer.drain(..3);
            self.position.column += 3;
            self.emit_document_start()?;
            self.state = StreamState::InDocument;
        } else if !self.buffer.is_empty() {
            // Implicit document start
            self.emit_document_start()?;
            self.state = StreamState::InDocument;
            self.parse_document_content()?;
        }

        Ok(())
    }

    /// Parse document content incrementally
    fn parse_document_content(&mut self) -> Result<()> {
        while !self.buffer.is_empty() {
            // Check for document end marker (...)
            if self.buffer.starts_with("...") {
                self.buffer.drain(..3);
                self.position.column += 3;
                self.emit_document_end()?;
                self.state = StreamState::BetweenDocuments;
                self.context.reset();
                break;
            }

            // Parse based on context
            if self.context.in_block_scalar {
                self.parse_block_scalar_content()?;
            } else {
                self.parse_yaml_content()?;
            }

            // Break if we need more data
            if self.needs_more_data() {
                break;
            }
        }

        Ok(())
    }

    /// Parse YAML content (scalars, collections, etc.)
    fn parse_yaml_content(&mut self) -> Result<()> {
        self.skip_whitespace();

        if self.buffer.is_empty() {
            return Ok(());
        }

        let first_char = self.buffer.chars().next().unwrap();

        match first_char {
            '-' if self.is_sequence_item() => {
                self.parse_sequence_item()?;
            }
            '[' => {
                self.parse_flow_sequence()?;
            }
            '{' => {
                self.parse_flow_mapping()?;
            }
            '|' | '>' => {
                self.parse_block_scalar_start(first_char)?;
            }
            '&' => {
                self.parse_anchor()?;
            }
            '*' => {
                self.parse_alias()?;
            }
            '"' | '\'' => {
                self.parse_quoted_scalar(first_char)?;
            }
            '#' => {
                self.skip_comment();
            }
            '\n' => {
                self.buffer.remove(0);
                self.position.line += 1;
                self.position.column = 0;
            }
            _ if self.is_mapping_key() => {
                self.parse_mapping_entry()?;
            }
            _ => {
                self.parse_plain_scalar()?;
            }
        }

        Ok(())
    }

    /// Check if we need more data to continue parsing
    fn needs_more_data(&self) -> bool {
        // If buffer is small and doesn't contain a complete line
        if self.buffer.len() < 100 && !self.buffer.contains('\n') {
            return true;
        }

        // If we're in a block scalar and need more lines
        if self.context.in_block_scalar && !self.has_complete_block_scalar() {
            return true;
        }

        false
    }

    /// Check if buffer contains a complete block scalar
    fn has_complete_block_scalar(&self) -> bool {
        // Simplified check - in production, would be more sophisticated
        self.buffer.contains("\n\n") || self.buffer.contains("\n...")
    }

    /// Parse a sequence item
    fn parse_sequence_item(&mut self) -> Result<()> {
        self.buffer.remove(0); // Remove '-'
        self.position.column += 1;

        // Start sequence if needed
        if !self.context.collection_stack.iter().any(|&x| !x) {
            self.emit_sequence_start()?;
            self.context.collection_stack.push(false);
        }

        self.skip_whitespace();
        Ok(())
    }

    /// Parse a mapping entry
    fn parse_mapping_entry(&mut self) -> Result<()> {
        // Parse key
        let key_end = self.find_mapping_key_end();
        if let Some(end) = key_end {
            let key = self.buffer.drain(..end).collect::<String>();
            self.position.column += key.len();

            // Skip ':' and whitespace
            if self.buffer.starts_with(':') {
                self.buffer.remove(0);
                self.position.column += 1;
            }

            // Start mapping if needed
            if !self.context.collection_stack.iter().any(|&x| x) {
                self.emit_mapping_start()?;
                self.context.collection_stack.push(true);
            }

            // Emit key as scalar
            self.emit_scalar(key.trim().to_string())?;
        }

        Ok(())
    }

    /// Helper methods for parsing
    fn skip_whitespace(&mut self) {
        while let Some(ch) = self.buffer.chars().next() {
            if ch == ' ' || ch == '\t' {
                self.buffer.remove(0);
                self.position.column += 1;
            } else {
                break;
            }
        }
    }

    fn skip_comment(&mut self) {
        if let Some(newline_pos) = self.buffer.find('\n') {
            self.buffer.drain(..newline_pos);
            self.position.column = 0;
        } else {
            self.buffer.clear();
        }
    }

    fn is_sequence_item(&self) -> bool {
        self.buffer.starts_with("- ")
    }

    fn is_mapping_key(&self) -> bool {
        // Simplified check for mapping key
        self.buffer.contains(':') && !self.buffer.starts_with(':')
    }

    fn find_mapping_key_end(&self) -> Option<usize> {
        self.buffer.find(':')
    }

    fn parse_flow_sequence(&mut self) -> Result<()> {
        // Simplified flow sequence parsing
        if let Some(end) = self.buffer.find(']') {
            let content = self.buffer.drain(..=end).collect::<String>();
            self.emit_sequence_start()?;
            // Parse content (simplified)
            self.emit_sequence_end()?;
            self.position.column += content.len();
        }
        Ok(())
    }

    fn parse_flow_mapping(&mut self) -> Result<()> {
        // Simplified flow mapping parsing
        if let Some(end) = self.buffer.find('}') {
            let content = self.buffer.drain(..=end).collect::<String>();
            self.emit_mapping_start()?;
            // Parse content (simplified)
            self.emit_mapping_end()?;
            self.position.column += content.len();
        }
        Ok(())
    }

    fn parse_block_scalar_start(&mut self, _indicator: char) -> Result<()> {
        self.buffer.remove(0); // Remove indicator
        self.context.in_block_scalar = true;
        // Parse block scalar header (simplified)
        Ok(())
    }

    fn parse_block_scalar_content(&mut self) -> Result<()> {
        // Simplified block scalar parsing
        if let Some(end) = self.find_block_scalar_end() {
            let content = self.buffer.drain(..end).collect::<String>();
            self.emit_scalar(content)?;
            self.context.in_block_scalar = false;
        }
        Ok(())
    }

    fn find_block_scalar_end(&self) -> Option<usize> {
        // Simplified - find dedent or document marker
        self.buffer.find("\n\n").or(self.buffer.find("\n..."))
    }

    fn parse_anchor(&mut self) -> Result<()> {
        self.buffer.remove(0); // Remove '&'
        let end = self.find_identifier_end();
        if let Some(end) = end {
            let anchor = self.buffer.drain(..end).collect::<String>();
            self.context.pending_anchor = Some(anchor);
            self.position.column += end + 1;
        }
        Ok(())
    }

    fn parse_alias(&mut self) -> Result<()> {
        self.buffer.remove(0); // Remove '*'
        let end = self.find_identifier_end();
        if let Some(end) = end {
            let alias = self.buffer.drain(..end).collect::<String>();
            self.emit_alias(alias)?;
            self.position.column += end + 1;
        }
        Ok(())
    }

    fn parse_quoted_scalar(&mut self, quote: char) -> Result<()> {
        self.buffer.remove(0); // Remove opening quote
        if let Some(end) = self.buffer.find(quote) {
            let content = self.buffer.drain(..end).collect::<String>();
            self.buffer.remove(0); // Remove closing quote
            let content_len = content.len();
            self.emit_scalar(content)?;
            self.position.column += content_len + 2;
        }
        Ok(())
    }

    fn parse_plain_scalar(&mut self) -> Result<()> {
        let end = self.find_plain_scalar_end();
        if let Some(end) = end {
            let content = self.buffer.drain(..end).collect::<String>();
            self.emit_scalar(content.trim().to_string())?;
            self.position.column += end;
        }
        Ok(())
    }

    fn find_identifier_end(&self) -> Option<usize> {
        for (i, ch) in self.buffer.char_indices() {
            if !ch.is_alphanumeric() && ch != '_' && ch != '-' {
                return Some(i);
            }
        }
        None
    }

    fn find_plain_scalar_end(&self) -> Option<usize> {
        // Find end of plain scalar (simplified)
        for (i, ch) in self.buffer.char_indices() {
            if ch == '\n' || ch == ':' || ch == '#' {
                return Some(i);
            }
        }
        Some(self.buffer.len())
    }

    /// Event emission methods
    fn emit_stream_start(&mut self) -> Result<()> {
        self.events.push_back(Event {
            event_type: EventType::StreamStart,
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_stream_end(&mut self) -> Result<()> {
        self.events.push_back(Event {
            event_type: EventType::StreamEnd,
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_document_start(&mut self) -> Result<()> {
        self.events.push_back(Event {
            event_type: EventType::DocumentStart {
                version: None,
                tags: Vec::new(),
                implicit: false,
            },
            position: self.position,
        });
        self.stats.events_generated += 1;
        self.stats.documents_parsed += 1;
        Ok(())
    }

    fn emit_document_end(&mut self) -> Result<()> {
        self.events.push_back(Event {
            event_type: EventType::DocumentEnd { implicit: false },
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_sequence_start(&mut self) -> Result<()> {
        let anchor = self.context.pending_anchor.take();
        let tag = self.context.pending_tag.take();

        self.events.push_back(Event {
            event_type: EventType::SequenceStart {
                anchor,
                tag,
                flow_style: false,
            },
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_sequence_end(&mut self) -> Result<()> {
        self.events.push_back(Event {
            event_type: EventType::SequenceEnd,
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_mapping_start(&mut self) -> Result<()> {
        let anchor = self.context.pending_anchor.take();
        let tag = self.context.pending_tag.take();

        self.events.push_back(Event {
            event_type: EventType::MappingStart {
                anchor,
                tag,
                flow_style: false,
            },
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_mapping_end(&mut self) -> Result<()> {
        self.events.push_back(Event {
            event_type: EventType::MappingEnd,
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_scalar(&mut self, value: String) -> Result<()> {
        let anchor = self.context.pending_anchor.take();
        let tag = self.context.pending_tag.take();

        self.events.push_back(Event {
            event_type: EventType::Scalar {
                value,
                anchor,
                tag,
                style: crate::parser::ScalarStyle::Plain,
                plain_implicit: true,
                quoted_implicit: true,
            },
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    fn emit_alias(&mut self, anchor: String) -> Result<()> {
        self.events.push_back(Event {
            event_type: EventType::Alias { anchor },
            position: self.position,
        });
        self.stats.events_generated += 1;
        Ok(())
    }

    /// Get the next event if available
    pub fn next_event(&mut self) -> Option<Event> {
        self.events.pop_front()
    }

    /// Check if there are events available
    pub fn has_events(&self) -> bool {
        !self.events.is_empty()
    }

    /// Get current statistics
    pub fn stats(&self) -> &StreamStats {
        &self.stats
    }

    /// Get remaining buffer size
    pub fn buffer_size(&self) -> usize {
        self.buffer.len()
    }
}

/// Iterator implementation for streaming parser
impl<R: BufRead> Iterator for StreamingYamlParser<R> {
    type Item = Result<Event>;

    fn next(&mut self) -> Option<Self::Item> {
        // Try to get an event from the buffer
        if let Some(event) = self.next_event() {
            return Some(Ok(event));
        }

        // Parse more data if needed
        match self.parse_next() {
            Ok(true) => self.next_event().map(Ok),
            Ok(false) if self.state == StreamState::EndOfStream => {
                if !self.events.is_empty() {
                    self.next_event().map(Ok)
                } else {
                    None
                }
            }
            Ok(false) => None,
            Err(e) => Some(Err(e)),
        }
    }
}

/// Create a streaming parser from a file path
pub fn stream_from_file<P: AsRef<Path>>(
    path: P,
    config: StreamConfig,
) -> Result<StreamingYamlParser<BufReader<std::fs::File>>> {
    let file = std::fs::File::open(path)?;
    let reader = BufReader::with_capacity(config.buffer_size, file);
    Ok(StreamingYamlParser::new(reader, config))
}

/// Create a streaming parser from a string
pub fn stream_from_string(
    input: String,
    config: StreamConfig,
) -> StreamingYamlParser<BufReader<std::io::Cursor<String>>> {
    let cursor = std::io::Cursor::new(input);
    let reader = BufReader::new(cursor);
    StreamingYamlParser::new(reader, config)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Cursor;

    #[test]
    fn test_basic_streaming() {
        let yaml = "---\nkey: value\n...\n---\nother: data\n...";
        let cursor = Cursor::new(yaml.to_string());
        let reader = BufReader::new(cursor);
        let mut parser = StreamingYamlParser::new(reader, StreamConfig::default());

        let mut events = Vec::new();
        while let Some(event) = parser.next() {
            events.push(event.unwrap());
        }

        assert!(events.len() > 0);
        assert!(matches!(events[0].event_type, EventType::StreamStart));
    }

    #[test]
    fn test_incremental_parsing() {
        let yaml = "key: value\nlist:\n  - item1\n  - item2";
        let mut parser = stream_from_string(yaml.to_string(), StreamConfig::default());

        // Parse incrementally
        let mut event_count = 0;
        while parser.parse_next().unwrap() {
            while let Some(_event) = parser.next_event() {
                event_count += 1;
            }
        }

        assert!(event_count > 0);
    }

    #[test]
    fn test_large_buffer_handling() {
        let mut yaml = String::new();
        for i in 0..1000 {
            yaml.push_str(&format!("item{}: value{}\n", i, i));
        }

        let config = StreamConfig::large_file();
        let mut parser = stream_from_string(yaml, config);

        let mut events = Vec::new();
        for event in parser.take(100) {
            events.push(event.unwrap());
        }

        assert!(events.len() > 0);
    }
}