monocoque-rs-zmtp 0.3.0

//! Stream and Sink adapters for ZeroMQ sockets.
//!
//! This module provides adapters that implement `futures::Stream` and `futures::Sink`
//! for ZeroMQ sockets, enabling integration with the Rust async ecosystem.
//!
//! # Examples
//!
//! ```no_run
//! use monocoque_zmtp::DealerSocket;
//! use monocoque_zmtp::stream_sink::SocketStream;
//! use futures::StreamExt;
//!
//! # async fn example() -> std::io::Result<()> {
//! let mut socket = DealerSocket::connect("127.0.0.1:5555").await?;
//! let mut stream = SocketStream::new(socket);
//!
//! while let Some(msg) = stream.next().await {
//!     println!("Received: {:?}", msg?);
//! }
//! # Ok(())
//! # }
//! ```

use bytes::Bytes;
use futures::sink::Sink;
use futures::stream::Stream;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};

use crate::Socket;

/// Adapter that implements `Stream` for any socket implementing the `Socket` trait.
///
/// This allows using ZeroMQ sockets with stream combinators like `filter`, `map`,
/// `for_each`, etc.
///
/// # Examples
///
/// ```no_run
/// use monocoque_zmtp::{DealerSocket, stream_sink::SocketStream};
/// use futures::StreamExt;
///
/// # async fn example() -> std::io::Result<()> {
/// let socket = DealerSocket::connect("127.0.0.1:5555").await?;
/// let stream = SocketStream::new(socket);
///
/// // Use stream combinators
/// stream
///     .filter(|msg| futures::future::ready(msg.is_ok()))
///     .for_each(|msg| async move {
///         println!("Message: {:?}", msg);
///     })
///     .await;
/// # Ok(())
/// # }
/// ```
pub struct SocketStream<S> {
    socket: S,
}

impl<S> SocketStream<S> {
    /// Create a new stream adapter for a socket.
    pub const fn new(socket: S) -> Self {
        Self { socket }
    }

    /// Get a reference to the underlying socket.
    pub const fn get_ref(&self) -> &S {
        &self.socket
    }

    /// Get a mutable reference to the underlying socket.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.socket
    }

    /// Consume the adapter and return the underlying socket.
    pub fn into_inner(self) -> S {
        self.socket
    }
}

impl<S: Socket + Unpin> Stream for SocketStream<S> {
    type Item = io::Result<Vec<Bytes>>;

    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // Stream::poll_next cannot store the async recv() future across poll calls
        // without a Pin-boxed future field. Return an explicit error instead of
        // silently hanging (Poll::Pending) so callers notice the limitation.
        // Use socket.recv().await directly instead of this Stream adapter.
        Poll::Ready(Some(Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "Use socket.recv().await directly; Stream adapter requires storing an async future",
        ))))
    }
}

/// Adapter that implements `Sink` for any socket implementing the `Socket` trait.
///
/// This allows using ZeroMQ sockets with sink combinators like `send`, `send_all`, etc.
///
/// # Examples
///
/// ```no_run
/// use monocoque_zmtp::{DealerSocket, stream_sink::SocketSink};
/// use futures::SinkExt;
/// use bytes::Bytes;
///
/// # async fn example() -> std::io::Result<()> {
/// let socket = DealerSocket::connect("127.0.0.1:5555").await?;
/// let mut sink = SocketSink::new(socket);
///
/// // Use sink methods
/// sink.send(vec![Bytes::from("Hello")]).await?;
/// # Ok(())
/// # }
/// ```
pub struct SocketSink<S> {
    socket: S,
}

impl<S> SocketSink<S> {
    /// Create a new sink adapter for a socket.
    pub const fn new(socket: S) -> Self {
        Self { socket }
    }

    /// Get a reference to the underlying socket.
    pub const fn get_ref(&self) -> &S {
        &self.socket
    }

    /// Get a mutable reference to the underlying socket.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.socket
    }

    /// Consume the adapter and return the underlying socket.
    pub fn into_inner(self) -> S {
        self.socket
    }
}

