req2router.cpp

/ req4router.cpp : Defines the entry point for the console application.
//
#include <string>
#include "zmq2.h"
#include "boost\lexical_cast.hpp"
#include "boost\bind\bind.hpp"
#include "boost\thread\thread.hpp"

const char* SK_ADDR_CLI = "tcp://localhost:5380";
const char* SK_ADDR_SRV = "tcp://*:5380";

boost::mutex mio;
void dumpId(const char* tag, const char* msg)
{
    boost::lock_guard<boost::mutex> lock(mio);
    std::string id = boost::lexical_cast<std::string>(boost::this_thread::get_id());

    std::cout << id << ' ' << tag << ' ' << msg << std::endl;
}

void req4router()
{

    dumpId("REQ startup", SK_ADDR_CLI);
    zmq::context_t context(1);

    zmq::Socket skReq(context, ZMQ_REQ); // 1
    skReq.connect(SK_ADDR_CLI); // 2
    dumpId("REQ CLI on", SK_ADDR_CLI);

    std::string id = boost::lexical_cast<std::string>(boost::this_thread::get_id()); // 3
    skReq.send(id); // 4
    std::string msg = skReq.recvAsString();
    dumpId(msg.c_str(), SK_ADDR_CLI);
}

void router(int nReq)
{
    dumpId("ROUTER startup", SK_ADDR_SRV);
    zmq::context_t context(1);

    zmq::Socket skRouter(context, ZMQ_ROUTER); // 1
    skRouter.bind(SK_ADDR_SRV);
    dumpId("ROUTER SRV on", SK_ADDR_SRV);

    std::vector<zmq::Frames> msgs; // 2
    for(int i = 0; i < nReq; ++i)
    {
        msgs.push_back(skRouter.blockingRecv(3)); // 3
    }
    dumpId("all received", SK_ADDR_SRV);

    std::for_each(msgs.begin(), msgs.end(), [&skRouter](zmq::Frames& frames)
    {
        skRouter.send(frames); // 4
    });

    dumpId("ROUTER done", SK_ADDR_SRV);
}

int main(int argc, char* argv[])
{
    int nReq = atoi(argv[1]);
    boost::thread_group threads;
for(int i = 0; i < nReq; ++i) // 1
    threads.create_thread(req4router); // 2
threads.create_thread(std::bind(router, nReq)); // 3
threads.join_all();
return 0;
}