#include <zmq.hpp>
#include <ctime>
#include <iostream>

// in this version socket isn't eventually destroed and socketpair fd's never released. we run out of fd's or reach zmq max_sockets
static void *device (void * data) {
    zmq::context_t& context = *(static_cast<zmq::context_t*>(data));

    zmq::socket_t frontend (context, ZMQ_XREP);
    zmq::socket_t backend (context, ZMQ_XREQ);

    int zero = 0;
    frontend.setsockopt (ZMQ_LINGER, &zero, sizeof (zero));
    frontend.bind ("inproc://frontend");

    uint64_t queue_size = 1;
    backend.setsockopt (ZMQ_HWM, &queue_size, sizeof (queue_size));
    backend.setsockopt (ZMQ_LINGER, &zero, sizeof (zero));

    // we don't connect backend socket. we simulate backend service being unavailable.

    zmq::device (ZMQ_QUEUE, frontend, backend);
}

static void *client (void * data) {
    sleep(1);
    zmq::context_t& context = *static_cast<zmq::context_t*> (data);
    for (int i = 0; i < 2000; i++) {
        std::cout << i << std::endl;
        zmq::socket_t socket (context, ZMQ_REQ);
        int zero = 0;
        socket.setsockopt (ZMQ_LINGER, &zero, sizeof (zero));
        socket.connect("inproc://frontend");

        zmq::message_t message (1);
        socket.send (message);

        zmq_pollitem_t item;
        item.socket = (void*)socket;
        item.events = ZMQ_POLLIN | ZMQ_POLLERR;
        // wait for 10ms before tearing down the socket
        if (zmq::poll (&item, 1, 10000) && (item.revents & ZMQ_POLLIN)) {
            // we never should get here because noone is listening
            std::cout << "abort!" << std::endl;
            abort();
        }
    }
}

int main (void) {
    pthread_t client_thid;
    zmq::context_t context (1);

    pthread_create (&client_thid, NULL, client, &context);

    zmq::socket_t frontend (context, ZMQ_XREP);
    zmq::socket_t backend (context, ZMQ_XREQ);

    int zero = 0;
    frontend.setsockopt (ZMQ_LINGER, &zero, sizeof (zero));
    frontend.bind ("inproc://frontend");

    uint64_t queue_size = 0;
    backend.setsockopt (ZMQ_HWM, &queue_size, sizeof (queue_size));
    backend.setsockopt (ZMQ_LINGER, &zero, sizeof (zero));

    // we don't connect backend socket. we simulate backend service being unavailable.

    zmq::device (ZMQ_QUEUE, frontend, backend);

    return 0;
}