epub_stream/

streaming.rs

//! Streaming chapter reader for memory-efficient EPUB processing.
//!
//! Provides truly streaming chapter processing that reads directly from ZIP
//! without materializing the full chapter content.

extern crate alloc;

use alloc::string::{String, ToString};
use alloc::vec::Vec;
use core::cmp::min;

#[cfg(feature = "std")]
use crate::render_prep::{RenderPrepError, RenderPrepOptions, StyledEventOrRun};

/// Scratch buffer pool for streaming operations.
///
/// Pre-allocated buffers that can be reused across operations to avoid
/// repeated allocations in hot paths.
#[derive(Debug)]
pub struct ScratchBuffers {
    /// Primary buffer for reading chunks from ZIP
    pub read_buf: Vec<u8>,
    /// Buffer for XML parsing events
    pub xml_buf: Vec<u8>,
    /// Buffer for text accumulation
    pub text_buf: String,
}

impl ScratchBuffers {
    /// Create scratch buffers with specified capacities.
    pub fn new(read_capacity: usize, xml_capacity: usize) -> Self {
        Self {
            read_buf: Vec::with_capacity(read_capacity),
            xml_buf: Vec::with_capacity(xml_capacity),
            text_buf: String::with_capacity(4096),
        }
    }

    /// Create buffers suitable for embedded use (small, bounded).
    pub fn embedded() -> Self {
        Self::new(8192, 4096)
    }

    /// Create buffers for desktop use (larger, more performant).
    pub fn desktop() -> Self {
        Self::new(65536, 32768)
    }

    /// Clear all buffers without deallocating.
    pub fn clear(&mut self) {
        self.read_buf.clear();
        self.xml_buf.clear();
        self.text_buf.clear();
    }
}

/// Chunking limits for incremental processing.
///
/// Prevents single large allocations by breaking work into smaller chunks.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ChunkLimits {
    /// Maximum bytes to process in a single read operation.
    pub max_read_chunk: usize,
    /// Maximum bytes for accumulated text before forcing a flush.
    pub max_text_accumulation: usize,
    /// Maximum number of events to process before yielding control.
    pub max_events_per_yield: usize,
    /// Maximum depth for element stack.
    pub max_stack_depth: usize,
}

impl Default for ChunkLimits {
    fn default() -> Self {
        Self {
            max_read_chunk: 16384,       // 16KB read chunks
            max_text_accumulation: 8192, // 8KB text buffer
            max_events_per_yield: 1000,  // Process 1000 events at a time
            max_stack_depth: 256,        // 256 levels of nesting
        }
    }
}

impl ChunkLimits {
    /// Conservative limits for embedded environments.
    pub fn embedded() -> Self {
        Self {
            max_read_chunk: 4096,        // 4KB read chunks
            max_text_accumulation: 2048, // 2KB text buffer
            max_events_per_yield: 500,   // Process 500 events at a time
            max_stack_depth: 64,         // 64 levels of nesting
        }
    }
}

/// Stateful pagination context for resumable page layout.
///
/// Tracks parsing/layout state so page N+1 can continue from where
/// page N left off without re-parsing from the start.
#[derive(Clone, Debug)]
pub struct PaginationContext {
    /// Current byte offset in the source document.
    pub byte_offset: usize,
    /// Current event/token index.
    pub event_index: usize,
    /// Current element stack (path from root to current element).
    pub element_stack: Vec<String>,
    /// Accumulated text since last page break.
    pub text_accumulator: String,
    /// Current page number.
    pub page_number: usize,
}

impl Default for PaginationContext {
    fn default() -> Self {
        Self {
            byte_offset: 0,
            event_index: 0,
            element_stack: Vec::with_capacity(32),
            text_accumulator: String::with_capacity(4096),
            page_number: 0,
        }
    }
}

impl PaginationContext {
    /// Create a new context for starting at the beginning.
    pub fn new() -> Self {
        Self::default()
    }

    /// Reset for a new chapter.
    pub fn reset(&mut self) {
        self.byte_offset = 0;
        self.event_index = 0;
        self.element_stack.clear();
        self.text_accumulator.clear();
        self.page_number = 0;
    }

    /// Advance to the next page.
    pub fn next_page(&mut self) {
        self.page_number += 1;
        self.text_accumulator.clear();
    }

    /// Update byte offset.
    pub fn advance_bytes(&mut self, bytes: usize) {
        self.byte_offset += bytes;
    }

    /// Update event index.
    pub fn advance_events(&mut self, events: usize) {
        self.event_index += events;
    }

    /// Push element onto stack.
    pub fn push_element(&mut self, tag: &str) {
        self.element_stack.push(tag.to_string());
    }

    /// Pop element from stack.
    pub fn pop_element(&mut self) -> Option<String> {
        self.element_stack.pop()
    }

    /// Accumulate text.
    pub fn append_text(&mut self, text: &str, max_len: usize) {
        let remaining = max_len.saturating_sub(self.text_accumulator.len());
        if remaining > 0 {
            let to_add = &text[..min(text.len(), remaining)];
            self.text_accumulator.push_str(to_add);
        }
    }
}

/// Memory chunk allocator for bounded allocations.
///
/// Manages a pool of fixed-size chunks to avoid large contiguous allocations.
pub struct ChunkAllocator {
    chunk_size: usize,
    max_chunks: usize,
    chunks: Vec<Vec<u8>>,
    allocated: usize,
}

