rustzmq2 0.1.0

//! Per-socket configuration: identity, heartbeats, security mechanism.

use crate::codec::mechanism::ZmqMechanism;
use crate::PeerIdentity;

use std::collections::HashMap;

bitflags::bitflags! {
    /// Conditions under which the reconnect task gives up rather than
    /// retrying. Mirrors libzmq's `ZMQ_RECONNECT_STOP` flag set.
    ///
    /// Combine with `|` and pass to
    /// [`SocketBuilder::reconnect_stop`](crate::SocketBuilder::reconnect_stop):
    ///
    /// ```rust,no_run
    /// use rustzmq2::prelude::*;
    ///
    /// let _sock = rustzmq2::DealerSocket::builder()
    ///     .reconnect_stop(ReconnectStop::CONN_REFUSED | ReconnectStop::HANDSHAKE_FAILED)
    ///     .build();
    /// ```
    #[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Hash)]
    pub struct ReconnectStop: u32 {
        /// Stop reconnecting when the connection is refused (`ECONNREFUSED`).
        const CONN_REFUSED = 0x1;
        /// Stop reconnecting when the ZMTP handshake fails (bad mechanism,
        /// timeout, ZAP denial).
        const HANDSHAKE_FAILED = 0x2;
        /// Stop reconnecting once a connected peer has disconnected.
        const AFTER_DISCONNECT = 0x4;
    }
}

/// Per-socket configuration passed to [`Socket::with_options`](crate::Socket::with_options).
///
/// Prefer [`SocketBuilder`] (via [`Socket::builder`](crate::Socket::builder)) for a
/// more ergonomic fluent API.
#[derive(Debug, Clone)]
pub struct SocketOptions {
    pub(crate) peer_id: Option<PeerIdentity>,

    // ── Heartbeat ───────────────────────────────────────────────────────────
    pub(crate) heartbeat_interval: Option<std::time::Duration>,
    pub(crate) heartbeat_timeout: Option<std::time::Duration>,
    pub(crate) heartbeat_ttl: Option<std::time::Duration>,

    // ── Security ────────────────────────────────────────────────────────────
    pub(crate) mechanism: ZmqMechanism,

    // ── PLAIN ───────────────────────────────────────────────────────────────
    pub(crate) plain_username: Option<bytes::Bytes>,
    pub(crate) plain_password: Option<bytes::Bytes>,
    pub(crate) plain_server: bool,

    // ── CURVE ───────────────────────────────────────────────────────────────
    #[cfg(feature = "curve")]
    pub(crate) curve_server_key: Option<[u8; 32]>,
    #[cfg(feature = "curve")]
    pub(crate) curve_public_key: Option<[u8; 32]>,
    #[cfg(feature = "curve")]
    pub(crate) curve_secret_key: Option<[u8; 32]>,
    #[cfg(feature = "curve")]
    pub(crate) curve_server: bool,

    // ── ZAP ─────────────────────────────────────────────────────────────────
    pub(crate) zap_domain: Option<String>,

    // ── High-water marks ────────────────────────────────────────────────────
    /// Per-peer outbound channel bound (default 1000, matches libzmq `send_hwm`).
    pub(crate) send_hwm: usize,
    /// Shared inbound channel capacity (default 1000, matches libzmq `receive_hwm`).
    pub(crate) receive_hwm: usize,

    // ── Message size ────────────────────────────────────────────────────────
    /// Reject inbound messages exceeding this many bytes (default: unlimited).
    pub(crate) max_msg_size: Option<usize>,

    // ── Receive / send timeouts ──────────────────────────────────────────────
    /// Timeout for `recv()` calls (default: block forever).
    pub(crate) receive_timeout: Option<std::time::Duration>,
    /// Timeout for `send()` calls (default: block forever).
    pub(crate) send_timeout: Option<std::time::Duration>,

    // ── Reconnect ───────────────────────────────────────────────────────────
    /// Initial reconnect interval (default 100 ms).
    pub(crate) reconnect_interval: std::time::Duration,
    /// Maximum reconnect interval (default 30 s). Set equal to `reconnect_interval`
    /// to disable exponential backoff.
    pub(crate) reconnect_interval_max: std::time::Duration,

    // ── TCP / connection ────────────────────────────────────────────────────
    /// TCP connect timeout (default: OS default, i.e. no explicit timeout).
    pub(crate) connect_timeout: Option<std::time::Duration>,
    /// TCP listen backlog (default 100, matches libzmq).
    pub(crate) backlog: u32,
    /// Enable IPv6 dual-stack listener / IPv6 connects (default false).
    pub(crate) ipv6: bool,

    // ── Handshake ───────────────────────────────────────────────────────────
    /// Max time a peer has to complete the ZMTP greeting + mechanism + READY
    /// exchange. Disconnect on expiry. `None` disables the timeout. Default
    /// matches libzmq (30 seconds).
    pub(crate) handshake_interval: Option<std::time::Duration>,

    // ── Invert matching ─────────────────────────────────────────────────────
    /// When true on PUB/XPUB, messages are delivered to subscribers whose
    /// filters do NOT match the message topic. Default false (normal match).
    pub(crate) invert_matching: bool,

    // ── XPUB tweaks ─────────────────────────────────────────────────────────
    /// XPUB: when `true`, every SUBSCRIBE/UNSUBSCRIBE is surfaced via
    /// `recv`, including duplicates. When `false` (default, matching
    /// libzmq's `ZMQ_XPUB_VERBOSE=0`) only the first SUBSCRIBE for a
    /// given topic and the UNSUBSCRIBE that drops it are surfaced;
    /// redundant duplicates are absorbed by the backend.
    pub(crate) xpub_verbose: bool,

