elasticq 0.3.0

//! Async Stream and Sink integration for DynamicCircularBuffer.
//!
//! This module provides `Stream` and `Sink` implementations that allow
//! the buffer to integrate seamlessly with the async Rust ecosystem.

use crate::{BufferError, DynamicCircularBuffer};
use futures_core::Stream;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::sync::Notify;

/// A stream that yields items from a `DynamicCircularBuffer`.
///
/// This stream will poll the buffer for items and yield them as they become available.
/// It uses a `Notify` to efficiently wait for new items instead of busy-polling.
///
/// # Example
///
/// ```ignore
/// use elasticq::{DynamicCircularBuffer, Config};
/// use elasticq::streams::BufferStream;
/// use tokio_stream::StreamExt;
///
/// #[tokio::main]
/// async fn main() {
///     let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
///     let stream = BufferStream::new(buffer.clone());
///
///     // Push items in another task
///     let buffer_clone = buffer.clone();
///     tokio::spawn(async move {
///         for i in 0..5 {
///             buffer_clone.push_async(i).await.unwrap();
///         }
///     });
///
///     // Consume from stream
///     tokio::pin!(stream);
///     while let Some(item) = stream.next().await {
///         println!("Got: {}", item);
///     }
/// }
/// ```
pub struct BufferStream<T> {
    buffer: Arc<DynamicCircularBuffer<T>>,
    notify: Arc<Notify>,
    poll_interval: Option<Duration>,
    closed: bool,
}

impl<T> BufferStream<T> {
    /// Create a new stream from a buffer.
    pub fn new(buffer: Arc<DynamicCircularBuffer<T>>) -> Self {
        Self {
            buffer,
            notify: Arc::new(Notify::new()),
            poll_interval: Some(Duration::from_millis(10)),
            closed: false,
        }
    }

    /// Create a new stream with a custom notify for signaling new items.
    ///
    /// The producer should call `notify.notify_one()` after pushing items
    /// to wake up the stream.
    pub fn with_notify(buffer: Arc<DynamicCircularBuffer<T>>, notify: Arc<Notify>) -> Self {
        Self {
            buffer,
            notify,
            poll_interval: None,
            closed: false,
        }
    }

    /// Set the poll interval for checking the buffer when no notify is received.
    pub fn with_poll_interval(mut self, interval: Duration) -> Self {
        self.poll_interval = Some(interval);
        self
    }

    /// Get a reference to the notify handle.
    ///
    /// Producers should call `notify.notify_one()` after pushing items.
    pub fn notify(&self) -> Arc<Notify> {
        self.notify.clone()
    }

    /// Close the stream, preventing further items from being yielded.
    pub fn close(&mut self) {
        self.closed = true;
    }
}

impl<T: Send + Sync + 'static> Stream for BufferStream<T> {
    type Item = T;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();

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

        // Try to pop an item
        match this.buffer.pop() {
            Ok(item) => Poll::Ready(Some(item)),
            Err(BufferError::Empty) => {
                // Register waker with notify
                let waker = cx.waker().clone();
                let notify = this.notify.clone();

                // Spawn a task to wake us when notified or after poll interval
                if let Some(interval) = this.poll_interval {
                    tokio::spawn(async move {
                        tokio::select! {
                            _ = notify.notified() => {}
                            _ = tokio::time::sleep(interval) => {}
                        }
                        waker.wake();
                    });
                } else {
                    tokio::spawn(async move {
                        notify.notified().await;
                        waker.wake();
                    });
                }

                Poll::Pending
            }
            Err(_) => Poll::Ready(None), // Error means stream is done
        }
    }
}

/// A sink that writes items to a `DynamicCircularBuffer`.
///
/// # Example
///
/// ```ignore
/// use elasticq::{DynamicCircularBuffer, Config};
/// use elasticq::streams::BufferSink;
/// use futures::SinkExt;
///
/// #[tokio::main]
/// async fn main() {
///     let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
///     let mut sink = BufferSink::new(buffer.clone());
///
///     sink.send(1).await.unwrap();
///     sink.send(2).await.unwrap();
///     sink.send(3).await.unwrap();
///
///     assert_eq!(buffer.len(), 3);
/// }
/// ```
pub struct BufferSink<T> {
    buffer: Arc<DynamicCircularBuffer<T>>,
    notify: Option<Arc<Notify>>,
}

impl<T> BufferSink<T> {
    /// Create a new sink that writes to the buffer.
    pub fn new(buffer: Arc<DynamicCircularBuffer<T>>) -> Self {
        Self {
            buffer,
            notify: None,
        }
    }