impl ChunkAllocator {
    /// Create a new chunk allocator.
    pub fn new(chunk_size: usize, max_chunks: usize) -> Self {
        Self {
            chunk_size,
            max_chunks,
            chunks: Vec::with_capacity(max_chunks),
            allocated: 0,
        }
    }

    /// Get a chunk from the pool or allocate new.
    pub fn acquire(&mut self) -> Option<Vec<u8>> {
        if let Some(chunk) = self.chunks.pop() {
            Some(chunk)
        } else if self.allocated < self.max_chunks {
            self.allocated += 1;
            Some(Vec::with_capacity(self.chunk_size))
        } else {
            None
        }
    }

    /// Return a chunk to the pool.
    pub fn release(&mut self, mut chunk: Vec<u8>) {
        if self.chunks.len() < self.max_chunks {
            chunk.clear();
            self.chunks.push(chunk);
        }
    }

    /// Current number of available chunks.
    pub fn available(&self) -> usize {
        self.chunks.len()
    }
}

/// Statistics for streaming operations.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct StreamingStats {
    /// Total bytes read from source.
    pub bytes_read: usize,
    /// Total bytes processed.
    pub bytes_processed: usize,
    /// Number of events emitted.
    pub events_emitted: usize,
    /// Number of chunks processed.
    pub chunks_processed: usize,
    /// Peak memory usage estimate.
    pub peak_memory_estimate: usize,
}

/// Streaming chapter processor that reads incrementally from ZIP.
///
/// This type provides true streaming without materializing the full
/// chapter content in memory.
pub struct StreamingChapterProcessor {
    #[allow(dead_code)]
    limits: ChunkLimits,
    #[allow(dead_code)]
    state: StreamingParseState,
}

/// Current state of streaming parse.
#[derive(Clone, Debug)]
#[allow(dead_code)]
enum StreamingParseState {
    /// Initial state, ready to start.
    Initial,
    /// Parsing in progress with partial content buffered.
    Parsing {
        /// Bytes processed so far
        bytes_processed: usize,
        /// Events emitted so far
        events_emitted: usize,
        /// Current element stack depth
        stack_depth: usize,
    },
    /// Parsing complete.
    Complete,
    /// Error occurred during parsing.
    Error(String),
}

#[cfg(feature = "std")]
impl StreamingChapterProcessor {
    /// Create a new streaming processor.
    pub fn new(_options: RenderPrepOptions, limits: ChunkLimits) -> Self {
        Self {
            limits,
            state: StreamingParseState::Initial,
        }
    }

    /// Process a chunk of HTML bytes and emit styled items.
    ///
    /// Returns the number of items emitted. When the chunk is exhausted
    /// but the document is not complete, returns Ok(0) to indicate
    /// more data is needed.
    pub fn process_chunk<F>(
        &mut self,
        _html_chunk: &[u8],
        mut _on_item: F,
    ) -> Result<usize, RenderPrepError>
    where
        F: FnMut(StyledEventOrRun),
    {
        let count = 0usize;

        // For now, this is a placeholder implementation
        // Full implementation would use incremental XML parsing
        self.state = StreamingParseState::Complete;

        Ok(count)
    }

    /// Check if parsing is complete.
    pub fn is_complete(&self) -> bool {
        matches!(self.state, StreamingParseState::Complete)
    }
}

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

    #[test]
    fn test_scratch_buffers_embedded() {
        let buffers = ScratchBuffers::embedded();
        assert!(buffers.read_buf.capacity() <= 8192);
        assert!(buffers.xml_buf.capacity() <= 4096);
    }

    #[test]
    fn test_scratch_buffers_clear_preserves_capacity() {
        let mut buffers = ScratchBuffers::desktop();
        let read_cap = buffers.read_buf.capacity();

        buffers.read_buf.extend_from_slice(b"test data");
        buffers.clear();

        assert!(buffers.read_buf.is_empty());
        assert_eq!(buffers.read_buf.capacity(), read_cap);
    }

    #[test]
    fn test_pagination_context_basic() {
        let mut ctx = PaginationContext::new();
        assert_eq!(ctx.page_number, 0);
        assert_eq!(ctx.byte_offset, 0);

        ctx.advance_bytes(100);
        assert_eq!(ctx.byte_offset, 100);

        ctx.next_page();
        assert_eq!(ctx.page_number, 1);
        assert!(ctx.text_accumulator.is_empty());
    }

    #[test]
    fn test_pagination_context_stack() {
        let mut ctx = PaginationContext::new();
        ctx.push_element("html");
        ctx.push_element("body");
        ctx.push_element("p");

        assert_eq!(ctx.element_stack.len(), 3);
        assert_eq!(ctx.pop_element(), Some("p".to_string()));
        assert_eq!(ctx.element_stack.len(), 2);
    }

    #[test]
    fn test_chunk_allocator_basic() {
        let mut allocator = ChunkAllocator::new(1024, 10);
        assert_eq!(allocator.available(), 0);

        let chunk = allocator.acquire().unwrap();
        assert_eq!(chunk.capacity(), 1024);

        allocator.release(chunk);
        assert_eq!(allocator.available(), 1);
    }

    #[test]
    fn test_chunk_allocator_exhaustion() {
        let mut allocator = ChunkAllocator::new(1024, 2);
        let _ = allocator.acquire();
        let _ = allocator.acquire();
        assert!(allocator.acquire().is_none()); // Exhausted
    }
}