monocoque/zmq/

dealer.rs

//! DEALER socket implementation.

use super::common::{channel_to_io_error, parse_tcp_endpoint};
use bytes::Bytes;
use monocoque_core::monitor::{SocketEvent, SocketEventSender, SocketMonitor, create_monitor};
use monocoque_core::rt::TcpStream;
use monocoque_zmtp::dealer::DealerSocket as InternalDealer;
use std::io;

/// A DEALER socket for asynchronous request-reply patterns.
///
/// DEALER sockets are fair-queuing clients that distribute messages
/// across multiple server endpoints. They're used for:
///
/// - Load-balanced request-reply
/// - Async RPC clients
/// - Worker pools
///
/// ## ZeroMQ Compatibility
///
/// Compatible with `zmq::DEALER` and `zmq::ROUTER` sockets from libzmq.
///
/// ## Example
///
/// ```rust,no_run
/// use monocoque::zmq::DealerSocket;
/// use bytes::Bytes;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Connect to server
/// let mut socket = DealerSocket::connect("127.0.0.1:5555").await?;
///
/// // Send request
/// socket.send(vec![Bytes::from("REQUEST")]).await?;
///
/// // Receive reply
/// if let Ok(Some(reply)) = socket.recv().await {
///     println!("Got reply: {:?}", reply);
/// }
/// # Ok(())
/// # }
/// ```
pub struct DealerSocket<S = TcpStream>
where
    S: compio_io::AsyncRead + compio_io::AsyncWrite + Unpin,
{
    inner: InternalDealer<S>,
    monitor: Option<SocketEventSender>,
}

