//! Port of `libzmq/tests/test_push_pull.cpp` (subset)
//! PUSH/PULL: fan-out work distribution, multiple pushers/pullers.
#![allow(clippy::borrow_as_ptr, clippy::ref_as_ptr)]

mod helpers;

use std::ffi::CString;
use std::mem::size_of;
use std::time::Duration;

use omq_zmq::{
    zmq_bind, zmq_close, zmq_connect, zmq_ctx_new, zmq_ctx_term, zmq_recv, zmq_send,
    zmq_setsockopt, zmq_socket,
};

const ZMQ_PUSH: i32 = 8;
const ZMQ_PULL: i32 = 7;
const ZMQ_RCVTIMEO: i32 = 27;
const ZMQ_SNDTIMEO: i32 = 28;
const ZMQ_DONTWAIT: i32 = 1;
const TIMEOUT_MS: i32 = 2000;

fn set_timeo(sock: *mut std::ffi::c_void, opt: i32, ms: i32) {
    zmq_setsockopt(sock, opt, (&ms as *const i32).cast(), size_of::<i32>());
}

/// From `libzmq/tests/test_push_pull.cpp`: basic push/pull
#[test]
fn push_pull_basic_tcp() {
    const N: usize = 10;

    let port = helpers::free_port();
    let addr = CString::new(format!("tcp://127.0.0.1:{port}")).unwrap();

    let ctx = zmq_ctx_new();
    let pull = zmq_socket(ctx, ZMQ_PULL);
    let push = zmq_socket(ctx, ZMQ_PUSH);

    assert_eq!(zmq_bind(pull, addr.as_ptr()), 0);
    assert_eq!(zmq_connect(push, addr.as_ptr()), 0);
    std::thread::sleep(Duration::from_millis(100));

    set_timeo(pull, ZMQ_RCVTIMEO, TIMEOUT_MS);
    set_timeo(push, ZMQ_SNDTIMEO, TIMEOUT_MS);

    let payload = b"message";

    for i in 0..N {
        let rc = zmq_send(push, payload.as_ptr().cast(), payload.len(), 0);
        assert_eq!(rc as usize, payload.len(), "send {i} failed");
    }

    let mut buf = [0u8; 32];
    for i in 0..N {
        let rc = zmq_recv(pull, buf.as_mut_ptr().cast(), buf.len(), 0);
        assert_eq!(rc as usize, payload.len(), "recv {i} failed");
        assert_eq!(&buf[..payload.len()], payload);
    }

    zmq_close(push);
    zmq_close(pull);
    zmq_ctx_term(ctx);
}

/// Multiple pushers -> one puller (fan-in).
#[test]
fn push_pull_multiple_pushers() {
    const N_PUSHERS: usize = 3;
    const MSG_PER_PUSHER: usize = 5;

    let port = helpers::free_port();
    let addr_str = format!("tcp://127.0.0.1:{port}");
    let addr = CString::new(addr_str.clone()).unwrap();

    let ctx = zmq_ctx_new();
    let pull = zmq_socket(ctx, ZMQ_PULL);
    zmq_bind(pull, addr.as_ptr());

    let mut pushers = Vec::new();
    for _ in 0..N_PUSHERS {
        let p = zmq_socket(ctx, ZMQ_PUSH);
        let a = CString::new(addr_str.clone()).unwrap();
        zmq_connect(p, a.as_ptr());
        pushers.push(p);
    }

    std::thread::sleep(Duration::from_millis(100));
    set_timeo(pull, ZMQ_RCVTIMEO, TIMEOUT_MS);

    for push in &pushers {
        for _ in 0..MSG_PER_PUSHER {
            zmq_send(*push, b"x".as_ptr().cast(), 1, 0);
        }
    }

    let mut buf = [0u8; 8];
    let total = N_PUSHERS * MSG_PER_PUSHER;
    for _ in 0..total {
        let rc = zmq_recv(pull, buf.as_mut_ptr().cast(), buf.len(), 0);
        assert_eq!(rc, 1);
    }

    for p in pushers {
        zmq_close(p);
    }
    zmq_close(pull);
    zmq_ctx_term(ctx);
}