    // ── Metadata ────────────────────────────────────────────────────────────
    /// Custom application metadata attached to the ZMTP READY command and
    /// visible to the peer. Reserved names (`Socket-Type`, `Identity`,
    /// `Resource`, `User-Id`) are rejected by the builder.
    pub(crate) metadata: HashMap<String, bytes::Bytes>,

    // ── Auto-sent messages ──────────────────────────────────────────────────
    /// Message auto-enqueued on every new peer after the handshake completes.
    pub(crate) hello_msg: Option<crate::message::ZmqMessage>,
    /// Message auto-sent just before the peer is dropped (best-effort).
    pub(crate) disconnect_msg: Option<crate::message::ZmqMessage>,
    /// SUB only: message synthesized into the inbound channel when a
    /// previously-connected publisher reconnects.
    pub(crate) hiccup_msg: Option<crate::message::ZmqMessage>,

    // ── Reconnect-stop bitmask ──────────────────────────────────────────────
    /// Bitmask of conditions under which the reconnect task bails out
    /// rather than retrying. See [`ReconnectStop`].
    pub(crate) reconnect_stop: ReconnectStop,

    // ── SOCKS5 proxy ────────────────────────────────────────────────────────
    /// `host:port` of a SOCKS5 proxy to connect through. `None` = direct.
    pub(crate) socks_proxy: Option<String>,
    /// Optional SOCKS5 username (user/pass auth).
    pub(crate) socks_username: Option<String>,
    /// Optional SOCKS5 password.
    pub(crate) socks_password: Option<String>,

    // ── Batch sizes ─────────────────────────────────────────────────────────
    /// Max payload bytes the writer task drains from the outbound channel
    /// into one `writev(2)`. `None` means "no byte budget — only the
    /// message-count ceiling applies". Default `Some(8192)`, matching
    /// libzmq's `ZMQ_OUT_BATCH_SIZE`.
    ///
    /// The semantics match libzmq: we accumulate payload bytes from the
    /// outbound channel until the running total ≥ this value, then flush.
    /// At least one message is always drained per wake even if it alone
    /// exceeds the budget.
    pub(crate) out_batch_size: Option<usize>,
    /// Hard ceiling on messages drained per batch, independent of
    /// `out_batch_size`.
    ///
    /// Outer `None` means "use the per-socket-type default"
    /// (`SocketType::default_out_batch_msgs`); outer `Some(inner)`
    /// means the user set it explicitly:
    /// - `Some(None)` — no message-count ceiling (rely solely on
    ///   `out_batch_size`).
    /// - `Some(Some(n))` — drain at most `n` messages per writev.
    ///
    /// Per-type defaults (applied at `SocketCore::new` when this is
    /// `None`):
    /// - PUB / XPUB: `Some(32)` — caps per-peer pending depth so
    ///   high-fanout small-message workloads don't overflow HWM
    ///   under burst (see commit 2192266f).
    /// - Everything else: `Some(256)` — point-to-point sockets only
    ///   have one peer in flight per send; raising the cap to
    ///   `IOV_CAP/2` amortizes the writev syscall ~8× more than the
    ///   PUB default would.
    ///
    /// At least one of `out_batch_size` / `out_batch_msgs` should be
    /// `Some`; if both are `None` the drain only stops when the
    /// channel empties, which may let the pending queue grow
    /// unbounded under sustained load.
    #[allow(clippy::option_option)]
    pub(crate) out_batch_msgs: Option<Option<usize>>,
    /// Max messages the reader drains from the framed decoder per wake,
    /// before yielding control back to the peer-loop select. `None`
    /// disables batching — one message per wake. Default `None`.
    ///
    /// Unlike libzmq's byte-based `ZMQ_IN_BATCH_SIZE`, this is a
    /// message-count limit: the codec already buffers socket reads
    /// internally, so what we're controlling here is how many decoded
    /// messages we forward into the inbound channel before letting other
    /// select arms (shutdown, heartbeat, outbound) fire.
    ///
    /// Tradeoff: batched reads save one scheduler round-trip per
    /// message on read-heavy bursts, but starve the writer arm of the
    /// peer-loop select on mixed RTT workloads — measured ~15-20%
    /// regression on `pub_xsub tcp` with `Some(32)`. Default is off;
    /// users with known reader-heavy patterns (PULL behind a fast
    /// producer, etc.) can opt in.
    pub(crate) in_batch_msgs: Option<usize>,

    // ── Inline-write fast path ──────────────────────────────────────────────
    /// Caller-thread inline-write: when enabled, `send`/`send_flushed`
    /// attempt to encode + write the message directly from the caller's
    /// thread (one `try_write_vectored` syscall) before bouncing through
    /// the outbound channel + peer loop. Saves one task wake per RTT at
    /// the cost of bypassing `drain_batch` coalescing, which is
    /// throughput-toxic on patterns where the sender outpaces the wire
    /// (PUB fanout, PUSH workers).
    ///
    /// Outer `None` means "use the per-socket-type default"
    /// (`SocketType::default_inline_write_max`). Outer `Some(inner)`
    /// means the user set it explicitly:
    /// - `Some(None)` — explicitly disabled
    /// - `Some(Some(0))` — explicitly enabled, uncapped
    /// - `Some(Some(n))` — explicitly enabled with payload cap `n` bytes
    ///
    /// `SocketCore::new` resolves this to a concrete `Option<usize>`
    /// applied to every spawned `PeerEngine`.
    #[allow(clippy::option_option)]
    pub(crate) inline_write_max: Option<Option<usize>>,

    // ── Linger ──────────────────────────────────────────────────────────────
    /// How long to wait for outbound messages to drain on drop.
    /// `None` = block indefinitely (TODO: blocking linger is a follow-up).
    /// `Some(ZERO)` = discard immediately (current default behaviour).
    pub(crate) linger: Option<std::time::Duration>,