impl DealerSocket {
    /// Connect to a ZeroMQ peer and create a DEALER socket.
    ///
    /// Supports both TCP and IPC endpoints:
    /// - TCP: `"tcp://127.0.0.1:5555"` or `"127.0.0.1:5555"`
    /// - IPC: `"ipc:///tmp/socket.sock"` (Unix only)
    ///
    /// # Arguments
    ///
    /// * `endpoint` - Endpoint to connect to
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The connection fails (network unreachable, connection refused, etc.)
    /// - DNS resolution fails for TCP endpoints
    /// - Invalid endpoint format
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use monocoque::zmq::DealerSocket;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// // TCP connection
    /// let socket1 = DealerSocket::connect("tcp://127.0.0.1:5555").await?;
    ///
    /// // IPC connection (Unix only)
    /// #[cfg(unix)]
    /// let socket2 = DealerSocket::connect("ipc:///tmp/socket.sock").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect(endpoint: &str) -> io::Result<Self> {
        let addr = parse_tcp_endpoint(endpoint)?;
        let inner = InternalDealer::connect_with_options(
            addr,
            monocoque_core::options::SocketOptions::default(),
        )
        .await?;
        let sock = Self {
            inner,
            monitor: None,
        };
        sock.emit_event(SocketEvent::Connected(
            monocoque_core::endpoint::Endpoint::Tcp(addr),
        ));
        Ok(sock)
    }

    /// Connect to a ZeroMQ peer with custom socket options.
    ///
    /// Stores the endpoint so the socket can reconnect automatically after failures.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use monocoque::zmq::{DealerSocket, SocketOptions};
    /// use std::time::Duration;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut socket = DealerSocket::connect_with_options(
    ///     "tcp://127.0.0.1:5555",
    ///     SocketOptions::default().with_send_hwm(100),
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect_with_options(
        endpoint: &str,
        options: monocoque_core::options::SocketOptions,
    ) -> io::Result<Self> {
        let addr = parse_tcp_endpoint(endpoint)?;
        let inner = InternalDealer::connect_with_options(addr, options).await?;
        let sock = Self {
            inner,
            monitor: None,
        };
        sock.emit_event(SocketEvent::Connected(
            monocoque_core::endpoint::Endpoint::Tcp(addr),
        ));
        Ok(sock)
    }

    /// Connect to a ZeroMQ peer via IPC (Unix domain sockets).
    ///
    /// Unix-only. Accepts IPC paths with or without `ipc://` prefix.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # #[cfg(unix)]
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use monocoque::zmq::DealerSocket;
    ///
    /// let mut socket = DealerSocket::connect_ipc("/tmp/dealer.sock").await?;
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(unix)]
    pub async fn connect_ipc(
        path: &str,
    ) -> io::Result<DealerSocket<monocoque_core::rt::UnixStream>> {
        use std::path::PathBuf;

        let clean_path = path.strip_prefix("ipc://").unwrap_or(path);
        let ipc_path = PathBuf::from(clean_path);

        let stream = monocoque_core::ipc::connect(&ipc_path).await?;
        let sock = DealerSocket::from_unix_stream(stream).await?;
        sock.emit_event(SocketEvent::Connected(
            monocoque_core::endpoint::Endpoint::Ipc(ipc_path),
        ));
        Ok(sock)
    }

    /// Bind to an address and accept the first connection.
    ///
    /// This creates a server-side DEALER socket that accepts incoming connections.
    /// Useful for broker patterns where workers (REP sockets) connect to a DEALER backend.
    ///
    /// # Returns
    ///
    /// A tuple of `(listener, socket)` where:
    /// - `listener` can be used to accept additional connections
    /// - `socket` is ready to send/receive with the first peer
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The address is already in use
    /// - Permission denied (e.g., binding to privileged port without root)
    /// - Invalid address format
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use monocoque::zmq::DealerSocket;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// // Bind DEALER backend for worker connections
    /// let (listener, socket) = DealerSocket::bind("127.0.0.1:5556").await?;
    ///
    /// // Use socket for first connection
    /// // Accept more connections from listener if needed:
    /// // let (stream, _) = listener.accept().await?;
    /// // let socket2 = DealerSocket::from_tcp(stream).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn bind(
        addr: impl monocoque_core::rt::ToSocketAddrs,
    ) -> io::Result<(monocoque_core::rt::TcpListener, Self)> {
        let listener = monocoque_core::rt::TcpListener::bind(addr).await?;
        let (stream, _) = listener.accept().await?;
        let socket = Self::from_tcp(stream).await?;
        Ok((listener, socket))
    }

    /// Create a DEALER socket from a TCP stream with TCP_NODELAY enabled.
    ///
    /// This method automatically enables TCP_NODELAY for optimal performance,
    /// preventing Nagle's algorithm from buffering small packets.
    ///
    /// Uses default buffer sizes (8KB) and socket options. For custom configuration,
    /// use `with_options()`.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use monocoque::zmq::DealerSocket;
    /// use monocoque_core::rt::TcpStream;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let stream = TcpStream::connect("127.0.0.1:5555").await?;
    /// let socket = DealerSocket::from_tcp(stream).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn from_tcp(stream: TcpStream) -> io::Result<Self> {
        Ok(Self {
            inner: InternalDealer::from_tcp(stream).await?,
            monitor: None,
        })
    }

    /// Create a DEALER socket from a TCP stream with custom options.
    ///
    /// Provides full control over buffer sizes, HWM, timeouts, etc. through SocketOptions.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use monocoque::zmq::{DealerSocket, SocketOptions};
    /// use monocoque_core::rt::TcpStream;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let stream = TcpStream::connect("127.0.0.1:5555").await?;
    ///
    /// // Customize HWM only (uses default 8KB buffers)
    /// let socket = DealerSocket::from_tcp_with_options(
    ///     stream,
    ///     SocketOptions::default().with_send_hwm(100)
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// ```rust,no_run
    /// # use monocoque::zmq::{DealerSocket, SocketOptions};
    /// # use monocoque_core::rt::TcpStream;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let stream = TcpStream::connect("127.0.0.1:5555").await?;
    /// // Customize both buffers and HWM
    /// let socket = DealerSocket::from_tcp_with_options(
    ///     stream,
    ///     SocketOptions::default()
    ///         .with_buffer_sizes(4096, 4096)  // 4KB buffers for low latency
    ///         .with_send_hwm(100)
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn from_tcp_with_options(
        stream: TcpStream,
        options: monocoque_core::options::SocketOptions,
    ) -> io::Result<Self> {
        Ok(Self {
            inner: InternalDealer::from_tcp_with_options(stream, options).await?,
            monitor: None,
        })
    }

    /// Create a DEALER socket from any stream with custom options.
    ///
    /// This is the most flexible constructor - works with TCP, Unix, or in-memory streams.
    /// Useful for testing with duplex streams.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use monocoque::zmq::{DealerSocket, SocketOptions};
    /// use monocoque_core::rt::TcpStream;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let stream = TcpStream::connect("127.0.0.1:5555").await?;
    /// let socket = DealerSocket::with_options(
    ///     stream,
    ///     SocketOptions::default().with_send_hwm(10)
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn with_options<Stream>(
        stream: Stream,
        options: monocoque_core::options::SocketOptions,
    ) -> io::Result<DealerSocket<Stream>>
    where
        Stream: compio_io::AsyncRead + compio_io::AsyncWrite + Unpin,
    {
        Ok(DealerSocket {
            inner: InternalDealer::with_options(stream, options).await?,
            monitor: None,
        })
    }

    /// Try to reconnect to the stored endpoint.
    ///
    /// Only sockets built with [`connect`](Self::connect) or
    /// [`connect_with_options`](Self::connect_with_options) store an endpoint;
    /// one built from a raw stream has nothing to reconnect to.
    ///
    /// # Errors
    ///
    /// Returns an error if no endpoint is stored or the reconnection fails.
    pub async fn try_reconnect(&mut self) -> io::Result<()> {
        self.inner.try_reconnect().await
    }

    /// Send with automatic reconnection on network error.
    ///
    /// Retries up to `max_reconnect_attempts`, applying the configured backoff
    /// between attempts.
    ///
    /// # Errors
    ///
    /// Returns an error if the send fails and reconnection does not recover it.
    pub async fn send_with_reconnect(&mut self, msg: Vec<Bytes>) -> io::Result<()> {
        self.inner.send_with_reconnect(msg).await
    }

    /// Receive with automatic reconnection on EOF or network error.
    ///
    /// # Errors
    ///
    /// Returns an error if the receive fails and reconnection does not recover
    /// it.
    pub async fn recv_with_reconnect(&mut self) -> io::Result<Option<Vec<Bytes>>> {
        self.inner.recv_with_reconnect().await
    }
}