impl<S: Socket + Unpin> Sink<Vec<Bytes>> for SocketSink<S> {
    type Error = io::Error;

    fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // ZeroMQ sockets are always ready to accept sends (they buffer internally)
        Poll::Ready(Ok(()))
    }

    fn start_send(self: Pin<&mut Self>, _item: Vec<Bytes>) -> Result<(), Self::Error> {
        // Cannot implement send without storing an async future across poll calls.
        // Return an error instead of silently dropping data.
        // Use socket.send(msg).await directly instead of this Sink adapter.
        Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "Use socket.send(msg).await directly; Sink adapter requires storing an async future",
        ))
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // Placeholder - full implementation would send buffered messages
        Poll::Ready(Ok(()))
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // Flush any pending messages and close
        self.poll_flush(cx)
    }
}

/// Combined Stream + Sink adapter for bidirectional sockets.
///
/// This provides both `Stream` and `Sink` implementations for sockets that support
/// both sending and receiving (DEALER, ROUTER, REQ, REP, PAIR).
///
/// # Examples
///
/// ```no_run
/// use monocoque_zmtp::{DealerSocket, stream_sink::SocketStreamSink};
/// use futures::{StreamExt, SinkExt};
/// use bytes::Bytes;
///
/// # async fn example() -> std::io::Result<()> {
/// let socket = DealerSocket::connect("127.0.0.1:5555").await?;
/// let mut stream_sink = SocketStreamSink::new(socket);
///
/// // Send a message
/// stream_sink.send(vec![Bytes::from("Hello")]).await?;
///
/// // Receive a response
/// if let Some(msg) = stream_sink.next().await {
///     println!("Response: {:?}", msg?);
/// }
/// # Ok(())
/// # }
/// ```
pub struct SocketStreamSink<S> {
    socket: S,
    pending_send: Option<Vec<Bytes>>,
}

impl<S> SocketStreamSink<S> {
    /// Create a new combined stream/sink adapter.
    pub const fn new(socket: S) -> Self {
        Self {
            socket,
            pending_send: None,
        }
    }

    /// Get a reference to the underlying socket.
    pub const fn get_ref(&self) -> &S {
        &self.socket
    }

    /// Get a mutable reference to the underlying socket.
    pub fn get_mut(&mut self) -> &mut S {
        &mut self.socket
    }

    /// Consume the adapter and return the underlying socket.
    pub fn into_inner(self) -> S {
        self.socket
    }
}

impl<S: Socket + Unpin> Stream for SocketStreamSink<S> {
    type Item = io::Result<Vec<Bytes>>;

    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // Stream::poll_next cannot store the async recv() future across poll calls
        // without a Pin-boxed future field. Return an explicit error instead of
        // silently hanging (Poll::Pending) so callers notice the limitation.
        // Use socket.recv().await directly instead of this Stream adapter.
        Poll::Ready(Some(Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "Use socket.recv().await directly; Stream adapter requires storing an async future",
        ))))
    }
}

impl<S: Socket + Unpin> Sink<Vec<Bytes>> for SocketStreamSink<S> {
    type Error = io::Error;

    fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn start_send(mut self: Pin<&mut Self>, item: Vec<Bytes>) -> Result<(), Self::Error> {
        // Store the item to make the pending_send field meaningful, but
        // poll_flush will return an error rather than silently discarding it.
        self.pending_send = Some(item);
        Ok(())
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // Cannot send without storing an async future across poll calls.
        // Return an error if there's a pending item instead of silently dropping it.
        if self.pending_send.is_some() {
            return Poll::Ready(Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "Use socket.send(msg).await directly; Sink adapter requires storing an async future",
            )));
        }
        Poll::Ready(Ok(()))
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.poll_flush(cx)
    }
}

#[cfg(test)]
mod tests {
    // Note: Full integration tests would require actual sockets
    // These are placeholder tests for the adapter structure

    #[test]
    fn test_stream_adapter_creation() {
        // We can't easily create a real socket in a unit test,
        // but we can test the adapter structure
        // Full tests should be in integration tests
    }
}