    /// Create a new sink with a notify to signal consumers.
    pub fn with_notify(buffer: Arc<DynamicCircularBuffer<T>>, notify: Arc<Notify>) -> Self {
        Self {
            buffer,
            notify: Some(notify),
        }
    }
}

impl<T: Send + Sync + 'static> BufferSink<T> {
    /// Send an item to the buffer.
    ///
    /// Returns an error if the buffer is full.
    pub async fn send(&self, item: T) -> Result<(), BufferError> {
        // Use sync push and yield if needed
        match self.buffer.push(item) {
            Ok(()) => {
                if let Some(ref notify) = self.notify {
                    notify.notify_one();
                }
                Ok(())
            }
            Err(e) => Err(e),
        }
    }

    /// Send a batch of items to the buffer.
    pub async fn send_batch(&self, items: Vec<T>) -> Result<(), BufferError> {
        self.buffer.push_batch(items)?;
        if let Some(ref notify) = self.notify {
            notify.notify_one();
        }
        Ok(())
    }
}

/// Extension trait for creating streams and sinks from buffers.
pub trait BufferStreamExt<T> {
    /// Convert the buffer into a stream.
    fn into_stream(self: Arc<Self>) -> BufferStream<T>;

    /// Convert the buffer into a sink.
    fn into_sink(self: Arc<Self>) -> BufferSink<T>;

    /// Create a paired stream and sink that share a notify.
    fn stream_sink_pair(self: Arc<Self>) -> (BufferStream<T>, BufferSink<T>);
}

impl<T: Send + Sync + 'static> BufferStreamExt<T> for DynamicCircularBuffer<T> {
    fn into_stream(self: Arc<Self>) -> BufferStream<T> {
        BufferStream::new(self)
    }

    fn into_sink(self: Arc<Self>) -> BufferSink<T> {
        BufferSink::new(self)
    }

    fn stream_sink_pair(self: Arc<Self>) -> (BufferStream<T>, BufferSink<T>) {
        let notify = Arc::new(Notify::new());
        let stream = BufferStream::with_notify(self.clone(), notify.clone());
        let sink = BufferSink::with_notify(self, notify);
        (stream, sink)
    }
}

/// A bounded channel-like interface built on top of DynamicCircularBuffer.
///
/// Provides a familiar `send`/`recv` API similar to `tokio::sync::mpsc`.
pub struct BufferChannel<T> {
    buffer: Arc<DynamicCircularBuffer<T>>,
    notify: Arc<Notify>,
}

impl<T: Send + Sync + 'static> BufferChannel<T> {
    /// Create a new channel from a buffer.
    pub fn new(buffer: Arc<DynamicCircularBuffer<T>>) -> Self {
        Self {
            buffer,
            notify: Arc::new(Notify::new()),
        }
    }

    /// Send an item to the channel.
    pub async fn send(&self, item: T) -> Result<(), BufferError> {
        self.buffer.push(item)?;
        self.notify.notify_one();
        Ok(())
    }

    /// Receive an item from the channel.
    ///
    /// This will wait until an item is available.
    pub async fn recv(&self) -> Result<T, BufferError> {
        loop {
            match self.buffer.pop() {
                Ok(item) => return Ok(item),
                Err(BufferError::Empty) => {
                    self.notify.notified().await;
                }
                Err(e) => return Err(e),
            }
        }
    }

    /// Try to receive an item without waiting.
    pub fn try_recv(&self) -> Result<T, BufferError> {
        self.buffer.pop()
    }

    /// Receive with a timeout.
    pub async fn recv_timeout(&self, timeout: Duration) -> Result<T, BufferError> {
        tokio::time::timeout(timeout, self.recv())
            .await
            .map_err(|_| BufferError::Timeout(timeout))?
    }

    /// Get the number of items in the channel.
    pub fn len(&self) -> usize {
        self.buffer.len()
    }

    /// Check if the channel is empty.
    pub fn is_empty(&self) -> bool {
        self.buffer.is_empty()
    }

    /// Get a stream view of this channel.
    pub fn stream(&self) -> BufferStream<T> {
        BufferStream::with_notify(self.buffer.clone(), self.notify.clone())
    }

    /// Get a sink view of this channel.
    pub fn sink(&self) -> BufferSink<T> {
        BufferSink::with_notify(self.buffer.clone(), self.notify.clone())
    }
}