// Generic impl - works with any stream type
impl<S> DealerSocket<S>
where
    S: compio_io::AsyncRead + compio_io::AsyncWrite + Unpin,
{
    /// Enable monitoring for this socket.
    ///
    /// Returns a receiver for socket lifecycle events. Once enabled, the socket
    /// will emit events like Connected, Disconnected, etc.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use monocoque::zmq::{DealerSocket, SocketEvent};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut socket = DealerSocket::connect("127.0.0.1:5555").await?;
    /// let monitor = socket.monitor();
    ///
    /// // Spawn task to handle events
    /// monocoque::rt::spawn_detached(async move {
    ///     while let Ok(event) = monitor.recv_async().await {
    ///         println!("Socket event: {}", event);
    ///     }
    /// });
    /// # Ok(())
    /// # }
    /// ```
    pub fn monitor(&mut self) -> SocketMonitor {
        let (sender, receiver) = create_monitor();
        self.monitor = Some(sender);
        receiver
    }

    /// Helper to emit monitoring events (if monitoring is enabled).
    fn emit_event(&self, event: SocketEvent) {
        if let Some(monitor) = &self.monitor {
            monocoque_core::monitor::emit(monitor, event);
        }
    }

    /// Send a multipart message.
    ///
    /// Messages are sent asynchronously - this returns immediately after
    /// queuing the message for transmission.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying connection is closed or broken.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use monocoque::zmq::DealerSocket;
    /// # use bytes::Bytes;
    /// # async fn example(mut socket: DealerSocket) -> Result<(), Box<dyn std::error::Error>> {
    /// socket.send(vec![
    ///     Bytes::from("part1"),
    ///     Bytes::from("part2"),
    /// ]).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn send(&mut self, msg: Vec<Bytes>) -> io::Result<()> {
        channel_to_io_error(self.inner.send(msg).await)
    }

    /// Send a message to the internal buffer without flushing.
    ///
    /// Use this for batching multiple messages before a single flush.
    /// Call `flush()` to send all buffered messages.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use monocoque::zmq::DealerSocket;
    /// # use bytes::Bytes;
    /// # async fn example(mut socket: DealerSocket) -> Result<(), Box<dyn std::error::Error>> {
    /// // Batch 100 messages
    /// for i in 0..100 {
    ///     socket.send_buffered(vec![Bytes::from(format!("msg {}", i))])?;
    /// }
    /// // Single I/O operation for all 100 messages
    /// socket.flush().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn send_buffered(&mut self, msg: Vec<Bytes>) -> io::Result<()> {
        channel_to_io_error(self.inner.send_buffered(msg))
    }

    /// Flush all buffered messages to the network.
    ///
    /// Sends all messages buffered by `send_buffered()` in a single I/O operation.
    pub async fn flush(&mut self) -> io::Result<()> {
        channel_to_io_error(self.inner.flush().await)
    }

    /// Send multiple messages in a single batch (convenience method).
    ///
    /// This is equivalent to calling `send_buffered()` for each message
    /// followed by `flush()`, but more ergonomic.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use monocoque::zmq::DealerSocket;
    /// # use bytes::Bytes;
    /// # async fn example(mut socket: DealerSocket) -> Result<(), Box<dyn std::error::Error>> {
    /// let messages = vec![
    ///     vec![Bytes::from("msg1")],
    ///     vec![Bytes::from("msg2")],
    ///     vec![Bytes::from("msg3")],
    /// ];
    /// socket.send_batch(&messages).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn send_batch(&mut self, messages: &[Vec<Bytes>]) -> io::Result<()> {
        channel_to_io_error(self.inner.send_batch(messages).await)
    }

    /// Get the number of bytes currently buffered.
    #[inline]
    pub fn buffered_bytes(&self) -> usize {
        self.inner.buffered_bytes()
    }

    /// Get the socket type.
    ///
    /// # ZeroMQ Compatibility
    /// Get current socket events (read/write readiness).
    ///
    /// Returns a bitmask:
    /// - `1` (POLLIN): Can receive without blocking
    /// - `2` (POLLOUT): Can send without blocking
    ///
    /// # ZeroMQ Compatibility
    ///
    /// Corresponds to `ZMQ_EVENTS` (15) option.
    pub fn events(&self) -> u32 {
        self.inner.events()
    }

    /// Receive a multipart message.
    ///
    /// Returns `None` if the connection is closed.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use monocoque::zmq::DealerSocket;
    /// # async fn example(mut socket: DealerSocket) -> Result<(), Box<dyn std::error::Error>> {
    /// while let Ok(Some(msg)) = socket.recv().await {
    ///     println!("Received {} parts", msg.len());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn recv(&mut self) -> io::Result<Option<Vec<Bytes>>> {
        self.inner.recv().await
    }

    /// Receive a message into a caller-provided buffer, reusing its allocation.
    ///
    /// Allocation-free counterpart to [`recv`](Self::recv). Returns `Ok(true)`
    /// when a complete message was read into `out` (cleared on entry), or
    /// `Ok(false)` on a closed connection.
    pub async fn recv_into(&mut self, out: &mut Vec<Bytes>) -> io::Result<bool> {
        self.inner.recv_into(out).await
    }

    /// Try to receive a message into `out` without a kernel read.
    pub fn try_recv_into(&mut self, out: &mut Vec<Bytes>) -> io::Result<bool> {
        self.inner.try_recv_into(out)
    }

    /// Receive a single-frame message, returning just its frame.
    pub async fn recv_one(&mut self) -> io::Result<Option<Bytes>> {
        self.inner.recv_one().await
    }
}