/// One pusher -> multiple pullers (work stealing / round-robin).
#[test]
fn push_pull_multiple_pullers() {
    const N_PULLERS: usize = 3;
    const TOTAL: usize = 30;

    let port = helpers::free_port();
    let addr = CString::new(format!("tcp://127.0.0.1:{port}")).unwrap();

    let ctx = zmq_ctx_new();
    let push = zmq_socket(ctx, ZMQ_PUSH);
    zmq_bind(push, addr.as_ptr());

    let mut pullers = Vec::new();
    for _ in 0..N_PULLERS {
        let p = zmq_socket(ctx, ZMQ_PULL);
        let a = CString::new(format!("tcp://127.0.0.1:{port}")).unwrap();
        zmq_connect(p, a.as_ptr());
        pullers.push(p);
    }

    std::thread::sleep(Duration::from_millis(100));
    set_timeo(push, ZMQ_SNDTIMEO, TIMEOUT_MS);
    for p in &pullers {
        set_timeo(*p, ZMQ_RCVTIMEO, 200); // short; not all pullers get messages
    }

    for _ in 0..TOTAL {
        zmq_send(push, b"w".as_ptr().cast(), 1, 0);
    }

    // Each puller receives some messages; total across all pullers must be TOTAL.
    let mut received = 0usize;
    let mut buf = [0u8; 8];
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    loop {
        for p in &pullers {
            while let rc = zmq_recv(*p, buf.as_mut_ptr().cast(), buf.len(), ZMQ_DONTWAIT)
                && rc > 0
            {
                received += 1;
            }
        }
        if received >= TOTAL {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "timeout waiting for all messages: got {received}/{TOTAL}"
        );
        std::thread::sleep(Duration::from_millis(10));
    }
    assert_eq!(received, TOTAL);

    for p in pullers {
        zmq_close(p);
    }
    zmq_close(push);
    zmq_ctx_term(ctx);
}

/// Push/pull over inproc.
#[test]
fn push_pull_inproc() {
    let ctx = zmq_ctx_new();
    let pull = zmq_socket(ctx, ZMQ_PULL);
    let push = zmq_socket(ctx, ZMQ_PUSH);

    let addr = CString::new("inproc://test-pp-inproc").unwrap();
    zmq_bind(pull, addr.as_ptr());
    zmq_connect(push, addr.as_ptr());
    std::thread::sleep(Duration::from_millis(20));

    set_timeo(pull, ZMQ_RCVTIMEO, TIMEOUT_MS);

    let payload = b"inproc-payload";
    let rc = zmq_send(push, payload.as_ptr().cast(), payload.len(), 0);
    assert_eq!(rc as usize, payload.len());

    let mut buf = [0u8; 32];
    let rc = zmq_recv(pull, buf.as_mut_ptr().cast(), buf.len(), 0);
    assert_eq!(rc as usize, payload.len());
    assert_eq!(&buf[..payload.len()], payload);

    zmq_close(push);
    zmq_close(pull);
    zmq_ctx_term(ctx);
}

/// `DONTWAIT` send to a socket with no connected peers returns `EAGAIN`
/// (the peer-side channel is empty so `send_tx.try_send` fills up, or the
/// socket returns `WouldBlock` internally).
/// Note: this tests the C-side channel HWM. Since the send pump may drain
/// the C-side channel very quickly, the exact count before EAGAIN is not
/// guaranteed — we only verify the errno is correct when EAGAIN does occur.
#[test]
fn push_dontwait_eagain_semantics() {
    let ctx = zmq_ctx_new();
    let push = zmq_socket(ctx, ZMQ_PUSH);
    let pull = zmq_socket(ctx, ZMQ_PULL);

    let port = helpers::free_port();
    let addr = CString::new(format!("tcp://127.0.0.1:{port}")).unwrap();
    zmq_bind(pull, addr.as_ptr());
    zmq_connect(push, addr.as_ptr());
    std::thread::sleep(Duration::from_millis(100));

    // Non-blocking recv when no messages: EAGAIN.
    let mut buf = [0u8; 8];
    let rc = zmq_recv(pull, buf.as_mut_ptr().cast(), buf.len(), ZMQ_DONTWAIT);
    assert_eq!(rc, -1);
    assert_eq!(omq_zmq::zmq_errno(), libc::EAGAIN);

    zmq_close(push);
    zmq_close(pull);
    zmq_ctx_term(ctx);
}