    // ── Conflate ─────────────────────────────────────────────────────────────
    /// Keep only the most recent inbound message per peer (default false).
    pub(crate) conflate: bool,

    // ── Immediate ───────────────────────────────────────────────────────────
    /// When true, `send()` returns `ReturnToSender` immediately when no peers
    /// are connected, rather than blocking until a peer arrives (default false).
    pub(crate) immediate: bool,

    // ── TCP socket knobs ────────────────────────────────────────────────────
    /// `SO_SNDBUF` (OS-level TCP send buffer size, bytes). `None` = OS default.
    pub(crate) tcp_send_buffer: Option<usize>,
    /// `SO_RCVBUF` (OS-level TCP receive buffer size, bytes). `None` = OS default.
    pub(crate) tcp_receive_buffer: Option<usize>,
    /// Enable `SO_KEEPALIVE`. `None` = OS default.
    pub(crate) tcp_keepalive: Option<bool>,
    /// `TCP_KEEPIDLE`: time before sending the first keepalive probe.
    pub(crate) tcp_keepalive_idle: Option<std::time::Duration>,
    /// `TCP_KEEPINTVL`: interval between subsequent keepalive probes.
    pub(crate) tcp_keepalive_interval: Option<std::time::Duration>,
    /// `TCP_KEEPCNT`: max failed keepalive probes before giving up.
    pub(crate) tcp_keepalive_count: Option<u32>,
    /// `TCP_USER_TIMEOUT` (Linux) / `TCP_MAXRT` (Windows).
    pub(crate) tcp_max_retransmit: Option<std::time::Duration>,
    /// IP `TOS` (DSCP bits). `None` = OS default.
    pub(crate) type_of_service: Option<u32>,
    /// Linux `SO_BINDTODEVICE`. Ignored on other platforms.
    pub(crate) bind_to_device: Option<String>,
}

#[cfg(feature = "inproc")]
impl SocketOptions {
    /// Returns the list of option names that are set on this `SocketOptions`
    /// but have no effect over the inproc transport. Used by `bind`/`connect`
    /// to emit one `log::warn!` + `SocketEvent::OptionIgnoredOnTransport`
    /// per (option, socket) pair when the caller routes through inproc.
    ///
    /// libzmq also silently ignores these options on inproc
    /// (`libzmq/src/socket_base.cpp:559-568` for bind, `:823-920` for connect;
    /// no auth branches). We warn rather than erroring because a single
    /// socket can legitimately span transports — a DEALER with CURVE
    /// configured might connect a TCP peer *and* an inproc peer, and only
    /// the TCP side uses CURVE. Returning `Err` would break that pattern.
    ///
    /// Stays `&'static str` so the set can be stored as `HashSet<&'static str>`
    /// for de-dup without clone churn.
    pub(crate) fn inproc_ignored_options(&self) -> Vec<&'static str> {
        let mut v: Vec<&'static str> = Vec::new();

        if !matches!(self.mechanism, ZmqMechanism::NULL) {
            v.push("mechanism");
        }
        if self.plain_server || self.plain_username.is_some() || self.plain_password.is_some() {
            v.push("plain_*");
        }
        #[cfg(feature = "curve")]
        {
            if self.curve_server
                || self.curve_public_key.is_some()
                || self.curve_secret_key.is_some()
                || self.curve_server_key.is_some()
            {
                v.push("curve_*");
            }
        }
        if self.zap_domain.is_some() {
            v.push("zap_domain");
        }
        if self.handshake_interval.is_some() {
            v.push("handshake_interval");
        }
        if self.heartbeat_interval.is_some()
            || self.heartbeat_timeout.is_some()
            || self.heartbeat_ttl.is_some()
        {
            v.push("heartbeat_*");
        }
        if !self.metadata.is_empty() {
            v.push("metadata");
        }
        if self.socks_proxy.is_some() {
            v.push("socks_*");
        }
        if self.tcp_send_buffer.is_some()
            || self.tcp_receive_buffer.is_some()
            || self.tcp_keepalive.is_some()
            || self.tcp_keepalive_idle.is_some()
            || self.tcp_keepalive_interval.is_some()
            || self.tcp_keepalive_count.is_some()
            || self.tcp_max_retransmit.is_some()
            || self.type_of_service.is_some()
            || self.bind_to_device.is_some()
        {
            v.push("tcp_*");
        }
        if self.connect_timeout.is_some() {
            v.push("connect_timeout");
        }
        if self.ipv6 {
            v.push("ipv6");
        }
        // backlog default is 100; flag only when the user changed it.
        if self.backlog != 100 {
            v.push("backlog");
        }
        // reconnect_* don't apply — inproc connections don't reconnect. We
        // don't flag the defaults (100ms / 30s) since every socket has them;
        // only flag an explicit disable via reconnect_stop.
        if !self.reconnect_stop.is_empty() {
            v.push("reconnect_stop");
        }
        v
    }
}

