zeromq/libzmq

//
//  Synchronized publisher in C++
//

#include <zmq.hpp>
#include <zmq_addon.hpp>
#include <iostream>
#include <chrono>
#include <thread>
//  We wait for 10 subscribers
#define SUBSCRIBERS_EXPECTED  1

int main () {
    zmq::context_t context(1);

    //  Socket to talk to clients
    zmq::socket_t publisher (context, zmq::socket_type::pub);
    publisher.bind("tcp://*:5561");

    //  Socket to receive signals
    zmq::socket_t syncservice (context, zmq::socket_type::rep);
    syncservice.bind("tcp://*:5562");

    //  Get synchronization from subscribers
    int subscribers = 0;
    while (subscribers < SUBSCRIBERS_EXPECTED) {
        zmq::message_t message;
        //  - wait for synchronization request
        syncservice.recv(message, zmq::recv_flags::none);

        //  - send synchronization reply
        zmq::message_t reply;
        syncservice.send (reply, zmq::send_flags::none);


        subscribers++;
    }
    // Give the subscribers a chance to connect, so they don't lose any messages

    // std::this_thread::sleep_for(std::chrono::milliseconds(20));

    std::cout << "Start broadcasting\n";
    //  Now broadcast exactly 1M updates followed by END
    int update_nbr;
    for (update_nbr = 0; update_nbr < 1000; update_nbr++) {    
        publisher.send (zmq::str_buffer("SYNC"), zmq::send_flags::sndmore);
        publisher.send(zmq::str_buffer("Rhubarb"));
    }
    publisher.send (zmq::str_buffer(""), zmq::send_flags::sndmore);
    publisher.send (zmq::str_buffer("END"));

    //  Give 0MQ time to flush output
    std::this_thread::sleep_for(std::chrono::seconds(1));
    return 0;
}

//
//  Synchronized subscriber in C++
//

#include <zmq.hpp>
#include <zmq_addon.hpp>
#include <iostream>
#include <chrono>
using namespace std::chrono;
int main (int argc, char *argv[])
{
   zmq::context_t context(1);

    //  First, connect our subscriber socket
    zmq::socket_t subscriber (context, zmq::socket_type::sub);
    subscriber.connect("tcp://localhost:5561");
    subscriber.set(zmq::sockopt::subscribe, "");

    //  Second, synchronize with publisher
    zmq::socket_t syncclient (context, zmq::socket_type::req);
    syncclient.connect("tcp://localhost:5562");
    std::cout << "Send\n";
    //  - send a synchronization request
    zmq::message_t request;
    syncclient.send (request, zmq::send_flags::none);

    //  - wait for synchronization reply
    //  Get the reply.
    zmq::message_t reply;
    syncclient.recv (reply, zmq::recv_flags::none);
    std::cout << "Got reply\n";
    //  Third, get our updates and report how many we got
    int update_nbr = 0;
    auto start = high_resolution_clock::now();
    while (1) {
               // Receive all parts of the message
        std::vector<zmq::message_t> recv_msgs;
        zmq::recv_result_t result =
        zmq::recv_multipart(subscriber, std::back_inserter(recv_msgs));
        assert(result && "recv failed");
        assert(*result == 2);

        if (recv_msgs[1].to_string().compare("END") == 0) {
            break;
        }
        // std::cout << "Received " <<recv_msgs[0].to_string()<<" "<<update_nbr<<std::endl;
        update_nbr++;
    }
    std::cout << "Received " << update_nbr << " updates" << std::endl;
    auto stop = high_resolution_clock::now();
    auto duration = duration_cast<microseconds>(stop - start);

// To get the value of duration use the count()
// member function on the duration object
std::cout << duration.count() << std::endl;
    return 0;
}