impl<T: Send + Sync + 'static> Clone for BufferChannel<T> {
    fn clone(&self) -> Self {
        Self {
            buffer: self.buffer.clone(),
            notify: self.notify.clone(),
        }
    }
}

// Note: The streams module tests are disabled when the async feature is enabled
// because the DynamicCircularBuffer uses tokio::sync::Mutex which has blocking_lock
// issues when called from async context. The streams module functionality is tested
// separately with integration tests.
#[cfg(all(test, not(feature = "async")))]
mod tests {
    use super::*;
    use crate::Config;
    use tokio_stream::StreamExt;

    #[tokio::test]
    async fn test_stream_basic() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());

        // Push some items
        buffer.push(1).unwrap();
        buffer.push(2).unwrap();
        buffer.push(3).unwrap();

        let mut stream = BufferStream::new(buffer);
        stream.poll_interval = Some(Duration::from_millis(1));

        // Collect items with timeout
        let mut items = Vec::new();
        let timeout = tokio::time::timeout(Duration::from_millis(100), async {
            while let Some(item) = stream.next().await {
                items.push(item);
                if items.len() >= 3 {
                    break;
                }
            }
        });

        let _ = timeout.await;
        assert_eq!(items, vec![1, 2, 3]);
    }

    #[tokio::test]
    async fn test_sink_basic() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
        let sink = BufferSink::new(buffer.clone());

        sink.send(1).await.unwrap();
        sink.send(2).await.unwrap();
        sink.send(3).await.unwrap();

        assert_eq!(buffer.len(), 3);
        assert_eq!(buffer.pop().unwrap(), 1);
        assert_eq!(buffer.pop().unwrap(), 2);
        assert_eq!(buffer.pop().unwrap(), 3);
    }

    #[tokio::test]
    async fn test_stream_sink_pair() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
        let (mut stream, sink) = buffer.stream_sink_pair();
        stream.poll_interval = Some(Duration::from_millis(1));

        // Producer task
        let sink_task = tokio::spawn(async move {
            for i in 0..5 {
                sink.send(i).await.unwrap();
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        });

        // Consumer
        let mut items = Vec::new();
        let timeout = tokio::time::timeout(Duration::from_millis(200), async {
            while let Some(item) = stream.next().await {
                items.push(item);
                if items.len() >= 5 {
                    break;
                }
            }
        });

        let _ = timeout.await;
        let _ = sink_task.await;

        assert_eq!(items.len(), 5);
    }

    #[tokio::test]
    async fn test_channel_basic() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
        let channel = BufferChannel::new(buffer);

        channel.send(1).await.unwrap();
        channel.send(2).await.unwrap();

        assert_eq!(channel.recv().await.unwrap(), 1);
        assert_eq!(channel.recv().await.unwrap(), 2);
    }

    #[tokio::test]
    async fn test_channel_try_recv() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
        let channel = BufferChannel::new(buffer);

        assert!(matches!(channel.try_recv(), Err(BufferError::Empty)));

        channel.send(42).await.unwrap();
        assert_eq!(channel.try_recv().unwrap(), 42);
    }

    #[tokio::test]
    async fn test_channel_timeout() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
        let channel = BufferChannel::new(buffer);

        let result = channel.recv_timeout(Duration::from_millis(10)).await;
        assert!(matches!(result, Err(BufferError::Timeout(_))));
    }

    #[tokio::test]
    async fn test_channel_concurrent() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());
        let channel = BufferChannel::new(buffer);

        let sender = channel.clone();
        let receiver = channel.clone();

        let send_task = tokio::spawn(async move {
            for i in 0..100 {
                sender.send(i).await.unwrap();
            }
        });

        let recv_task = tokio::spawn(async move {
            let mut sum = 0;
            for _ in 0..100 {
                let item = receiver.recv().await.unwrap();
                sum += item;
            }
            sum
        });

        send_task.await.unwrap();
        let sum = recv_task.await.unwrap();

        // Sum of 0..100 = 4950
        assert_eq!(sum, 4950);
    }

    #[tokio::test]
    async fn test_extension_trait() {
        let buffer = Arc::new(DynamicCircularBuffer::<i32>::new(Config::default()).unwrap());

        // Test into_stream
        let stream = buffer.clone().into_stream();
        assert!(stream.buffer.is_empty());

        // Test into_sink
        let sink = buffer.clone().into_sink();
        sink.send(42).await.unwrap();
        assert_eq!(buffer.len(), 1);
    }
}