impl Default for SocketOptions {
    fn default() -> Self {
        Self {
            peer_id: None,
            heartbeat_interval: None,
            heartbeat_timeout: None,
            heartbeat_ttl: None,
            mechanism: ZmqMechanism::default(),
            plain_username: None,
            plain_password: None,
            plain_server: false,
            #[cfg(feature = "curve")]
            curve_server_key: None,
            #[cfg(feature = "curve")]
            curve_public_key: None,
            #[cfg(feature = "curve")]
            curve_secret_key: None,
            #[cfg(feature = "curve")]
            curve_server: false,
            zap_domain: None,
            send_hwm: 1000,
            receive_hwm: 1000,
            max_msg_size: None,
            receive_timeout: None,
            send_timeout: None,
            reconnect_interval: std::time::Duration::from_millis(100),
            reconnect_interval_max: std::time::Duration::from_secs(30),
            connect_timeout: None,
            backlog: 100,
            ipv6: false,
            handshake_interval: Some(std::time::Duration::from_secs(30)),
            invert_matching: false,
            xpub_verbose: false,
            metadata: HashMap::new(),
            hello_msg: None,
            disconnect_msg: None,
            hiccup_msg: None,
            reconnect_stop: ReconnectStop::empty(),
            socks_proxy: None,
            socks_username: None,
            socks_password: None,
            out_batch_size: Some(8192),
            // `None` here = "use type default", resolved at SocketCore::new.
            out_batch_msgs: None,
            // `None` = no reader batching. Batched reads (budget N > 1)
            // starve the writer arm of the peer-loop select on mixed
            // RTT workloads — measured ~15-20% regression on pub_xsub
            // tcp with `Some(32)`. The knob is retained so users with
            // known reader-heavy patterns can opt in.
            in_batch_msgs: None,
            // `None` here = "use type default", resolved at SocketCore::new.
            inline_write_max: None,
            linger: Some(std::time::Duration::ZERO),
            conflate: false,
            immediate: false,
            tcp_send_buffer: None,
            tcp_receive_buffer: None,
            tcp_keepalive: None,
            tcp_keepalive_idle: None,
            tcp_keepalive_interval: None,
            tcp_keepalive_count: None,
            tcp_max_retransmit: None,
            type_of_service: None,
            bind_to_device: None,
        }
    }
}

// ── SocketBuilder ────────────────────────────────────────────────────────────

/// Fluent builder for any socket type.
///
/// Obtain one via [`Socket::builder`](crate::Socket::builder), configure it
/// with the chainable methods below, then call [`build`](SocketBuilder::build)
/// to construct the socket.
///
/// ```rust,ignore
/// let rep = RepSocket::builder()
///     .curve_server(pub_key, sec_key)
///     .zap_domain("my-app")
///     .heartbeat_interval(std::time::Duration::from_secs(30))
///     .build();
/// ```
///
/// Option names match libzmq conventions where possible — see
/// [`zmq_setsockopt(3)`](https://libzmq.readthedocs.io/en/latest/zmq_setsockopt.html) for
/// the canonical semantics of each option.
#[must_use = "SocketBuilder is a no-op until you call .build()"]
#[derive(Debug, Clone)]
pub struct SocketBuilder<S> {
    options: SocketOptions,
    _marker: std::marker::PhantomData<fn() -> S>,
}

impl<S: super::Socket> SocketBuilder<S> {
    pub(crate) fn new() -> Self {
        Self {
            options: SocketOptions::default(),
            _marker: std::marker::PhantomData,
        }
    }

    /// Consume the builder and create the socket.
    pub fn build(self) -> S {
        S::with_options(self.options)
    }

    /// Set a custom peer identity (up to 255 bytes). Used as the routing ID
    /// in ROUTER sockets and visible to the remote peer.
    pub fn peer_identity(mut self, peer_id: PeerIdentity) -> Self {
        self.options.peer_id = Some(peer_id);
        self
    }

    /// How often to send a ZMTP `PING` to an otherwise-idle peer.
    /// Without heartbeats a half-open TCP connection (NAT timeout, peer
    /// crashed, cable yanked) can sit broken indefinitely.
    ///
    /// Pair with [`heartbeat_timeout`](Self::heartbeat_timeout) to drop
    /// peers that stop responding. Disabled by default.
    ///
    /// ```rust,no_run
    /// use rustzmq2::Socket;
    /// use std::time::Duration;
    ///
    /// // Probe every 30s; tear down if no response within 10s.
    /// let _sock = rustzmq2::DealerSocket::builder()
    ///     .heartbeat_interval(Duration::from_secs(30))
    ///     .heartbeat_timeout(Duration::from_secs(10))
    ///     .build();
    /// ```
    pub fn heartbeat_interval(mut self, interval: std::time::Duration) -> Self {
        self.options.heartbeat_interval = Some(interval);
        self
    }

