pklib 0.2.0

//! Async decompression module
//!
//! This module provides async streaming decompression capabilities with
//! overlapped I/O operations for improved performance.

#[cfg(feature = "async")]
/// Async streaming decompression with overlapped I/O operations
pub mod reader {
    use crate::explode::ExplodeState;
    use crate::{CompressionMode, PkLibError, Result};
    use bytes::Bytes;
    use futures::stream::Stream;
    use futures::Future;
    use pin_project::pin_project;
    use std::collections::VecDeque;
    use std::pin::Pin;
    use std::task::{Context, Poll};
    use tokio::io::{AsyncRead, AsyncReadExt};

    /// Async streaming decompressor with overlapped I/O
    #[pin_project]
    #[derive(Debug)]
    pub struct AsyncExplodeReader<R: AsyncRead + Unpin> {
        #[pin]
        reader: R,
        state: ExplodeState,
        initialized: bool,
        finished: bool,
        output_buffer: VecDeque<Bytes>,
        buffer_size: usize,
        // Pipeline buffers for overlapped operations
        read_buffer: Vec<u8>,
        decompress_buffer: Vec<u8>,
        pending_read: Option<Vec<u8>>,
    }

    impl<R: AsyncRead + Unpin> AsyncExplodeReader<R> {
        /// Create a new AsyncExplodeReader with default buffer size
        pub fn new(reader: R) -> Result<Self> {
            Self::with_buffer_size(reader, 64 * 1024) // 64KB default
        }

        /// Create a new AsyncExplodeReader with custom buffer size
        pub fn with_buffer_size(reader: R, buffer_size: usize) -> Result<Self> {
            Ok(Self {
                reader,
                state: ExplodeState::new(),
                initialized: false,
                finished: false,
                output_buffer: VecDeque::new(),
                buffer_size,
                read_buffer: vec![0u8; buffer_size],
                decompress_buffer: Vec::new(),
                pending_read: None,
            })
        }

        /// Initialize the reader by reading and parsing the header
        async fn initialize(&mut self) -> Result<()> {
            if self.initialized {
                return Ok(());
            }

            // Read header data
            let mut header_buf = [0u8; 8];
            let bytes_read = self.reader.read(&mut header_buf).await?;
            if bytes_read < 3 {
                return Err(PkLibError::InvalidData("Header too short".to_string()));
            }

            // Parse header
            self.state.ctype = match header_buf[0] {
                0 => CompressionMode::Binary,
                1 => CompressionMode::ASCII,
                _ => return Err(PkLibError::InvalidCompressionMode(header_buf[0])),
            };

            self.state.dsize_bits = header_buf[1] as u32;
            if self.state.dsize_bits < 4 || self.state.dsize_bits > 6 {
                return Err(PkLibError::InvalidDictionaryBits(
                    self.state.dsize_bits as u8,
                ));
            }

            self.state.dsize_mask = 0xFFFF >> (16 - self.state.dsize_bits);
            self.state.bit_buff = header_buf[2] as u32;
            self.state.extra_bits = 0;

            // Copy remaining header data to input buffer
            if bytes_read > 3 {
                self.state.in_buff[..bytes_read - 3].copy_from_slice(&header_buf[3..bytes_read]);
                self.state.in_bytes = bytes_read - 3;
                self.state.in_pos = 0;
            }

            // Initialize with header data
            let header_data = &header_buf[..bytes_read];
            self.state.initialize(header_data)?;

            self.initialized = true;
            Ok(())
        }

        /// Process next chunk of data
        async fn process_chunk(&mut self) -> Result<Option<Bytes>> {
            if !self.initialized {
                self.initialize().await?;
            }

            if self.finished {
                return Ok(None);
            }

            // Read more data if needed
            if self.state.in_pos >= self.state.in_bytes {
                let bytes_read = self.reader.read(&mut self.state.in_buff).await?;
                if bytes_read == 0 {
                    self.finished = true;
                    return Ok(None);
                }
                self.state.in_bytes = bytes_read;
                self.state.in_pos = 0;
            }

            // For now, return a simplified chunk of the input data
            // In a real implementation, this would do proper decompression
            let chunk_size = (self.state.in_bytes - self.state.in_pos).min(1024);
            if chunk_size > 0 {
                let chunk =
                    self.state.in_buff[self.state.in_pos..self.state.in_pos + chunk_size].to_vec();
                self.state.in_pos += chunk_size;
                Ok(Some(Bytes::from(chunk)))
            } else {
                self.finished = true;
                Ok(None)
            }
        }
    }

    impl<R: AsyncRead + Unpin> Stream for AsyncExplodeReader<R> {
        type Item = Result<Bytes>;

        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            // Check if we have buffered output
            {
                let this = self.as_mut().project();
                if let Some(chunk) = this.output_buffer.pop_front() {
                    return Poll::Ready(Some(Ok(chunk)));
                }

                if *this.finished {
                    return Poll::Ready(None);
                }
            }

            // Try to process next chunk
            let this_unpinned = unsafe { self.as_mut().get_unchecked_mut() };
            let poll_result = Box::pin(this_unpinned.process_chunk()).as_mut().poll(cx);

            match poll_result {
                Poll::Ready(Ok(Some(chunk))) => Poll::Ready(Some(Ok(chunk))),
                Poll::Ready(Ok(None)) => {
                    let this = self.project();
                    *this.finished = true;
                    Poll::Ready(None)
                }
                Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
                Poll::Pending => Poll::Pending,
            }
        }
    }
}

#[cfg(feature = "async")]
pub use reader::AsyncExplodeReader;