rustzmq2 0.1.0

//! Tests for the extended libzmq socket options added for full parity:
//! `HANDSHAKE_IVL`, `INVERT_MATCHING`, TCP knobs, batch sizes,
//! `ONLY_FIRST_SUBSCRIBE`, `METADATA`, `HELLO_MSG`, `DISCONNECT_MSG`, `HICCUP_MSG`,
//! `RECONNECT_STOP`, and `SOCKS_PROXY`.

#[cfg(all(test, feature = "tokio", feature = "tcp"))]
mod test {
    use rustzmq2::__async_rt as async_rt;
    use rustzmq2::prelude::*;
    use rustzmq2::{
        PubSocket, PullSocket, PushSocket, ReconnectStop, RepSocket, ReqSocket, SubSocket,
        XPubSocket, ZmqMessage,
    };

    use std::time::Duration;

    // ── HANDSHAKE_IVL ────────────────────────────────────────────────────────────

    /// Connect to a listener that accepts but sends nothing; short `handshake_interval`
    /// should cause the connect to fail quickly rather than block indefinitely.
    #[async_rt::test]
    async fn handshake_interval_enforced() {
        use tokio::net::TcpListener;
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        // Accept-and-hold task; do not send any bytes.
        let accept_task = async_rt::task::spawn(async move {
            let (_stream, _) = listener.accept().await.unwrap();
            async_rt::task::sleep(Duration::from_secs(5)).await;
        });

        let mut req = ReqSocket::builder()
            .handshake_interval(Some(Duration::from_millis(200)))
            .connect_timeout(Duration::from_millis(200))
            .build();
        let start = std::time::Instant::now();
        let _ = req.connect(&format!("tcp://{}", addr)).await;
        // Even if connect returns before handshake, send should fail quickly.
        let send_result = async_rt::task::timeout(
            Duration::from_secs(2),
            req.send(ZmqMessage::from(b"ping".to_vec())),
        )
        .await;
        let elapsed = start.elapsed();
        assert!(
            elapsed < Duration::from_secs(2),
            "handshake_interval did not bound operation; took {:?}",
            elapsed
        );
        let _ = send_result;
        drop(accept_task);
    }

    // ── INVERT_MATCHING ──────────────────────────────────────────────────────────

    /// PUB with `invert_matching=true` delivers to subscribers whose filter does
    /// NOT match the message topic.
    #[async_rt::test]
    async fn invert_matching_flips_filter() {
        let mut publisher = PubSocket::builder().invert_matching(true).build();
        let ep = publisher.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut sub = SubSocket::new();
        sub.connect(&ep.to_string()).await.unwrap();
        sub.subscribe("A").await.unwrap();

        async_rt::task::sleep(Duration::from_millis(100)).await;

        // Topic "B" does NOT match filter "A" — with invert=true, subscriber
        // SHOULD receive it.
        publisher
            .send(ZmqMessage::from(b"B-hello".to_vec()))
            .await
            .unwrap();

        let msg = async_rt::task::timeout(Duration::from_millis(500), sub.recv())
            .await
            .expect("recv timed out");
        let frame = msg.unwrap().get(0).unwrap().clone();
        assert!(frame.starts_with(b"B"));

        // Topic "A" DOES match filter "A" — with invert=true, should NOT arrive.
        publisher
            .send(ZmqMessage::from(b"A-skip".to_vec()))
            .await
            .unwrap();
        let missed = async_rt::task::timeout(Duration::from_millis(200), sub.recv()).await;
        assert!(
            missed.is_err(),
            "expected A-topic to be suppressed by invert"
        );
    }

    // ── TCP KNOBS ────────────────────────────────────────────────────────────────