    /// How long to wait for a `PONG` after sending a `PING` before
    /// considering the peer dead and disconnecting it. Only meaningful
    /// alongside [`heartbeat_interval`](Self::heartbeat_interval).
    pub fn heartbeat_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.options.heartbeat_timeout = Some(timeout);
        self
    }

    /// TTL we *advertise to the peer* in our outgoing `PING` (capped at
    /// 6553.5 s by ZMTP). Tells the peer "if you don't hear from me
    /// within this long, you can declare me dead." Independent of how
    /// *we* time out the peer (that's [`heartbeat_timeout`](Self::heartbeat_timeout)).
    pub fn heartbeat_ttl(mut self, ttl: std::time::Duration) -> Self {
        self.options.heartbeat_ttl = Some(ttl);
        self
    }

    /// Configure PLAIN client credentials. Sets mechanism to PLAIN.
    ///
    /// PLAIN sends the username/password in cleartext — safe only over IPC,
    /// loopback, or a separately encrypted channel. See [RFC 24].
    ///
    /// ```rust,ignore
    /// use bytes::Bytes;
    /// use rustzmq2::prelude::*;
    ///
    /// let mut dealer = rustzmq2::DealerSocket::builder()
    ///     .plain_client(Bytes::from_static(b"alice"), Bytes::from_static(b"s3cr3t"))
    ///     .build();
    /// dealer.connect("tcp://127.0.0.1:5555").await?;
    /// ```
    ///
    /// [RFC 24]: https://rfc.zeromq.org/spec/24/
    pub fn plain_client(
        mut self,
        username: impl Into<bytes::Bytes>,
        password: impl Into<bytes::Bytes>,
    ) -> Self {
        self.options.mechanism = ZmqMechanism::PLAIN;
        self.options.plain_username = Some(username.into());
        self.options.plain_password = Some(password.into());
        self.options.plain_server = false;
        self
    }

    /// Configure as PLAIN server (expects HELLO from clients). Sets mechanism to PLAIN.
    ///
    /// PLAIN server by itself does not validate credentials — pair with a
    /// [`zap_domain`](Self::zap_domain) + [`set_zap_handler`](crate::set_zap_handler)
    /// to check them against a real store.
    ///
    /// ```rust,ignore
    /// use rustzmq2::prelude::*;
    ///
    /// let mut rep = rustzmq2::RepSocket::builder().plain_server().build();
    /// rep.bind("tcp://127.0.0.1:5555").await?;
    /// ```
    pub fn plain_server(mut self) -> Self {
        self.options.mechanism = ZmqMechanism::PLAIN;
        self.options.plain_server = true;
        self
    }

    /// Configure CURVE client. Sets mechanism to CURVE.
    ///
    /// `server_key`: server's long-term public key (32 bytes), known
    /// out-of-band. `public_key` / `secret_key`: this client's long-term
    /// keypair (32 bytes each). See [RFC 25].
    ///
    /// ```rust,ignore
    /// use crypto_box::{SecretKey, aead::OsRng};
    /// use rustzmq2::prelude::*;
    ///
    /// let client_sec = SecretKey::generate(&mut OsRng);
    /// let client_pub = *client_sec.public_key().as_bytes();
    /// let server_pub: [u8; 32] = /* from config */
    /// # [0u8; 32];
    ///
    /// let mut dealer = rustzmq2::DealerSocket::builder()
    ///     .curve_client(server_pub, client_pub, client_sec.to_bytes())
    ///     .build();
    /// dealer.connect("tcp://127.0.0.1:5555").await?;
    /// ```
    ///
    /// [RFC 25]: https://rfc.zeromq.org/spec/25/
    #[cfg(feature = "curve")]
    pub fn curve_client(
        mut self,
        server_key: [u8; 32],
        public_key: [u8; 32],
        secret_key: [u8; 32],
    ) -> Self {
        self.options.mechanism = ZmqMechanism::CURVE;
        self.options.curve_server_key = Some(server_key);
        self.options.curve_public_key = Some(public_key);
        self.options.curve_secret_key = Some(secret_key);
        self.options.curve_server = false;
        self
    }

    /// Configure CURVE server. Sets mechanism to CURVE.
    ///
    /// `public_key` / `secret_key`: this server's long-term keypair
    /// (32 bytes each). Clients must have `public_key` in advance.
    ///
    /// ```rust,ignore
    /// use crypto_box::{SecretKey, aead::OsRng};
    /// use rustzmq2::prelude::*;
    ///
    /// let server_sec = SecretKey::generate(&mut OsRng);
    /// let server_pub = *server_sec.public_key().as_bytes();
    ///
    /// let mut rep = rustzmq2::RepSocket::builder()
    ///     .curve_server(server_pub, server_sec.to_bytes())
    ///     .build();
    /// rep.bind("tcp://127.0.0.1:5555").await?;
    /// ```
    #[cfg(feature = "curve")]
    pub fn curve_server(mut self, public_key: [u8; 32], secret_key: [u8; 32]) -> Self {
        self.options.mechanism = ZmqMechanism::CURVE;
        self.options.curve_public_key = Some(public_key);
        self.options.curve_secret_key = Some(secret_key);
        self.options.curve_server = true;
        self
    }

    /// Set the ZAP domain. When set, every new connection is authenticated
    /// via the registered ZAP handler before the READY exchange.
    ///
    /// Register the handler with [`set_zap_handler`](crate::set_zap_handler)
    /// *before* any socket binds or connects. See [RFC 27].
    ///
    /// [RFC 27]: https://rfc.zeromq.org/spec/27/
    pub fn zap_domain(mut self, domain: impl Into<String>) -> Self {
        self.options.zap_domain = Some(domain.into());
        self
    }

    /// Set the send high-water mark (per-peer outbound queue depth).
    /// Messages are blocked (or dropped for PUB) when the queue is full.
    /// Must be ≥ 1; values of 0 are clamped to 1.
    ///
    /// ```rust,no_run
    /// use rustzmq2::Socket;
    ///
    /// // Bound back-pressure: PUB will drop frames if subscribers can't keep up.
    /// let _pub_ = rustzmq2::PubSocket::builder().send_hwm(64).build();
    ///
    /// // DEALER blocks `send().await` instead of dropping when full.
    /// let _dealer = rustzmq2::DealerSocket::builder().send_hwm(2_000).build();
    /// ```
    pub fn send_hwm(mut self, hwm: usize) -> Self {
        self.options.send_hwm = hwm.max(1);
        self
    }

    /// Set the receive high-water mark (shared inbound channel capacity).
    /// Must be ≥ 1; values of 0 are clamped to 1.
    pub fn receive_hwm(mut self, hwm: usize) -> Self {
        self.options.receive_hwm = hwm.max(1);
        self
    }

    /// Reject inbound messages whose total byte length exceeds `max`.
    /// The peer is disconnected when a too-large message arrives.
    pub fn max_msg_size(mut self, max: usize) -> Self {
        self.options.max_msg_size = Some(max);
        self
    }

    /// Timeout for `recv()` calls. If the socket receives no message within
    /// this duration, `recv()` returns `ZmqError::NoMessage`.
    pub fn receive_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.options.receive_timeout = Some(timeout);
        self
    }

    /// Timeout for `send()` calls. If the send cannot complete within this
    /// duration (e.g. HWM reached), `send()` returns an error.
    pub fn send_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.options.send_timeout = Some(timeout);
        self
    }

    /// Minimum interval before the first reconnect attempt (default 100 ms).
    pub fn reconnect_interval(mut self, ivl: std::time::Duration) -> Self {
        self.options.reconnect_interval = ivl;
        self
    }

    /// Maximum reconnect interval after exponential backoff (default 30 s).
    /// Set equal to `reconnect_interval` to use a fixed interval.
    pub fn reconnect_interval_max(mut self, max: std::time::Duration) -> Self {
        self.options.reconnect_interval_max = max;
        self
    }

    /// Timeout for establishing a TCP connection (default: OS default).
    pub fn connect_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.options.connect_timeout = Some(timeout);
        self
    }

    /// Maximum number of pending TCP connections waiting in the kernel
    /// accept queue (the `backlog` arg to `listen(2)`). Tune up only
    /// for servers that see bursty connection storms — for steady-rate
    /// workloads the default 100 is fine.
    pub fn backlog(mut self, backlog: u32) -> Self {
        self.options.backlog = backlog;
        self
    }

    /// Bind/connect on IPv6, accepting IPv4 too via `IPV6_V6ONLY=0`
    /// dual-stack mode. With this off (default), `tcp://[::]` only
    /// accepts IPv6 and `tcp://0.0.0.0` only accepts IPv4 — set this to
    /// `true` if you want a single bind to serve both families.
    pub fn ipv6(mut self, enabled: bool) -> Self {
        self.options.ipv6 = enabled;
        self
    }

    /// Configure linger behaviour on socket drop.
    ///
    /// `Some(Duration::ZERO)` discards pending messages immediately (current
    /// default). `None` requests blocking until all messages drain (blocking
    /// linger is not yet implemented and behaves the same as `Some(ZERO)` for
    /// now).
    pub fn linger(mut self, linger: Option<std::time::Duration>) -> Self {
        self.options.linger = linger;
        self
    }

    /// Keep only the most recent inbound message per peer (default false).
    ///
    /// When enabled, the per-peer inbound slot is overwritten on each new
    /// message arrival. Older messages are silently discarded. Useful for
    /// state-update streams where only the latest value matters.
    pub fn conflate(mut self, enabled: bool) -> Self {
        self.options.conflate = enabled;
        self
    }

    /// Return `ReturnToSender` immediately when no peers are connected (default false).
    ///
    /// When false (default), `send()` blocks until a peer connects and the
    /// message can be delivered. When true, `send()` returns an error immediately
    /// if there are currently no connected peers.
    pub fn immediate(mut self, enabled: bool) -> Self {
        self.options.immediate = enabled;
        self
    }

    /// Override the kernel's TCP send-buffer size (`SO_SNDBUF`) in bytes.
    /// Larger buffers help on high-bandwidth-delay-product paths
    /// (think transcontinental TCP); on a LAN the OS default is fine.
    /// `None` (the default) leaves it at the OS-tuned value.
    pub fn tcp_send_buffer(mut self, bytes: usize) -> Self {
        self.options.tcp_send_buffer = Some(bytes);
        self
    }

    /// Override the kernel's TCP receive-buffer size (`SO_RCVBUF`) in
    /// bytes. Same tuning rationale as
    /// [`tcp_send_buffer`](Self::tcp_send_buffer) — bump it for fat
    /// long-haul links, otherwise leave OS defaults alone.
    pub fn tcp_receive_buffer(mut self, bytes: usize) -> Self {
        self.options.tcp_receive_buffer = Some(bytes);
        self
    }

    /// Enable kernel TCP keepalive (`SO_KEEPALIVE`). The kernel sends
    /// periodic empty packets and drops the socket if they go
    /// unanswered, catching half-open connections invisibly to the
    /// application.
    ///
    /// Tune the timing with the three `tcp_keepalive_*` knobs below;
    /// without them the kernel uses its (very long) defaults — typically
    /// 2 hours idle on Linux.
    ///
    /// Independent of [`heartbeat_interval`](Self::heartbeat_interval),
    /// which is a higher-level ZMTP-level probe rather than a kernel
    /// one. Either is fine; pick whichever you prefer.
    pub fn tcp_keepalive(mut self, enabled: bool) -> Self {
        self.options.tcp_keepalive = Some(enabled);
        self
    }

    /// Idle time before the kernel sends the first TCP keepalive probe
    /// (`TCP_KEEPIDLE` on Linux). On a healthy connection no probes
    /// fire — it just gates how long a *broken* one sits before the
    /// kernel notices.
    pub fn tcp_keepalive_idle(mut self, idle: std::time::Duration) -> Self {
        self.options.tcp_keepalive_idle = Some(idle);
        self
    }

    /// Spacing between successive TCP keepalive probes once the idle
    /// timer has fired (`TCP_KEEPINTVL`).
    pub fn tcp_keepalive_interval(mut self, interval: std::time::Duration) -> Self {
        self.options.tcp_keepalive_interval = Some(interval);
        self
    }

    /// How many unanswered TCP keepalive probes trigger a drop
    /// (`TCP_KEEPCNT`). Total time-to-detection is roughly
    /// `keepalive_idle + count * keepalive_interval`.
    pub fn tcp_keepalive_count(mut self, count: u32) -> Self {
        self.options.tcp_keepalive_count = Some(count);
        self
    }

    /// `TCP_USER_TIMEOUT` on Linux, `TCP_MAXRT` on Windows: max time a
    /// transmitted packet may remain unacknowledged before the connection is
    /// forcibly closed.
    pub fn tcp_max_retransmit(mut self, duration: std::time::Duration) -> Self {
        self.options.tcp_max_retransmit = Some(duration);
        self
    }

    /// Set the IPv4 `IP_TOS` byte (or IPv6 `IPV6_TCLASS`) on the TCP
    /// socket. The upper 6 bits are the [DSCP][] class — used by
    /// network gear to put traffic into different forwarding queues
    /// (low latency vs. bulk, etc.); the lower 2 bits are ECN.
    ///
    /// Common values (note: shifted into the high 6 bits of the byte):
    ///
    /// | DSCP class | `tos` value |
    /// |---|---|
    /// | Best-effort (default) | `0x00` |
    /// | Expedited Forwarding (low latency) | `0xb8` (DSCP 46) |
    /// | Assured Forwarding AF31 | `0x68` (DSCP 26) |
    /// | Class Selector CS5 (signaling) | `0xa0` (DSCP 40) |
    ///
    /// Has no effect unless your network honors DSCP — most public
    /// internet paths strip it.
    ///
    /// [DSCP]: https://en.wikipedia.org/wiki/Differentiated_services
    pub fn type_of_service(mut self, tos: u32) -> Self {
        self.options.type_of_service = Some(tos);
        self
    }

    /// Linux `SO_BINDTODEVICE`: restrict the socket to a named interface
    /// (e.g. `"eth0"`). No-op on non-Linux targets.
    pub fn bind_to_device(mut self, iface: impl Into<String>) -> Self {
        self.options.bind_to_device = Some(iface.into());
        self
    }

    /// Max time allowed for the ZMTP handshake to complete (default 30 s).
    /// Pass `None` to disable.
    pub fn handshake_interval(mut self, ivl: Option<std::time::Duration>) -> Self {
        self.options.handshake_interval = ivl;
        self
    }

    /// Max payload bytes the writer coalesces into a single `writev(2)`
    /// (default `Some(8192)`, matches libzmq's `ZMQ_OUT_BATCH_SIZE`).
    /// Pass `None` to disable the byte budget entirely — the drain will
    /// then be bounded only by `out_batch_msgs`. Values of `Some(0)` are
    /// clamped to `Some(1)`. At least one message is always drained even
    /// if it alone exceeds the budget.
    pub fn out_batch_size(mut self, n: Option<usize>) -> Self {
        self.options.out_batch_size = n.map(|v| v.max(1));
        self
    }

    /// Hard ceiling on messages drained from the outbound channel into
    /// one `writev(2)`. Caps per-peer pending-queue depth independently
    /// of [`Self::out_batch_size`].
    ///
    /// Pass `None` to disable the message-count ceiling (rely solely
    /// on the byte budget). `Some(0)` is clamped to `Some(1)`.
    ///
    /// **Default is per-socket-type** (only applied when this builder
    /// hasn't been called):
    ///
    /// - `PUB` / `XPUB`: `Some(32)`. Fanout sockets need a low cap so
    ///   one slow peer can't pile up enough backlog to overflow
    ///   [`Self::send_hwm`] and trigger drops (RFC 29 silent-drop).
    /// - Everything else: `Some(256)`. Point-to-point sockets have one
    ///   peer in flight per send and benefit from the larger
    ///   per-syscall amortization.
    ///
    /// Override only when you know your workload's shape — a low-rate
    /// PUB sending into a slow link might want `Some(8)` to bound
    /// per-peer memory; a tight DEALER pipeline might want
    /// `Some(1024)` for maximum amortization.
    pub fn out_batch_msgs(mut self, n: Option<usize>) -> Self {
        self.options.out_batch_msgs = Some(n.map(|v| v.max(1)));
        self
    }

    /// Max messages the reader drains from the framed decoder per
    /// scheduler wake (default `Some(32)`). Pass `None` to disable
    /// batching — one message per wake, matching pre-batching behavior.
    /// Values of `Some(0)` are clamped to `Some(1)`.
    pub fn in_batch_msgs(mut self, n: Option<usize>) -> Self {
        self.options.in_batch_msgs = n.map(|v| v.max(1));
        self
    }

    /// Caller-thread inline-write fast path. When enabled, `send` /
    /// `send_flushed` attempts to encode and write the message
    /// directly from the calling task before falling back to the
    /// outbound channel + peer-loop path. Saves one task wake per
    /// round trip.
    ///
    /// - `None` — disable inline (channel-only).
    /// - `Some(0)` — enable inline with no payload cap.
    /// - `Some(n)` — enable inline for messages whose total payload
    ///   is `< n` bytes; larger messages fall back to the channel.
    ///
    /// **Tradeoff**: inline saves a task wake per round trip (good
    /// for RTT-shaped workloads — REQ/REP/PAIR, heartbeats,
    /// pipelined RPC) but bypasses the writer's `drain_batch`
    /// coalescing (bad for throughput-shaped workloads where many
    /// queued messages would otherwise share one `writev` syscall).
    ///
    /// **Default is per-socket-type** (only applied when this
    /// builder hasn't been called):
    ///
    /// - `REQ` / `REP` / `PAIR`: `Some(0)`. Protocol enforces queue
    ///   depth ≤ 1 (REQ/REP lockstep, PAIR exclusive 1:1) so there
    ///   is nothing to batch — inline is pure win.
    /// - Everything else: `None`. PUB/PUSH/DEALER are routinely
    ///   used in throughput-shaped patterns; opt in only if you know
    ///   your workload is RTT-shaped (e.g. a low-rate DEALER control
    ///   channel).
    pub fn inline_write_max(mut self, n: Option<usize>) -> Self {
        self.options.inline_write_max = Some(n);
        self
    }

    /// Attach a custom metadata property (key, value) to the ZMTP READY
    /// command. Reserved names (`Socket-Type`, `Identity`, `Resource`,
    /// `User-Id`) are silently ignored. Call repeatedly to set multiple.
    ///
    /// ```rust,no_run
    /// use rustzmq2::Socket;
    ///
    /// let _sock = rustzmq2::DealerSocket::builder()
    ///     .metadata("App-Name", env!("CARGO_PKG_NAME"))
    ///     .metadata("App-Version", env!("CARGO_PKG_VERSION"))
    ///     .metadata("X-Tenant", &b"acme"[..])
    ///     .metadata("Identity", &b"ignored"[..]) // reserved → silently dropped
    ///     .build();
    /// ```
    pub fn metadata(mut self, key: impl Into<String>, value: impl Into<bytes::Bytes>) -> Self {
        let key = key.into();
        const RESERVED: &[&str] = &["Socket-Type", "Identity", "Resource", "User-Id"];
        if !RESERVED.iter().any(|r| r.eq_ignore_ascii_case(&key)) {
            self.options.metadata.insert(key, value.into());
        }
        self
    }

    /// Message auto-sent to every new peer once the handshake completes.
    /// Useful for advertising client info or pre-warming a session without
    /// the application having to send a separate "hello" round-trip.
    ///
    /// ```rust,no_run
    /// use rustzmq2::prelude::*;
    ///
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut sub = rustzmq2::SubSocket::builder()
    ///     .hello_msg("CLIENT-READY")
    ///     .build();
    /// sub.connect("tcp://127.0.0.1:5555").await?;
    /// sub.subscribe("").await?;
    /// // Peer's first inbound from us is "CLIENT-READY".
    /// # Ok(()) }
    /// ```
    pub fn hello_msg(mut self, msg: impl Into<crate::message::ZmqMessage>) -> Self {
        self.options.hello_msg = Some(msg.into());
        self
    }

    /// Message auto-sent to a peer just before it is dropped (best effort).
    /// Lets a peer learn that we're going away in normal-shutdown cases
    /// without a separate close-protocol round-trip.
    ///
    /// ```rust,no_run
    /// use rustzmq2::prelude::*;
    ///
    /// let _dealer = rustzmq2::DealerSocket::builder()
    ///     .disconnect_msg("BYE")
    ///     .build();
    /// ```
    pub fn disconnect_msg(mut self, msg: impl Into<crate::message::ZmqMessage>) -> Self {
        self.options.disconnect_msg = Some(msg.into());
        self
    }

    /// Conditions under which the reconnect task gives up rather than
    /// retry. Combine [`ReconnectStop`] flags with `|`.
    pub fn reconnect_stop(mut self, mask: ReconnectStop) -> Self {
        self.options.reconnect_stop = mask;
        self
    }

    /// Connect through a SOCKS5 proxy at `host:port`.
    pub fn socks_proxy(mut self, addr: impl Into<String>) -> Self {
        self.options.socks_proxy = Some(addr.into());
        self
    }

    /// SOCKS5 username/password credentials (optional).
    pub fn socks_credentials(
        mut self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.options.socks_username = Some(username.into());
        self.options.socks_password = Some(password.into());
        self
    }
}