impl<S> DealerSocket<S>
where
    S: compio_io::AsyncRead + compio_io::AsyncWrite + Unpin,
{
    /// Get the socket type.
    ///
    /// Always returns `SocketType::Dealer` for DEALER sockets.
    ///
    /// # ZeroMQ Compatibility
    ///
    /// Corresponds to `ZMQ_TYPE` (16) socket option.
    #[inline]
    pub const fn socket_type(&self) -> monocoque_zmtp::session::SocketType {
        self.inner.socket_type()
    }

    /// Get the last connected endpoint as a string.
    ///
    /// Returns the endpoint this socket connected to, if any.
    ///
    /// # ZeroMQ Compatibility
    ///
    /// Corresponds to `ZMQ_LAST_ENDPOINT` (32) socket option.
    #[inline]
    pub fn last_endpoint(&self) -> Option<&str> {
        self.inner.last_endpoint_string()
    }

    /// Check if more message frames are expected.
    ///
    /// For multipart messages, this indicates if more frames follow.
    ///
    /// # ZeroMQ Compatibility
    ///
    /// Corresponds to `ZMQ_RCVMORE` (13) socket option.
    #[inline]
    pub fn has_more(&self) -> bool {
        self.inner.has_more()
    }

    /// Get mutable access to socket options.
    ///
    /// Allows runtime modification of socket behavior.
    #[inline]
    pub fn options_mut(&mut self) -> &mut monocoque_core::options::SocketOptions {
        self.inner.options_mut()
    }

    /// Get immutable access to socket options.
    #[inline]
    pub const fn options(&self) -> &monocoque_core::options::SocketOptions {
        self.inner.options()
    }
}

// Unix-specific impl for IPC support
#[cfg(unix)]
impl DealerSocket<monocoque_core::rt::UnixStream> {
    /// Create a DEALER socket from an existing Unix domain socket stream (IPC).
    pub async fn from_unix_stream(stream: monocoque_core::rt::UnixStream) -> io::Result<Self> {
        Ok(Self {
            inner: InternalDealer::new(stream).await?,
            monitor: None,
        })
    }

    /// Create a DEALER socket from an existing Unix stream with custom options.
    pub async fn from_unix_stream_with_options(
        stream: monocoque_core::rt::UnixStream,
        options: monocoque_core::options::SocketOptions,
    ) -> io::Result<Self> {
        Ok(Self {
            inner: InternalDealer::with_options(stream, options).await?,
            monitor: None,
        })
    }
}

// Implement ProxySocket for the high-level DealerSocket wrapper. With native
// async-fn-in-trait on ProxySocket this is a plain forward, no hand-written
// Box::pin desugaring or per-call heap allocation.
impl monocoque_zmtp::proxy::ProxySocket for DealerSocket<TcpStream> {
    async fn recv_multipart(&mut self) -> io::Result<Option<Vec<Bytes>>> {
        self.recv().await
    }

    async fn send_multipart(&mut self, msg: Vec<Bytes>) -> io::Result<()> {
        self.send(msg).await
    }

    fn socket_desc(&self) -> &'static str {
        "DEALER"
    }
}