    /// Setting SNDBUF/RCVBUF is accepted and observable via socket2 after connect.
    #[async_rt::test]
    async fn tcp_send_receive_buffer_applied() {
        let mut pull = PullSocket::new();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut push = PushSocket::builder()
            .tcp_send_buffer(65536)
            .tcp_receive_buffer(65536)
            .tcp_keepalive(true)
            .build();
        push.connect(&ep.to_string()).await.unwrap();

        async_rt::task::sleep(Duration::from_millis(100)).await;
        // Sending should work with the configured knobs.
        push.send(ZmqMessage::from(b"hi".to_vec())).await.unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"hi");
    }

    // ── BATCH SIZES ──────────────────────────────────────────────────────────────

    /// Setting `out_batch_size=1` still delivers messages reliably.
    #[async_rt::test]
    async fn out_batch_size_one_is_functional() {
        let mut pull = PullSocket::new();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut push = PushSocket::builder().out_batch_size(Some(1)).build();
        push.connect(&ep.to_string()).await.unwrap();

        async_rt::task::sleep(Duration::from_millis(50)).await;
        for i in 0u32..5 {
            push.send(ZmqMessage::from(i.to_be_bytes().to_vec()))
                .await
                .unwrap();
        }
        for i in 0u32..5 {
            let got = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
                .await
                .unwrap()
                .unwrap();
            let frame = got.get(0).unwrap().clone();
            assert_eq!(u32::from_be_bytes(frame[..4].try_into().unwrap()), i);
        }
    }

    // ── XPUB_VERBOSE ─────────────────────────────────────────────────────────────

    /// With `xpub_verbose = false` (the default, matching libzmq), a second
    /// identical SUBSCRIBE from the same peer is absorbed and not surfaced
    /// via `recv`. Per-peer dedup, not global.
    #[async_rt::test]
    async fn xpub_verbose_false_dedups_events() {
        let mut xpub = XPubSocket::builder().xpub_verbose(false).build();
        let ep = xpub.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut sub = SubSocket::new();
        sub.connect(&ep.to_string()).await.unwrap();
        sub.subscribe("topic").await.unwrap();
        // Duplicate subscribe from the same SUB socket.
        sub.subscribe("topic").await.unwrap();

        // First SUBSCRIBE should surface.
        let ev1 = async_rt::task::timeout(Duration::from_millis(500), xpub.recv())
            .await
            .expect("first subscribe event")
            .unwrap();
        let f = ev1.get(0).unwrap().clone();
        assert_eq!(f[0], 1, "expected SUBSCRIBE opcode");

        // Second identical SUBSCRIBE from the same peer should be dedup'd.
        let ev2 = async_rt::task::timeout(Duration::from_millis(300), xpub.recv()).await;
        assert!(
            ev2.is_err(),
            "expected duplicate SUBSCRIBE from same peer to be suppressed"
        );
    }

    // ── METADATA ─────────────────────────────────────────────────────────────────

    /// Setting metadata on a builder is accepted without error and the socket
    /// still connects and sends messages.
    #[async_rt::test]
    async fn metadata_accepted_and_connection_succeeds() {
        let mut rep = RepSocket::new();
        let ep = rep.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut req = ReqSocket::builder()
            .metadata("App-Name", "zmq-rs-test")
            .metadata("App-Version", "1.0")
            .build();
        req.connect(&ep.to_string()).await.unwrap();

        req.send(ZmqMessage::from(b"ping".to_vec())).await.unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(500), rep.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"ping");
    }

    // ── HELLO_MSG ────────────────────────────────────────────────────────────────

    /// Setting `HELLO_MSG` on a PUSH causes a PULL to receive the hello message
    /// without the application sending anything.
    #[async_rt::test]
    async fn hello_msg_auto_sent_on_connect() {
        let mut push = PushSocket::builder()
            .hello_msg(ZmqMessage::from(b"welcome".to_vec()))
            .build();
        let ep = push.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut pull = PullSocket::new();
        pull.connect(&ep.to_string()).await.unwrap();

        let msg = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .expect("hello_msg should arrive")
            .unwrap();
        assert_eq!(msg.get(0).unwrap().as_ref(), b"welcome");
    }

    // ── RECONNECT_STOP ───────────────────────────────────────────────────────────

    /// After a successful connection is torn down, a socket with
    /// `RECONNECT_STOP_AFTER_DISCONNECT` should NOT emit a second `Connected`
    /// monitor event when the server comes back online — the reconnect task
    /// has exited.
    #[async_rt::test]
    async fn reconnect_stop_after_disconnect_halts_reconnect_task() {
        use futures::StreamExt;

        // Pre-allocate the port so rebinding succeeds.
        let tmp = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = tmp.local_addr().unwrap();
        drop(tmp);
        let ep_str = format!("tcp://{}", addr);

        let mut pull_a = PullSocket::new();
        pull_a.bind(&ep_str).await.unwrap();

        let mut push = PushSocket::builder()
            .reconnect_stop(ReconnectStop::AFTER_DISCONNECT)
            .reconnect_interval(Duration::from_millis(50))
            .build();
        let mut monitor = push.monitor();
        push.connect(&ep_str).await.unwrap();

        // Prove the session is up.
        async_rt::task::sleep(Duration::from_millis(100)).await;
        push.send(ZmqMessage::from(b"ping".to_vec())).await.unwrap();
        pull_a.recv().await.unwrap();

        // Drain monitor events up to and including the initial Connected.
        let mut saw_initial = false;
        while let Ok(Some(ev)) =
            async_rt::task::timeout(Duration::from_millis(100), monitor.next()).await
        {
            if matches!(ev, rustzmq2::SocketEvent::Connected(_, _)) {
                saw_initial = true;
            }
        }
        assert!(saw_initial, "expected initial Connected event");

        // Tear down the server; the reconnect task receives the disconnect
        // notification and (with AFTER_DISCONNECT) should exit.
        drop(pull_a);
        async_rt::task::sleep(Duration::from_millis(300)).await;

        // Bring the server back on the same port.
        let mut pull_b = PullSocket::new();
        pull_b.bind(&ep_str).await.unwrap();

        // No further Connected event should arrive because the reconnect
        // task has exited.
        let saw_reconnect = async_rt::task::timeout(Duration::from_millis(800), async {
            while let Some(ev) = monitor.next().await {
                if matches!(ev, rustzmq2::SocketEvent::Connected(_, _)) {
                    return true;
                }
            }
            false
        })
        .await;
        // Both outcomes — timeout and channel-drained-to-false — mean no
        // reconnect happened; only `Ok(true)` would indicate the bug.
        if matches!(saw_reconnect, Ok(true)) {
            panic!("reconnect_stop=AFTER_DISCONNECT should prevent reconnect");
        }
        drop(push);
        drop(pull_b);
    }

    /// Connecting to a closed port with `RECONNECT_STOP_CONN_REFUSED` should
    /// not retry forever — the socket gives up quickly.
    #[async_rt::test]
    async fn reconnect_stop_conn_refused_bails_out() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        drop(listener);

        let mut pull = PullSocket::builder()
            .reconnect_stop(ReconnectStop::CONN_REFUSED)
            .reconnect_interval(Duration::from_millis(50))
            .connect_timeout(Duration::from_millis(100))
            .build();
        let _ = async_rt::task::timeout(
            Duration::from_millis(500),
            pull.connect(&format!("tcp://{}", addr)),
        )
        .await;
    }

    // ── DISCONNECT_MSG ───────────────────────────────────────────────────────────

    /// `disconnect_msg` set on a PUSH is delivered to the peer when the PUSH's
    /// `peer_disconnected` path fires (i.e. the connection is torn down).
    #[async_rt::test]
    async fn disconnect_msg_delivered_on_peer_teardown() {
        let mut pull = PullSocket::new();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut push = PushSocket::builder()
            .disconnect_msg(ZmqMessage::from(b"goodbye".to_vec()))
            .linger(Some(Duration::from_millis(500)))
            .build();
        push.connect(&ep.to_string()).await.unwrap();

        // Let the handshake settle and send a normal message first.
        async_rt::task::sleep(Duration::from_millis(100)).await;
        push.send(ZmqMessage::from(b"hello".to_vec()))
            .await
            .unwrap();
        let first = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(first.get(0).unwrap().as_ref(), b"hello");

        // `close().await` runs unbind_all + linger_drain; the latter enqueues
        // `disconnect_msg` to every peer before waiting for the outbound
        // channel to drain.
        let _ = push.close().await;

        // The PULL should receive the goodbye frame before EOF.
        let second = async_rt::task::timeout(Duration::from_millis(1000), pull.recv()).await;
        match second {
            Ok(Ok(msg)) => assert_eq!(msg.get(0).unwrap().as_ref(), b"goodbye"),
            other => panic!("expected disconnect_msg, got {:?}", other),
        }
    }

    // ── HICCUP_MSG ───────────────────────────────────────────────────────────────

    /// A SUB with `hiccup_msg` set receives that message via `recv()` after the
    /// PUB reconnects. We drive this by binding the PUB, letting the SUB
    /// connect, then rebinding the PUB on the same port.
    #[async_rt::test]
    async fn hiccup_msg_delivered_after_publisher_reconnect() {
        // Allocate a port up front so the SUB's reconnect hits a deterministic
        // target.
        let tmp = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = tmp.local_addr().unwrap();
        drop(tmp);
        let ep_str = format!("tcp://{}", addr);

        let mut pub_a = PubSocket::new();
        pub_a.bind(&ep_str).await.unwrap();

        let mut sub = SubSocket::builder()
            .hiccup_msg(ZmqMessage::from(b"--hiccup--".to_vec()))
            .reconnect_interval(Duration::from_millis(50))
            .reconnect_interval_max(Duration::from_millis(200))
            .receive_timeout(Duration::from_secs(3))
            .build();
        sub.connect(&ep_str).await.unwrap();
        sub.subscribe("").await.unwrap();

        // Prove the session is up: send, receive.
        async_rt::task::sleep(Duration::from_millis(100)).await;
        pub_a
            .send(ZmqMessage::from(b"first".to_vec()))
            .await
            .unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(1000), sub.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"first");

        // Drop the PUB to trigger the SUB's reconnect loop.
        drop(pub_a);
        async_rt::task::sleep(Duration::from_millis(150)).await;

        // Bring the PUB back on the same endpoint.
        let mut pub_b = PubSocket::new();
        pub_b.bind(&ep_str).await.unwrap();

        // The SUB should receive the hiccup frame (injected by on_reconnect)
        // at some point after the reconnect completes.
        let mut saw_hiccup = false;
        let deadline = std::time::Instant::now() + Duration::from_secs(3);
        while std::time::Instant::now() < deadline {
            match async_rt::task::timeout(Duration::from_millis(500), sub.recv()).await {
                Ok(Ok(msg)) => {
                    if msg.get(0).unwrap().as_ref() == b"--hiccup--" {
                        saw_hiccup = true;
                        break;
                    }
                }
                _ => {
                    // Nudge the reconnect along in case the PUB just came up.
                    let _ = pub_b.send(ZmqMessage::from(b"nudge".to_vec())).await;
                }
            }
        }
        assert!(saw_hiccup, "expected hiccup_msg after publisher reconnect");
    }

    // ── METADATA (round-trip via a manual ZMTP peer) ─────────────────────────────

    /// Custom metadata set via `.metadata(...)` lands on the peer side of the
    /// ZMTP READY command. We spin up a hand-rolled ZMTP server that reads the
    /// raw READY frame and parses its property list.
    #[async_rt::test]
    async fn metadata_reaches_peer_in_ready_command() {
        use std::collections::HashMap;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::TcpListener;

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        // Server task: accept one connection, do the minimum ZMTP greeting
        // dance to get to READY, then extract the READY properties.
        let server = async_rt::task::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();

            // Read client's greeting (64 bytes total, sent in one go).
            let mut buf = [0u8; 64];
            stream.read_exact(&mut buf).await.unwrap();
            // Reply with a matching NULL-mechanism greeting.
            let mut greeting = [0u8; 64];
            greeting[0] = 0xFF;
            greeting[9] = 0x7F;
            greeting[10] = 0x03; // ZMTP 3.0
            greeting[11] = 0x01; // minor = 1
                                 // mechanism[20..40] = "NULL" padded with NUL
            greeting[12..16].copy_from_slice(b"NULL");
            stream.write_all(&greeting).await.unwrap();

            // Read client's READY command (short-command framing).
            // Frame: [flags=0x04] [len] [body...] where body =
            //   0x05 "READY" <property-list>
            let mut hdr = [0u8; 2];
            stream.read_exact(&mut hdr).await.unwrap();
            assert!(hdr[0] & 0x04 != 0, "expected command frame");
            let len = if hdr[0] & 0x02 != 0 {
                let mut lb = [0u8; 8];
                lb[0] = hdr[1];
                stream.read_exact(&mut lb[1..]).await.unwrap();
                u64::from_be_bytes(lb) as usize
            } else {
                hdr[1] as usize
            };
            let mut body = vec![0u8; len];
            stream.read_exact(&mut body).await.unwrap();

            // body[0] = 5 (name length), body[1..6] = "READY", then properties.
            assert_eq!(body[0], 5);
            assert_eq!(&body[1..6], b"READY");
            let mut i = 6;
            let mut props: HashMap<String, Vec<u8>> = HashMap::new();
            while i < body.len() {
                let nlen = body[i] as usize;
                i += 1;
                let name = std::str::from_utf8(&body[i..i + nlen]).unwrap().to_string();
                i += nlen;
                let vlen = u32::from_be_bytes(body[i..i + 4].try_into().unwrap()) as usize;
                i += 4;
                let value = body[i..i + vlen].to_vec();
                i += vlen;
                props.insert(name, value);
            }
            props
        });

        // Client side: build a ReqSocket with two custom metadata entries.
        let mut req = ReqSocket::builder()
            .metadata("X-App-Name", "zmq-rs-test")
            .metadata("X-App-Version", "42")
            .handshake_interval(Some(Duration::from_secs(2)))
            .build();
        let _ = async_rt::task::timeout(
            Duration::from_secs(2),
            req.connect(&format!("tcp://{}", addr)),
        )
        .await;

        let props = async_rt::task::timeout(Duration::from_secs(3), server)
            .await
            .expect("server task timed out")
            .expect("server join");

        assert_eq!(
            props.get("X-App-Name").map(|v| v.as_slice()),
            Some(b"zmq-rs-test".as_slice()),
            "expected X-App-Name in READY properties, got keys {:?}",
            props.keys().collect::<Vec<_>>()
        );
        assert_eq!(
            props.get("X-App-Version").map(|v| v.as_slice()),
            Some(b"42".as_slice()),
        );
    }

    // ── TCP SNDBUF/RCVBUF (observable) ───────────────────────────────────────────

    /// Set SNDBUF/RCVBUF on the PUSH, accept the connection with a raw TCP
    /// listener, and read back the buffer size via socket2 to prove the knob
    /// reached the kernel.
    #[async_rt::test]
    async fn tcp_send_receive_buffer_observable_via_socket2() {
        use tokio::net::TcpListener;
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        // On Linux the kernel doubles what you set; macOS/BSD accept nearly
        // verbatim. We pick a size that's clearly above the default on every
        // platform and only assert the kernel chose something >= half of it.
        const REQ: usize = 256 * 1024;
        let accept_task = async_rt::task::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            let sock = socket2::SockRef::from(&stream);
            let snd = sock.send_buffer_size().unwrap();
            let rcv = sock.recv_buffer_size().unwrap();
            (snd, rcv)
        });

        // Pull-like socket driven solely by our raw listener; we don't care
        // about ZMTP completing, only that the connect attempt lands.
        let mut push = PushSocket::builder()
            .tcp_send_buffer(REQ)
            .tcp_receive_buffer(REQ)
            .handshake_interval(Some(Duration::from_millis(300)))
            .build();
        let _ = async_rt::task::timeout(
            Duration::from_millis(800),
            push.connect(&format!("tcp://{}", addr)),
        )
        .await;

        let (snd, rcv) = async_rt::task::timeout(Duration::from_secs(2), accept_task)
            .await
            .unwrap()
            .unwrap();
        assert!(
            snd >= REQ / 2,
            "send_buffer_size={} too small (requested {})",
            snd,
            REQ
        );
        assert!(
            rcv >= REQ / 2,
            "recv_buffer_size={} too small (requested {})",
            rcv,
            REQ
        );
    }

    // ── RCVTIMEO ─────────────────────────────────────────────────────────────────

    /// `receive_timeout` causes `recv()` to return `ZmqError::NoMessage` after the
    /// timeout elapses.
    #[async_rt::test]
    async fn receive_timeout_returns_no_message() {
        let mut pull = PullSocket::builder()
            .receive_timeout(Duration::from_millis(200))
            .build();
        pull.bind("tcp://127.0.0.1:0").await.unwrap();
        let start = std::time::Instant::now();
        let res = pull.recv().await;
        let elapsed = start.elapsed();
        assert!(matches!(res, Err(rustzmq2::ZmqError::NoMessage)));
        assert!(
            elapsed < Duration::from_millis(800),
            "receive_timeout did not bound recv; took {:?}",
            elapsed
        );
    }

    // ── SNDTIMEO ─────────────────────────────────────────────────────────────────

    /// `send_timeout` bounds `send()` when the outbound HWM is full and the peer
    /// is not draining. Fill the channel until the next send hits the cap,
    /// then assert the timed `send` returns `NoMessage` within the deadline.
    #[allow(clippy::match_wild_err_arm)]
    #[async_rt::test]
    async fn send_timeout_returns_no_message_on_hwm_block() {
        use tokio::net::TcpListener;

        // Raw listener that accepts the PUSH but never reads beyond the ZMTP
        // greeting. The PUSH completes the handshake against us, but since we
        // don't drain application frames, the kernel TCP recv buffer + PUSH
        // outbound channel will fill.
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let _accept = async_rt::task::spawn(async move {
            use tokio::io::{AsyncReadExt, AsyncWriteExt};
            let (mut stream, _) = listener.accept().await.unwrap();
            // Handshake: read client's 64-byte greeting, reply with our own.
            let mut buf = [0u8; 64];
            let _ = stream.read_exact(&mut buf).await;
            let mut greeting = [0u8; 64];
            greeting[0] = 0xFF;
            greeting[9] = 0x7F;
            greeting[10] = 0x03;
            greeting[11] = 0x01;
            greeting[12..16].copy_from_slice(b"NULL");
            let _ = stream.write_all(&greeting).await;
            // Read READY command (we don't care about contents).
            let _ = stream.read(&mut [0u8; 1024]).await;
            // Send our READY so the PUSH finishes the handshake.
            let mut ready = vec![
                0x04, // command frame
                0x0B, // body len: 1 (name len) + 5 ("READY") + property list (5)
                0x05, b'R', b'E', b'A', b'D',
                b'Y', //
                      // Property: Socket-Type = PULL
            ];
            // Socket-Type property: len=11 "Socket-Type" + 4-byte value-len +
            // "PULL" (4 bytes) = 11+1+4+4 = 20 bytes. Reframe the short header
            // length. Easier: just craft with longer len.
            ready.clear();
            let mut body = Vec::new();
            body.push(5);
            body.extend_from_slice(b"READY");
            body.push(11);
            body.extend_from_slice(b"Socket-Type");
            body.extend_from_slice(&(4u32).to_be_bytes());
            body.extend_from_slice(b"PULL");
            ready.push(0x04);
            ready.push(body.len() as u8);
            ready.extend_from_slice(&body);
            let _ = stream.write_all(&ready).await;
            // Now stall — don't read anything else. The stream stays open so
            // the peer thinks we're connected.
            async_rt::task::sleep(Duration::from_secs(10)).await;
        });

        let mut push = PushSocket::builder()
            .send_hwm(2)
            .tcp_send_buffer(8192)
            .send_timeout(Duration::from_millis(300))
            .build();
        push.connect(&format!("tcp://{}", addr)).await.unwrap();
        async_rt::task::sleep(Duration::from_millis(200)).await;

        // Spam large messages until we either hit the timeout or run out of
        // tries. A correctly implemented send_timeout will eventually fire.
        let mut hit = false;
        for _ in 0..500 {
            match async_rt::task::timeout(
                Duration::from_millis(800),
                push.send(ZmqMessage::from(vec![0u8; 16 * 1024])),
            )
            .await
            {
                Ok(Err(rustzmq2::ZmqError::NoMessage)) => {
                    hit = true;
                    break;
                }
                Ok(Ok(())) => {}
                Ok(Err(e)) => panic!("unexpected send error: {:?}", e),
                Err(_) => panic!("outer timeout — send_timeout not enforced"),
            }
        }
        assert!(hit, "expected send_timeout to fire after HWM backpressure");
    }

    // ── CONNECT_TIMEOUT ──────────────────────────────────────────────────────────

    /// `connect_timeout` bounds TCP connect attempts to an unreachable host.
    /// We use a TEST-NET-1 address (RFC 5737) which is guaranteed not to route.
    #[async_rt::test]
    async fn connect_timeout_bounds_tcp_dial() {
        let mut push = PushSocket::builder()
            .connect_timeout(Duration::from_millis(250))
            .reconnect_interval(Duration::from_millis(50))
            .build();
        let start = std::time::Instant::now();
        // 192.0.2.1 is TEST-NET-1 — deliberately unroutable.
        let res = async_rt::task::timeout(
            Duration::from_secs(3),
            push.connect("tcp://192.0.2.1:12345"),
        )
        .await;
        let elapsed = start.elapsed();
        assert!(
            elapsed < Duration::from_secs(3),
            "connect_timeout did not bound dial; took {:?}",
            elapsed
        );
        let _ = res;
    }

    // ── MAX_MSG_SIZE ─────────────────────────────────────────────────────────────

    /// `max_msg_size` causes the receiver to reject oversized frames and
    /// tear down the peer.
    #[async_rt::test]
    async fn max_msg_size_rejects_oversized() {
        let mut pull = PullSocket::builder().max_msg_size(256).build();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut push = PushSocket::new();
        push.connect(&ep.to_string()).await.unwrap();
        async_rt::task::sleep(Duration::from_millis(100)).await;

        // Small frame: should arrive.
        push.send(ZmqMessage::from(vec![0u8; 10])).await.unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().len(), 10);

        // Oversized frame: should be rejected, peer torn down.
        // We don't care whether send errors; just that recv() doesn't
        // deliver the oversized frame.
        let _ = push.send(ZmqMessage::from(vec![0u8; 4096])).await;
        let rejected = async_rt::task::timeout(Duration::from_millis(500), pull.recv()).await;
        if let Ok(Ok(m)) = rejected {
            assert_ne!(
                m.get(0).unwrap().len(),
                4096,
                "oversized frame should not have arrived"
            );
        }
        // Timeout or error are both acceptable — peer torn down means recv()
        // will never yield the oversized frame.
    }

    // ── HEARTBEAT ────────────────────────────────────────────────────────────────

    /// Heartbeats keep idle TCP connections alive. We can't easily assert PINGs
    /// on the wire without a raw peer, so we verify the session survives longer
    /// than the heartbeat interval without either side sending data.
    #[async_rt::test]
    async fn heartbeat_keeps_idle_connection_alive() {
        let mut pull = PullSocket::builder()
            .heartbeat_interval(Duration::from_millis(100))
            .heartbeat_timeout(Duration::from_millis(300))
            .heartbeat_ttl(Duration::from_millis(500))
            .build();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut push = PushSocket::builder()
            .heartbeat_interval(Duration::from_millis(100))
            .heartbeat_timeout(Duration::from_millis(300))
            .heartbeat_ttl(Duration::from_millis(500))
            .build();
        push.connect(&ep.to_string()).await.unwrap();

        async_rt::task::sleep(Duration::from_millis(100)).await;
        push.send(ZmqMessage::from(b"hello".to_vec()))
            .await
            .unwrap();
        pull.recv().await.unwrap();

        // Remain idle for well over the heartbeat interval; heartbeats should
        // keep both sides alive.
        async_rt::task::sleep(Duration::from_millis(600)).await;

        // A subsequent send should still succeed.
        push.send(ZmqMessage::from(b"still-here".to_vec()))
            .await
            .unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .expect("connection died during idle")
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"still-here");
    }

    // ── TCP_KEEPALIVE_IDLE / INTERVAL / COUNT ────────────────────────────────────

    /// Setting keepalive idle/interval/count on the builder is accepted and
    /// the resulting socket still operates. `SO_KEEPALIVE` is a per-end flag
    /// applied to the fd owned inside zmq.rs, so we can't observe it from
    /// outside without an accessor; this is a smoke test for the glue.
    #[async_rt::test]
    async fn tcp_keepalive_tuning_applied() {
        let mut pull = PullSocket::new();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();

        let mut push = PushSocket::builder()
            .tcp_keepalive(true)
            .tcp_keepalive_idle(Duration::from_secs(30))
            .tcp_keepalive_interval(Duration::from_secs(5))
            .tcp_keepalive_count(3)
            .build();
        push.connect(&ep.to_string()).await.unwrap();
        async_rt::task::sleep(Duration::from_millis(100)).await;

        push.send(ZmqMessage::from(b"keepalive".to_vec()))
            .await
            .unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"keepalive");
    }

    // ── TOS ──────────────────────────────────────────────────────────────────────

    /// `tos` (`IP_TOS` / DSCP) is set on the connecting socket. Read it back
    /// via socket2 on the accepted side — TOS is per-packet, not observable
    /// on the peer. Instead verify the knob applied locally.
    #[async_rt::test]
    async fn tos_applied_to_local_socket() {
        // We can't observe TOS on the peer side; instead, configure a PUSH
        // with a specific TOS and just confirm the connect path doesn't fail.
        // This is a smoke test for the build/apply glue — a stronger assertion
        // would require a raw socket inspecting IP headers.
        let mut pull = PullSocket::new();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();
        let mut push = PushSocket::builder().type_of_service(0x10).build();
        push.connect(&ep.to_string()).await.unwrap();
        async_rt::task::sleep(Duration::from_millis(100)).await;
        push.send(ZmqMessage::from(b"tos".to_vec())).await.unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"tos");
    }

    // ── BACKLOG ──────────────────────────────────────────────────────────────────

    /// `backlog` is plumbed to listen(2). The kernel may clamp it; we only
    /// verify that setting an unusually large value still lets connections succeed.
    #[async_rt::test]
    async fn backlog_accepts_connections() {
        let mut pull = PullSocket::builder().backlog(512).build();
        let ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();
        let mut push = PushSocket::new();
        push.connect(&ep.to_string()).await.unwrap();
        async_rt::task::sleep(Duration::from_millis(100)).await;
        push.send(ZmqMessage::from(b"queued".to_vec()))
            .await
            .unwrap();
        let got = async_rt::task::timeout(Duration::from_millis(500), pull.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"queued");
    }

    // ── SOCKS5 PROXY (in-process test proxy) ─────────────────────────────────────

    /// Stand up a tiny inline SOCKS5 server, connect a PUSH through it, and
    /// confirm the end-to-end message arrives at a PULL on the "real" target.
    #[async_rt::test]
    async fn socks_proxy_end_to_end() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::TcpListener;

        // The real ZMTP server.
        let mut pull = PullSocket::new();
        let pull_ep = pull.bind("tcp://127.0.0.1:0").await.unwrap();
        let target_port = match &pull_ep {
            rustzmq2::Endpoint::Tcp(_, port) => *port,
            _ => unreachable!(),
        };

        // The SOCKS5 proxy.
        let proxy_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let proxy_addr = proxy_listener.local_addr().unwrap();
        let proxy_task = async_rt::task::spawn(async move {
            let (mut client, _) = proxy_listener.accept().await.unwrap();

            // Method negotiation: ver=5, nmethods, methods...
            let mut g = [0u8; 2];
            client.read_exact(&mut g).await.unwrap();
            let n = g[1] as usize;
            let mut methods = vec![0u8; n];
            client.read_exact(&mut methods).await.unwrap();
            client.write_all(&[0x05, 0x00]).await.unwrap(); // NO-AUTH

            // CONNECT request.
            let mut hdr = [0u8; 4];
            client.read_exact(&mut hdr).await.unwrap();
            assert_eq!(hdr[0], 0x05);
            assert_eq!(hdr[1], 0x01); // CMD=CONNECT
            let target_host = match hdr[3] {
                0x01 => {
                    let mut ip = [0u8; 4];
                    client.read_exact(&mut ip).await.unwrap();
                    std::net::IpAddr::V4(ip.into()).to_string()
                }
                0x03 => {
                    let mut len = [0u8; 1];
                    client.read_exact(&mut len).await.unwrap();
                    let mut host = vec![0u8; len[0] as usize];
                    client.read_exact(&mut host).await.unwrap();
                    String::from_utf8(host).unwrap()
                }
                0x04 => {
                    let mut ip = [0u8; 16];
                    client.read_exact(&mut ip).await.unwrap();
                    std::net::IpAddr::V6(ip.into()).to_string()
                }
                other => panic!("unexpected ATYP {:#x}", other),
            };
            let mut port = [0u8; 2];
            client.read_exact(&mut port).await.unwrap();
            let port = u16::from_be_bytes(port);
            assert_eq!(port, target_port);

            // Open the real upstream and splice.
            let upstream = tokio::net::TcpStream::connect((target_host.as_str(), port))
                .await
                .unwrap();
            // Reply: success, BND.ADDR = 0.0.0.0:0
            client
                .write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
                .await
                .unwrap();
            let (mut ci, mut co) = client.into_split();
            let (mut ui, mut uo) = upstream.into_split();
            let a = tokio::io::copy(&mut ci, &mut uo);
            let b = tokio::io::copy(&mut ui, &mut co);
            let _ = tokio::join!(a, b);
        });

        // PUSH dials through the proxy.
        let mut push = PushSocket::builder()
            .socks_proxy(proxy_addr.to_string())
            .build();
        push.connect(&pull_ep.to_string()).await.unwrap();

        async_rt::task::sleep(Duration::from_millis(150)).await;
        push.send(ZmqMessage::from(b"via-socks".to_vec()))
            .await
            .unwrap();
        let got = async_rt::task::timeout(Duration::from_secs(2), pull.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(got.get(0).unwrap().as_ref(), b"via-socks");
        drop(push);
        drop(pull);
        let _ = async_rt::task::timeout(Duration::from_millis(200), proxy_task).await;
    }
}