// ── Role-gated options ───────────────────────────────────────────────────────
//
// The methods below are only available on builders for specific socket
// kinds; the `where` bounds carry sealed role markers from
// [`crate::socket::family`]. Calling them on the wrong kind is a
// compile-time error rather than a silent no-op.

impl<S: super::Socket + super::family::Publisher> SocketBuilder<S> {
    /// PUB/XPUB only (`ZMQ_INVERT_MATCHING`): deliver to subscribers whose
    /// filters do NOT match the message topic (default false).
    pub fn invert_matching(mut self, enabled: bool) -> Self {
        self.options.invert_matching = enabled;
        self
    }
}

impl<S: super::Socket + super::family::ExtendedPublisher> SocketBuilder<S> {
    /// XPUB only (`ZMQ_XPUB_VERBOSE`): when `true`, every
    /// SUBSCRIBE/UNSUBSCRIBE frame is surfaced via `recv`, including
    /// duplicates. When `false` (default, matching libzmq) only the first
    /// SUBSCRIBE for a given topic and the UNSUBSCRIBE that drops it are
    /// surfaced; redundant duplicates are absorbed by the backend.
    pub fn xpub_verbose(mut self, enabled: bool) -> Self {
        self.options.xpub_verbose = enabled;
        self
    }
}

impl<S: super::Socket + super::family::Subscriber> SocketBuilder<S> {
    /// SUB only (`ZMQ_HICCUP_MSG`): message the socket delivers to the
    /// application via `recv` when a previously-connected publisher
    /// reconnects. Lets the application detect that it may have missed
    /// messages during the gap and resync (e.g. by re-fetching state).
    ///
    /// ```rust,no_run
    /// use rustzmq2::prelude::*;
    ///
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut sub = rustzmq2::SubSocket::builder()
    ///     .hiccup_msg("__HICCUP__")
    ///     .build();
    /// sub.connect("tcp://127.0.0.1:5555").await?;
    /// sub.subscribe("").await?;
    /// while let Ok(msg) = sub.recv().await {
    ///     if msg.first().map(|f| f.as_ref()) == Some(b"__HICCUP__") {
    ///         // Publisher reconnected — resync state if needed.
    ///         continue;
    ///     }
    ///     // ... handle real message ...
    /// }
    /// # Ok(()) }
    /// ```
    pub fn hiccup_msg(mut self, msg: impl Into<crate::message::ZmqMessage>) -> Self {
        self.options.hiccup_msg = Some(msg.into());
        self
    }
}