zeromq

#include <assert.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <zmq.hpp>
#include <iostream>

using namespace std;

// if true, the worker threads will take 5 seconds to bind the socket,
// so the clients will do the request while there's no worker.
bool DELAY_WORKER;

void* internal_client(void *arg)
{
    if (! DELAY_WORKER) sleep (5);
    zmq::context_t *ctx = (zmq::context_t*) arg;
    zmq::socket_t s (*ctx, ZMQ_REQ);
    s.connect ("tcp://localhost:12345");
    for (int i = 0; i != 10; i++) {
        zmq::message_t request (10);
        memset (request.data (), i+65, request.size ()-1);
        ((char*)request.data ())[request.size ()]=0;
        cout<<"client sending message Nr."<<i<<"with content"<<
                (char*)request.data()<<endl;
        s.send (request);
        //  Get the reply.
        zmq::message_t reply;
        cout<<"client waiting for reply ..."<<endl;
        s.recv (&reply);
        cout<<"client got reply: "<<(char*)reply.data()<<endl;
    }

    return 0;
}

void *worker_routine (void *arg)
{
    if (DELAY_WORKER) sleep (5);
    zmq::context_t *ctx = (zmq::context_t*) arg;
    zmq::socket_t s (*ctx, ZMQ_REP);
    //  Connect to the dispatcher (queue) running in the main thread.
    s.connect ("inproc://workers");
    cout<<"worker connected to queue device now"<<endl;
    while (true) {
        //  Get a request from the dispatcher.
        zmq::message_t request;
        cout << "worker waiting for request"<<endl;
        s.recv (&request);
        cout << "worker received request: " << (char*)request.data() << endl;
        sleep (1);//  to simulate actual processing.
        zmq::message_t reply (10);
        memset (reply.data (), ((char*)request.data())[0], reply.size ()-1);
        ((char*)reply.data ())[reply.size ()]=0;
        s.send (reply);
        cout << "worker sent a reply" <<endl;
    }
}

int main (int argc, char**argv)
{
    if (argc>1) {
            DELAY_WORKER = true;
            cout<<"worker thread will be delayed ==> deadlock"<<endl;
    } else {
            DELAY_WORKER = false;
            cout<<"client thread will be delayed ==> runs fine."<<endl;
    }
    zmq::context_t ctx (1);

    // worker side of queue listening for worker to connect
    zmq::socket_t workers (ctx, ZMQ_XREQ);
    workers.bind ("inproc://workers");
    // client side listening on socke5t for real clients to connect
    zmq::socket_t clients (ctx, ZMQ_XREP);
    clients.bind ("tcp://127.0.0.1:12345");

    // launch worker
    pthread_t worker;
    int rc = pthread_create (&worker, NULL, worker_routine, (void*) &ctx);
    assert (rc == 0);

    //  Launch client threads.
    rc = pthread_create (&worker, NULL, internal_client, (void*) &ctx);
    assert (rc == 0);

    cout<<"QUEUE device gets started"<<endl;
    zmq::device (ZMQ_QUEUE, clients, workers);
    return 0;
}