zmqtest.cpp

#include <zmq.hpp>
#include <sys/epoll.h>
#include <iostream>

zmq_msg_t msg()
{
    zmq_msg_t msg;
    zmq_msg_init_size(&msg, 5);
    std::memcpy(zmq_msg_data(&msg), "Hello", 5);
    return msg;
}
int msg_send(void* socket, bool more)
{
    int flags = ZMQ_DONTWAIT;
    if (more) {
        flags |= ZMQ_MORE;
    }
    zmq_msg_t m = msg();
    return zmq_msg_send(&m, socket, flags);
}

int main()
{
    void* ctx = zmq_ctx_new();
    void* socket = zmq_socket(ctx, ZMQ_PUSH);
    zmq_connect(socket, "tcp://127.0.0.1:5555");

    int hwm = 2;
    zmq_setsockopt(socket, ZMQ_SNDHWM, &hwm, sizeof(hwm));

    int fd;
    size_t size = sizeof(fd);
    zmq_getsockopt(socket, ZMQ_FD, &fd, &size);

    struct epoll_event event;
    event.events = EPOLLIN | EPOLLET;
    event.data.fd = fd;

    int epoll = epoll_create(1);
    epoll_ctl(epoll, EPOLL_CTL_ADD, fd, &event);

    struct epoll_event events[10];
    while (true)
    {
        std::cerr << "Waiting for epoll" << std::endl;
        int event_count = epoll_wait(epoll, events, 10, -1);
        int zmq_events;
        zmq_getsockopt(socket, ZMQ_EVENTS, &zmq_events, &size);
        while (zmq_events & ZMQ_POLLOUT)
        {
            std::cerr << "Before send" << std::endl;
            int res = msg_send(socket, true);
            if (res < 0)
            {
                std::cerr << "Error: " << zmq_errno() << std::endl;
            }
            std::cerr << "Sending: " << res << std::endl;
            zmq_getsockopt(socket, ZMQ_EVENTS, &zmq_events, &size);
            hwm = 1;
            zmq_setsockopt(socket, ZMQ_SNDHWM, &hwm, sizeof(hwm));
        }
    }
    return 0;
}