//  Espresso Pattern
//  This shows how to capture data using a pub-sub proxy

#include "czmq.h"

//  The subscriber thread requests messages starting with
//  A and B, then reads and counts incoming messages.

static void
subscriber_thread (void *args, zctx_t *ctx, void *pipe)
{
    //  Subscribe to "A" and "B"
    void *subscriber = zsocket_new (ctx, ZMQ_SUB);
    zsocket_connect (subscriber, "tcp://localhost:6001");
    zsocket_set_subscribe (subscriber, "A");
    zsocket_set_subscribe (subscriber, "B");

    int count = 0;
    while (count < 5) {
        char *string = zstr_recv (subscriber);
        if (!string)
            break;              //  Interrupted
        free (string);
        count++;
    }
    zsocket_destroy (ctx, subscriber);
}

//  .split publisher thread
//  The publisher sends random messages starting with A-J:

static void
publisher_thread (void *args, zctx_t *ctx, void *pipe)
{
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (publisher, "tcp://*:6000");

    while (!zctx_interrupted) {
        char string [10];
        sprintf (string, "%c-%05d", randof (10) + 'A', randof (100000));
        if (zstr_send (publisher, string) == -1)
            break;              //  Interrupted
        zclock_sleep (100);     //  Wait for 1/10th second
    }
}

//  .split listener thread
//  The listener receives all messages flowing through the proxy, on its
//  pipe. In CZMQ, the pipe is a pair of ZMQ_PAIR sockets that connect
//  attached child threads. In other languages your mileage may vary:

static void
listener_thread (void *args, zctx_t *ctx, void *pipe)
{
    //  Print everything that arrives on pipe
    while (true) {
        zframe_t *frame = zframe_recv (pipe);
        if (!frame)
            break;              //  Interrupted
        zframe_print (frame, NULL);
        zframe_destroy (&frame);
    }
}

//  .split main thread
//  The main task starts the subscriber and publisher, and then sets
//  itself up as a listening proxy. The listener runs as a child thread:

int main (void)
{
    //  Start child threads
    zctx_t *ctx = zctx_new ();
    zthread_fork (ctx, publisher_thread, NULL);
    zthread_fork (ctx, subscriber_thread, NULL);

    void *subscriber = zsocket_new (ctx, ZMQ_XSUB);
    zsocket_connect (subscriber, "tcp://localhost:6000");
    void *publisher = zsocket_new (ctx, ZMQ_XPUB);
    zsocket_bind (publisher, "tcp://*:6001");
    void *listener = zthread_fork (ctx, listener_thread, NULL);
    zmq_proxy (subscriber, publisher, listener);

    puts (" interrupted");
    //  Tell attached threads to exit
    zctx_destroy (&ctx);
    return 0;
}

#include <iostream>
#include <thread>
#include <zmq.hpp>
#include <string>
#include <chrono>
#include <unistd.h>

// Subscriber thread function
void subscriber_thread(zmq::context_t& ctx) {
	zmq::socket_t subscriber(ctx, ZMQ_SUB);
	subscriber.connect("tcp://localhost:6001");
	subscriber.set(zmq::sockopt::subscribe, "A");
	subscriber.set(zmq::sockopt::subscribe, "B");

	int count = 0;
	while (count < 5) {
		zmq::message_t message;
		if (subscriber.recv(message)) {
			std::string msg = std::string((char*)(message.data()), message.size());
			std::cout << "Received: " << msg << std::endl;
			count++;
		}
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
	}
}

// Publisher thread function
void publisher_thread(zmq::context_t& ctx) {
	zmq::socket_t publisher(ctx, ZMQ_PUB);
	publisher.bind("tcp://*:6000");

	while (true) {
		char string[10];
		sprintf(string, "%c-%05d", rand() % 10 + 'A', rand() % 100000);
		zmq::message_t message(string, strlen(string));
		publisher.send(message, zmq::send_flags::none);
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
	}
}

// Listener thread function
void listener_thread(zmq::context_t& ctx) {
	zmq::socket_t listener(ctx, ZMQ_PAIR);
	listener.connect("inproc://listener");

	while (true) {
		zmq::message_t message;
		if (listener.recv(message)) {
			std::string msg = std::string((char*)(message.data()), message.size());
			std::cout << "Listener Received: ";
			if (msg[0] == 0 || msg[0] == 1){
				std::cout << int(msg[0]);
				std::cout << msg[1]<< std::endl;
			} else {
				std::cout << msg << std::endl;
			}
		}
	}
}

int main() {
	zmq::context_t context(1);
	// Main thread acts as the listener proxy
	zmq::socket_t proxy(context, ZMQ_PAIR);
	proxy.bind("inproc://listener");
	zmq::socket_t xsub(context, ZMQ_XSUB);
	zmq::socket_t xpub(context, ZMQ_XPUB);
	xpub.bind("tcp://*:6001");
	sleep(1);


	// Start publisher and subscriber threads
	std::thread pub_thread(publisher_thread, std::ref(context));
	std::thread sub_thread(subscriber_thread, std::ref(context));

	// Set up listener thread
	std::thread lis_thread(listener_thread, std::ref(context));

	sleep(1);

	xsub.connect("tcp://localhost:6000");
	// Proxy messages between SUB and PUB sockets
	zmq_proxy(xsub, xpub, proxy);

	// Wait for threads to finish
	pub_thread.join();
	sub_thread.join();
	lis_thread.join();

	return 0;
}

package guide;

import java.util.Random;

import org.zeromq.*;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZThread.IAttachedRunnable;

//  Espresso Pattern
//  This shows how to capture data using a pub-sub proxy
public class espresso
{
    //  The subscriber thread requests messages starting with
    //  A and B, then reads and counts incoming messages.
    private static class Subscriber implements IAttachedRunnable
    {

        @Override
        public void run(Object[] args, ZContext ctx, Socket pipe)
        {
            //  Subscribe to "A" and "B"
            Socket subscriber = ctx.createSocket(SocketType.SUB);
            subscriber.connect("tcp://localhost:6001");
            subscriber.subscribe("A".getBytes(ZMQ.CHARSET));
            subscriber.subscribe("B".getBytes(ZMQ.CHARSET));

            int count = 0;
            while (count < 5) {
                String string = subscriber.recvStr();
                if (string == null)
                    break; //  Interrupted
                count++;
            }
            ctx.destroySocket(subscriber);
        }
    }

    //  .split publisher thread
    //  The publisher sends random messages starting with A-J:
    private static class Publisher implements IAttachedRunnable
    {
        @Override
        public void run(Object[] args, ZContext ctx, Socket pipe)
        {
            Socket publisher = ctx.createSocket(SocketType.PUB);
            publisher.bind("tcp://*:6000");
            Random rand = new Random(System.currentTimeMillis());

            while (!Thread.currentThread().isInterrupted()) {
                String string = String.format("%c-%05d", 'A' + rand.nextInt(10), rand.nextInt(100000));
                if (!publisher.send(string))
                    break; //  Interrupted
                try {
                    Thread.sleep(100); //  Wait for 1/10th second
                }
                catch (InterruptedException e) {
                }
            }
            ctx.destroySocket(publisher);
        }
    }

    //  .split listener thread
    //  The listener receives all messages flowing through the proxy, on its
    //  pipe. In CZMQ, the pipe is a pair of ZMQ_PAIR sockets that connect
    //  attached child threads. In other languages your mileage may vary:
    private static class Listener implements IAttachedRunnable
    {
        @Override
        public void run(Object[] args, ZContext ctx, Socket pipe)
        {
            //  Print everything that arrives on pipe
            while (true) {
                ZFrame frame = ZFrame.recvFrame(pipe);
                if (frame == null)
                    break; //  Interrupted
                frame.print(null);
                frame.destroy();
            }
        }
    }

    //  .split main thread
    //  The main task starts the subscriber and publisher, and then sets
    //  itself up as a listening proxy. The listener runs as a child thread:
    public static void main(String[] argv)
    {
        try (ZContext ctx = new ZContext()) {
            //  Start child threads
            ZThread.fork(ctx, new Publisher());
            ZThread.fork(ctx, new Subscriber());

            Socket subscriber = ctx.createSocket(SocketType.XSUB);
            subscriber.connect("tcp://localhost:6000");
            Socket publisher = ctx.createSocket(SocketType.XPUB);
            publisher.bind("tcp://*:6001");
            Socket listener = ZThread.fork(ctx, new Listener());
            ZMQ.proxy(subscriber, publisher, listener);

            System.out.println(" interrupted");

            // NB: child threads exit here when the context is closed
        }
    }
}

/**
*  Pub-Sub Tracing (Espresso Pattern) 
*  explained in
*  https://zguide.zeromq.org/docs/chapter5
*/

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.runListenerThread = exports.runPubThread = exports.runSubThread = void 0;

const zmq = require("zeromq"),
      publisher = new zmq.Publisher,
      pubKeypair = zmq.curveKeyPair(),
      publicKey = pubKeypair.publicKey;

var interrupted = false;

function getRandomInt(max) {
    return Math.floor(Math.random() * Math.floor(max));
}

async function runSubThread() {

    const subscriber = new zmq.Subscriber;
    const subKeypair = zmq.curveKeyPair();
    
    // Setup encryption.
    for (const s of [subscriber]) {
        subscriber.curveServerKey = publicKey; // '03P+E+f4AU6bSTcuzvgX&oGnt&Or<rN)FYIPyjQW'
        subscriber.curveSecretKey = subKeypair.secretKey;
        subscriber.curvePublicKey = subKeypair.publicKey;
    }
    
    await subscriber.connect("tcp://127.0.0.1:6000");
    console.log('subscriber connected! subscribing A,B,C and D..');
    
    //subscribe all at once - simultaneous subscriptions needed
    Promise.all([
        subscriber.subscribe("A"),
        subscriber.subscribe("B"),
        subscriber.subscribe("C"),
        subscriber.subscribe("D"),
        subscriber.subscribe("E"),
    ]);
    for await (const [msg] of subscriber) {
        console.log(`Received at subscriber: ${msg}`);
        if (interrupted) {
            await subscriber.disconnect("tcp://127.0.0.1:6000");
            await subscriber.close();
            break;
        }
    }
}

//Run the Publisher Thread!
async function runPubThread() {

   // Setup encryption.
    for (const s of [publisher]) {
        s.curveServer = true;
        s.curvePublicKey = publicKey;
        s.curveSecretKey = pubKeypair.secretKey;
    }
    await publisher.bind("tcp://127.0.0.1:6000");
    console.log(`Started publisher at tcp://127.0.0.1:6000 ..`);
    var subs = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
    while (!interrupted) { //until ctl+c 
        var str = `${subs.charAt(getRandomInt(10))}-${getRandomInt(100000).toString().padStart(6, '0')}`; //"%c-%05d";
        console.log(`Publishing ${str}`);
        if (-1 == await publisher.send(str))
            break; //Interrupted
        await new Promise(resolve => setTimeout(resolve, 1000));
    }
    //if(! publisher.closed())
    await publisher.close();
}

//Run the Pipe
async function runListenerThread() {
    //a pipe using 'Pair' which receives and transmits data
    const pipe = new zmq.Pair;
    await pipe.connect("tcp://127.0.0.1:6000");
    await pipe.bind("tcp://127.0.0.1:6001");
    
    console.log('starting pipe (using Pair)..');
    
    while (!interrupted) {
        await pipe.send(await pipe.receive());
    }

    setTimeout(() => {
        console.log('Terminating pipe..');
        pipe.close();
    }, 1000);
    
    //a pipe using 'Proxy' <= not working, but give it a try. 
    // Still working with Proxy
    /*
    const pipe = new zmq.Proxy (new zmq.Router, new zmq.Dealer)
    await pipe.backEnd.connect("tcp://127.0.0.1:6000")
    await pipe.frontEnd.bind("tcp://127.0.0.1:6001")    
    await pipe.run()
    setTimeout(() => {
        console.log('Terminating pipe..');
        await pipe.terminate()
     }, 10000);
     */
    
}

exports.runSubThread = runSubThread;
exports.runPubThread = runPubThread;
exports.runListenerThread = runListenerThread;

process.on('SIGINT', function () {
    interrupted = true;
});

process.setMaxListeners(30);

async function main() {
    //execute all at once
    Promise.all([
        runPubThread(),
        runListenerThread(),
        runSubThread(),
    ]);
}

main().catch(err => {
    console.error(err);
    process.exit(1);
});

# Espresso Pattern
# This shows how to capture data using a pub-sub proxy
#

import time

from random import randint
from string import ascii_uppercase as uppercase
from threading import Thread

import zmq
from zmq.devices import monitored_queue

from zhelpers import zpipe

# The subscriber thread requests messages starting with
# A and B, then reads and counts incoming messages.

def subscriber_thread():
    ctx = zmq.Context.instance()

    # Subscribe to "A" and "B"
    subscriber = ctx.socket(zmq.SUB)
    subscriber.connect("tcp://localhost:6001")
    subscriber.setsockopt(zmq.SUBSCRIBE, b"A")
    subscriber.setsockopt(zmq.SUBSCRIBE, b"B")

    count = 0
    while count < 5:
        try:
            msg = subscriber.recv_multipart()
        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break           # Interrupted
            else:
                raise
        count += 1

    print ("Subscriber received %d messages" % count)


# publisher thread
# The publisher sends random messages starting with A-J:

def publisher_thread():
    ctx = zmq.Context.instance()

    publisher = ctx.socket(zmq.PUB)
    publisher.bind("tcp://*:6000")

    while True:
        string = "%s-%05d" % (uppercase[randint(0,10)], randint(0,100000))
        try:
            publisher.send(string.encode('utf-8'))
        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break           # Interrupted
            else:
                raise
        time.sleep(0.1)         # Wait for 1/10th second


# listener thread
# The listener receives all messages flowing through the proxy, on its
# pipe. Here, the pipe is a pair of ZMQ_PAIR sockets that connects
# attached child threads via inproc. In other languages your mileage may vary:

def listener_thread (pipe):

    # Print everything that arrives on pipe
    while True:
        try:
            print (pipe.recv_multipart())
        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break           # Interrupted


# main thread
# The main task starts the subscriber and publisher, and then sets
# itself up as a listening proxy. The listener runs as a child thread:

def main ():

    # Start child threads
    ctx = zmq.Context.instance()
    p_thread = Thread(target=publisher_thread)
    s_thread = Thread(target=subscriber_thread)
    p_thread.start()
    s_thread.start()

    pipe = zpipe(ctx)

    subscriber = ctx.socket(zmq.XSUB)
    subscriber.connect("tcp://localhost:6000")

    publisher = ctx.socket(zmq.XPUB)
    publisher.bind("tcp://*:6001")

    l_thread = Thread(target=listener_thread, args=(pipe[1],))
    l_thread.start()

    try:
        monitored_queue(subscriber, publisher, pipe[0], b'pub', b'sub')
    except KeyboardInterrupt:
        print ("Interrupted")

    del subscriber, publisher, pipe
    ctx.term()

if __name__ == '__main__':
    main()

//  Pathological publisher
//  Sends out 1,000 topics and then one random update per second

#include "czmq.h"

int main (int argc, char *argv [])
{
    zctx_t *context = zctx_new ();
    void *publisher = zsocket_new (context, ZMQ_PUB);
    if (argc == 2)
        zsocket_bind (publisher, argv [1]);
    else
        zsocket_bind (publisher, "tcp://*:5556");

    //  Ensure subscriber connection has time to complete
    sleep (1);

    //  Send out all 1,000 topic messages
    int topic_nbr;
    for (topic_nbr = 0; topic_nbr < 1000; topic_nbr++) {
        zstr_sendfm (publisher, "%03d", topic_nbr);
        zstr_send (publisher, "Save Roger");
    }
    //  Send one random update per second
    srandom ((unsigned) time (NULL));
    while (!zctx_interrupted) {
        sleep (1);
        zstr_sendfm (publisher, "%03d", randof (1000));
        zstr_send (publisher, "Off with his head!");
    }
    zctx_destroy (&context);
    return 0;
}

//  Pathological publisher
//  Sends out 1,000 topics and then one random update per second

#include <thread>
#include <chrono>
#include "zhelpers.hpp"

int main (int argc, char *argv [])
{
    zmq::context_t context(1);
    zmq::socket_t publisher(context, ZMQ_PUB);

    //  Initialize random number generator
    srandom ((unsigned) time (NULL));

    if (argc == 2)
        publisher.bind(argv [1]);
    else
        publisher.bind("tcp://*:5556");

    //  Ensure subscriber connection has time to complete
    std::this_thread::sleep_for(std::chrono::seconds(1));

    //  Send out all 1,000 topic messages
    int topic_nbr;
    for (topic_nbr = 0; topic_nbr < 1000; topic_nbr++) {
        std::stringstream ss;
        ss << std::dec << std::setw(3) << std::setfill('0') << topic_nbr;

        s_sendmore (publisher, ss.str());
        s_send (publisher, std::string("Save Roger"));
    }

    //  Send one random update per second
    while (1) {

        std::this_thread::sleep_for(std::chrono::seconds(1));
        std::stringstream ss;
        ss << std::dec << std::setw(3) << std::setfill('0') << within(1000);

        s_sendmore (publisher, ss.str());
        s_send (publisher, std::string("Off with his head!"));
    }
    return 0;
}

package guide;

import java.util.Random;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;

//  Pathological publisher
//  Sends out 1,000 topics and then one random update per second
public class pathopub
{
    public static void main(String[] args) throws Exception
    {
        try (ZContext context = new ZContext()) {
            Socket publisher = context.createSocket(SocketType.PUB);
            if (args.length == 1)
                publisher.connect(args[0]);
            else publisher.bind("tcp://*:5556");

            //  Ensure subscriber connection has time to complete
            Thread.sleep(1000);

            //  Send out all 1,000 topic messages
            int topicNbr;
            for (topicNbr = 0; topicNbr < 1000; topicNbr++) {
                publisher.send(String.format("%03d", topicNbr), ZMQ.SNDMORE);
                publisher.send("Save Roger");
            }
            //  Send one random update per second
            Random rand = new Random(System.currentTimeMillis());
            while (!Thread.currentThread().isInterrupted()) {
                Thread.sleep(1000);
                publisher.send(
                    String.format("%03d", rand.nextInt(1000)), ZMQ.SNDMORE
                );
                publisher.send("Off with his head!");
            }
        }
    }
}

#
# Pathological publisher
# Sends out 1,000 topics and then one random update per second
#

import sys
import time

from random import randint

import zmq

def main(url=None):
    ctx = zmq.Context.instance()
    publisher = ctx.socket(zmq.PUB)
    if url:
        publisher.bind(url)
    else:
        publisher.bind("tcp://*:5556")
    # Ensure subscriber connection has time to complete
    time.sleep(1)

    # Send out all 1,000 topic messages
    for topic_nbr in range(1000):
        publisher.send_multipart([
            b"%03d" % topic_nbr,
            b"Save Roger",
        ])

    while True:
        # Send one random update per second
        try:
            time.sleep(1)
            publisher.send_multipart([
                b"%03d" % randint(0,999),
                b"Off with his head!",
            ])
        except KeyboardInterrupt:
            print "interrupted"
            break

if __name__ == '__main__':
    main(sys.argv[1] if len(sys.argv) > 1 else None)

#!/usr/bin/env ruby

#
# Pathological publisher
# Sends out 1,000 topics and then one random update per second
#
require 'ffi-rzmq'
context = ZMQ::Context.new

TOPIC_COUNT = 1_000

publisher = context.socket(ZMQ::PUB)
if ARGV[0]
  publisher.bind(ARGV[0])
else
  publisher.bind("tcp://*:5556")
end

# Ensure subscriber connection has time to complete
sleep 1

TOPIC_COUNT.times do |n|
  topic = "%03d" % [n]
  publisher.send_strings([topic, "Save Roger"])
end

loop do
  sleep 1
  topic = "%03d" % [rand(1000)]
  publisher.send_strings([topic, "Off with his head!"])
end

//  Pathological subscriber
//  Subscribes to one random topic and prints received messages

#include "czmq.h"

int main (int argc, char *argv [])
{
    zctx_t *context = zctx_new ();
    void *subscriber = zsocket_new (context, ZMQ_SUB);
    if (argc == 2)
        zsocket_connect (subscriber, argv [1]);
    else
        zsocket_connect (subscriber, "tcp://localhost:5556");

    srandom ((unsigned) time (NULL));
    char subscription [5];
    sprintf (subscription, "%03d", randof (1000));
    zsocket_set_subscribe (subscriber, subscription);
    
    while (true) {
        char *topic = zstr_recv (subscriber);
        if (!topic)
            break;
        char *data = zstr_recv (subscriber);
        assert (streq (topic, subscription));
        puts (data);
        free (topic);
        free (data);
    }
    zctx_destroy (&context);
    return 0;
}

//  Pathological subscriber
//  Subscribes to one random topic and prints received messages

#include "zhelpers.hpp"

int main (int argc, char *argv [])
{
    zmq::context_t context(1);
    zmq::socket_t subscriber (context, ZMQ_SUB);

    //  Initialize random number generator
    srandom ((unsigned) time (NULL));

    if (argc == 2)
        subscriber.connect(argv [1]);
    else
        subscriber.connect("tcp://localhost:5556");

    std::stringstream ss;
    ss << std::dec << std::setw(3) << std::setfill('0') << within(1000);
    std::cout << "topic:" << ss.str() << std::endl;

    subscriber.set( zmq::sockopt::subscribe, ss.str().c_str());

    while (1) {
		std::string topic = s_recv (subscriber);
		std::string data = s_recv (subscriber);
        if (topic != ss.str())
            break;
        std::cout << data << std::endl;
    }
    return 0;
}

package guide;

import java.util.Random;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;

//  Pathological subscriber
//  Subscribes to one random topic and prints received messages
public class pathosub
{
    public static void main(String[] args)
    {
        try (ZContext context = new ZContext()) {
            Socket subscriber = context.createSocket(SocketType.SUB);
            if (args.length == 1)
                subscriber.connect(args[0]);
            else subscriber.connect("tcp://localhost:5556");

            Random rand = new Random(System.currentTimeMillis());
            String subscription = String.format("%03d", rand.nextInt(1000));
            subscriber.subscribe(subscription.getBytes(ZMQ.CHARSET));

            while (true) {
                String topic = subscriber.recvStr();
                if (topic == null)
                    break;
                String data = subscriber.recvStr();
                assert (topic.equals(subscription));
                System.out.println(data);
            }
        }
    }
}

#
# Pathological subscriber
# Subscribes to one random topic and prints received messages
#

import sys
import time

from random import randint

import zmq

def main(url=None):
    ctx = zmq.Context.instance()
    subscriber = ctx.socket(zmq.SUB)
    if url is None:
        url = "tcp://localhost:5556"
    subscriber.connect(url)

    subscription = b"%03d" % randint(0,999)
    subscriber.setsockopt(zmq.SUBSCRIBE, subscription)

    while True:
        topic, data = subscriber.recv_multipart()
        assert topic == subscription
        print data

if __name__ == '__main__':
    main(sys.argv[1] if len(sys.argv) > 1 else None)

#!/usr/bin/env ruby

#
# Pathological subscriber
# Subscribes to one random topic and prints received messages
#
require 'ffi-rzmq'
context = ZMQ::Context.new

subscriber = context.socket(ZMQ::SUB)
subscriber.connect(ARGV[0] || "tcp://localhost:5556")

topic = "%03d" % [rand(1000)]
subscriber.setsockopt(ZMQ::SUBSCRIBE, topic)

loop do
  subscriber.recv_strings(parts = [])
  topic, data = parts
  puts "#{topic}: #{data}"
end

//  Last value cache
//  Uses XPUB subscription messages to re-send data

#include "czmq.h"

int main (void)
{
    zctx_t *context = zctx_new ();
    void *frontend = zsocket_new (context, ZMQ_SUB);
    zsocket_connect (frontend, "tcp://*:5557");
    void *backend = zsocket_new (context, ZMQ_XPUB);
    zsocket_bind (backend, "tcp://*:5558");

    //  Subscribe to every single topic from publisher
    zsocket_set_subscribe (frontend, "");

    //  Store last instance of each topic in a cache
    zhash_t *cache = zhash_new ();

    //  .split main poll loop
    //  We route topic updates from frontend to backend, and
    //  we handle subscriptions by sending whatever we cached,
    //  if anything:
    while (true) {
        zmq_pollitem_t items [] = {
            { frontend, 0, ZMQ_POLLIN, 0 },
            { backend,  0, ZMQ_POLLIN, 0 }
        };
        if (zmq_poll (items, 2, 1000 * ZMQ_POLL_MSEC) == -1)
            break;              //  Interrupted

        //  Any new topic data we cache and then forward
        if (items [0].revents & ZMQ_POLLIN) {
            char *topic = zstr_recv (frontend);
            char *current = zstr_recv (frontend);
            if (!topic)
                break;
            char *previous = zhash_lookup (cache, topic);
            if (previous) {
                zhash_delete (cache, topic);
                free (previous);
            }
            zhash_insert (cache, topic, current);
            zstr_sendm (backend, topic);
            zstr_send (backend, current);
            free (topic);
        }
        //  .split handle subscriptions
        //  When we get a new subscription, we pull data from the cache:
        if (items [1].revents & ZMQ_POLLIN) {
            zframe_t *frame = zframe_recv (backend);
            if (!frame)
                break;
            //  Event is one byte 0=unsub or 1=sub, followed by topic
            byte *event = zframe_data (frame);
            if (event [0] == 1) {
                char *topic = zmalloc (zframe_size (frame));
                memcpy (topic, event + 1, zframe_size (frame) - 1);
                printf ("Sending cached topic %s\n", topic);
                char *previous = zhash_lookup (cache, topic);
                if (previous) {
                    zstr_sendm (backend, topic);
                    zstr_send (backend, previous);
                }
                free (topic);
            }
            zframe_destroy (&frame);
        }
    }
    zctx_destroy (&context);
    zhash_destroy (&cache);
    return 0;
}

//  Last value cache
//  Uses XPUB subscription messages to re-send data

#include <unordered_map>
#include "zhelpers.hpp"

int main ()
{
    zmq::context_t context(1);
    zmq::socket_t  frontend(context, ZMQ_SUB);
    zmq::socket_t  backend(context, ZMQ_XPUB);

    frontend.connect("tcp://localhost:5557");
    backend.bind("tcp://*:5558");

    //  Subscribe to every single topic from publisher
    frontend.set(zmq::sockopt::subscribe, "");

    //  Store last instance of each topic in a cache
    std::unordered_map<std::string, std::string> cache_map;

    zmq::pollitem_t items[2] = {
        { static_cast<void*>(frontend), 0, ZMQ_POLLIN, 0 },
        { static_cast<void*>(backend), 0, ZMQ_POLLIN, 0 }
    };

    //  .split main poll loop
    //  We route topic updates from frontend to backend, and we handle
    //  subscriptions by sending whatever we cached, if anything:
    while (1)
    {
        if (zmq::poll(items, 2, 1000) == -1)
            break; //  Interrupted

        //  Any new topic data we cache and then forward
        if (items[0].revents & ZMQ_POLLIN)
        {
            std::string topic = s_recv(frontend);
            std::string data  = s_recv(frontend);

            if (topic.empty())
                break;

            cache_map[topic] = data;

            s_sendmore(backend, topic);
            s_send(backend, data);
        }

        //  .split handle subscriptions
        //  When we get a new subscription, we pull data from the cache:
        if (items[1].revents & ZMQ_POLLIN) {
            zmq::message_t msg;

            backend.recv(&msg);
            if (msg.size() == 0)
                break;

            //  Event is one byte 0=unsub or 1=sub, followed by topic
            uint8_t *event = (uint8_t *)msg.data();
            if (event[0] == 1) {
                std::string topic((char *)(event+1), msg.size()-1);

                auto i = cache_map.find(topic);
                if (i != cache_map.end())
                {
                    s_sendmore(backend, topic);
                    s_send(backend, i->second);
                }
            }
        }
    }

    return 0;
}

package guide;

import java.util.HashMap;
import java.util.Map;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

//  Last value cache
//  Uses XPUB subscription messages to re-send data
public class lvcache
{
    public static void main(String[] args)
    {
        try (ZContext context = new ZContext()) {
            Socket frontend = context.createSocket(SocketType.SUB);
            frontend.bind("tcp://*:5557");
            Socket backend = context.createSocket(SocketType.XPUB);
            backend.bind("tcp://*:5558");

            //  Subscribe to every single topic from publisher
            frontend.subscribe(ZMQ.SUBSCRIPTION_ALL);

            //  Store last instance of each topic in a cache
            Map<String, String> cache = new HashMap<String, String>();

            Poller poller = context.createPoller(2);
            poller.register(frontend, Poller.POLLIN);
            poller.register(backend, Poller.POLLIN);

            //  .split main poll loop
            //  We route topic updates from frontend to backend, and we handle
            //  subscriptions by sending whatever we cached, if anything:
            while (true) {
                if (poller.poll(1000) == -1)
                    break; //  Interrupted

                //  Any new topic data we cache and then forward
                if (poller.pollin(0)) {
                    String topic = frontend.recvStr();
                    String current = frontend.recvStr();

                    if (topic == null)
                        break;
                    cache.put(topic, current);
                    backend.sendMore(topic);
                    backend.send(current);
                }
                //  .split handle subscriptions
                //  When we get a new subscription, we pull data from the cache:
                if (poller.pollin(1)) {
                    ZFrame frame = ZFrame.recvFrame(backend);
                    if (frame == null)
                        break;
                    //  Event is one byte 0=unsub or 1=sub, followed by topic
                    byte[] event = frame.getData();
                    if (event[0] == 1) {
                        String topic = new String(event, 1, event.length - 1, ZMQ.CHARSET);
                        System.out.printf("Sending cached topic %s\n", topic);
                        String previous = cache.get(topic);
                        if (previous != null) {
                            backend.sendMore(topic);
                            backend.send(previous);
                        }
                    }
                    frame.destroy();
                }
            }
        }
    }
}

//  Last value cache
//  Uses XPUB subscription messages to re-send data

var zmq = require('zeromq');
var frontEnd = zmq.socket('sub');
var backend = zmq.socket('xpub');
var cache = {};

frontEnd.connect('tcp://127.0.0.1:5557');
frontEnd.subscribe('');

backend.bindSync('tcp://*:5558');

frontEnd.on('message', function(topic, message) {
  cache[topic] = message;
  backend.send([topic, message]);
});

backend.on('message', function(frame) {
  //  frame is one byte 0=unsub or 1=sub, followed by topic
  if (frame[0] === 1) {
    var topic = frame.slice(1);
    var previous = cache[topic];

    console.log('Sending cached topic ' + topic);

    if (typeof previous !== 'undefined') {
      backend.send([topic, previous]);
    }
  }
});

process.on('SIGINT', function() {
  frontEnd.close();
  backend.close();
  console.log('\nClosed')
});

#
# Last value cache
# Uses XPUB subscription messages to re-send data
#

import zmq

def main():
    ctx = zmq.Context.instance()
    frontend = ctx.socket(zmq.SUB)
    frontend.connect("tcp://*:5557")
    backend = ctx.socket(zmq.XPUB)
    backend.bind("tcp://*:5558")

    # Subscribe to every single topic from publisher
    frontend.setsockopt(zmq.SUBSCRIBE, b"")

    # Store last instance of each topic in a cache
    cache = {}

    # main poll loop
    # We route topic updates from frontend to backend, and
    # we handle subscriptions by sending whatever we cached,
    # if anything:
    poller = zmq.Poller()
    poller.register(frontend, zmq.POLLIN)
    poller.register(backend, zmq.POLLIN)
    while True:

        try:
            events = dict(poller.poll(1000))
        except KeyboardInterrupt:
            print("interrupted")
            break

        # Any new topic data we cache and then forward
        if frontend in events:
            msg = frontend.recv_multipart()
            topic, current = msg
            cache[topic] = current
            backend.send_multipart(msg)

        # handle subscriptions
        # When we get a new subscription we pull data from the cache:
        if backend in events:
            event = backend.recv()
            # Event is one byte 0=unsub or 1=sub, followed by topic
            if event[0] == 1:
                topic = event[1:]
                if topic in cache:
                    print ("Sending cached topic %s" % topic)
                    backend.send_multipart([ topic, cache[topic] ])

if __name__ == '__main__':
    main()

#!/usr/bin/env ruby

#
# Last value cache
# Uses XPUB subscription messages to re-send data
#
require 'ffi-rzmq'
context = ZMQ::Context.new

frontend = context.socket(ZMQ::SUB)
frontend.connect("tcp://*:5557")
backend = context.socket(ZMQ::XPUB)
backend.bind("tcp://*:5558")

# Subscribe to every single topic from publisher
frontend.setsockopt(ZMQ::SUBSCRIBE, "")

# Store last instance of each topic in a cache
cache = {}

# We route topic updates from frontend to backend, and we handle subscriptions
# by sending whatever we cached, if anything:
poller = ZMQ::Poller.new
[frontend, backend].each { |sock| poller.register_readable sock }
loop do
  poller.poll(1000)
  poller.readables.each do |sock|
    if sock == frontend
      # Any new topic data we cache and then forward
      frontend.recv_strings(parts = [])
      topic, data = parts
      cache[topic] = data
      backend.send_strings(parts)
    elsif sock == backend
      # When we get a new subscription we pull data from the cache:
      backend.recv_strings(parts = [])
      event, _ = parts
      # Event is one byte 0=unsub or 1=sub, followed by topic
      if event[0].ord == 1
        topic = event[1..-1]
        puts "Sending cached topic #{topic}"
        previous = cache[topic]
        backend.send_strings([topic, previous]) if previous
      end
    end
  end
end

//  Suicidal Snail

#include "czmq.h"

//  This is our subscriber. It connects to the publisher and subscribes
//  to everything. It sleeps for a short time between messages to
//  simulate doing too much work. If a message is more than one second
//  late, it croaks.

#define MAX_ALLOWED_DELAY   1000    //  msecs

static void
subscriber (void *args, zctx_t *ctx, void *pipe)
{
    //  Subscribe to everything
    void *subscriber = zsocket_new (ctx, ZMQ_SUB);
    zsocket_set_subscribe (subscriber, "");
    zsocket_connect (subscriber, "tcp://localhost:5556");

    //  Get and process messages
    while (true) {
        char *string = zstr_recv (subscriber);
        printf("%s\n", string);
        int64_t clock;
        int terms = sscanf (string, "%" PRId64, &clock);
        assert (terms == 1);
        free (string);

        //  Suicide snail logic
        if (zclock_time () - clock > MAX_ALLOWED_DELAY) {
            fprintf (stderr, "E: subscriber cannot keep up, aborting\n");
            break;
        }
        //  Work for 1 msec plus some random additional time
        zclock_sleep (1 + randof (2));
    }
    zstr_send (pipe, "gone and died");
}

//  .split publisher task
//  This is our publisher task. It publishes a time-stamped message to its
//  PUB socket every millisecond:

static void
publisher (void *args, zctx_t *ctx, void *pipe)
{
    //  Prepare publisher
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (publisher, "tcp://*:5556");

    while (true) {
        //  Send current clock (msecs) to subscribers
        char string [20];
        sprintf (string, "%" PRId64, zclock_time ());
        zstr_send (publisher, string);
        char *signal = zstr_recv_nowait (pipe);
        if (signal) {
            free (signal);
            break;
        }
        zclock_sleep (1);            //  1msec wait
    }
}

//  .split main task
//  The main task simply starts a client and a server, and then
//  waits for the client to signal that it has died:

int main (void)
{
    zctx_t *ctx = zctx_new ();
    void *pubpipe = zthread_fork (ctx, publisher, NULL);
    void *subpipe = zthread_fork (ctx, subscriber, NULL);
    free (zstr_recv (subpipe));
    zstr_send (pubpipe, "break");
    zclock_sleep (100);
    zctx_destroy (&ctx);
    return 0;
}

//
// Suicidal Snail
//
// Andreas Hoelzlwimmer <andreas.hoelzlwimmer@fh-hagenberg.at>
#include "zhelpers.hpp"
#include <thread>

// ---------------------------------------------------------------------
// This is our subscriber
// It connects to the publisher and subscribes to everything. It
// sleeps for a short time between messages to simulate doing too
// much work. If a message is more than 1 second late, it croaks.

#define MAX_ALLOWED_DELAY 1000 // msecs

namespace {
	bool Exit = false;
};

static void *
subscriber () {
    zmq::context_t context(1);

    // Subscribe to everything
    zmq::socket_t subscriber(context, ZMQ_SUB);
    subscriber.connect("tcp://localhost:5556");
    subscriber.set(zmq::sockopt::subscribe, "");

    std::stringstream ss;
    // Get and process messages
    while (1) {
        ss.clear();
        ss.str(s_recv (subscriber));
        int64_t clock;
        assert ((ss >> clock));
	const auto delay = s_clock () - clock;
        // Suicide snail logic
        if (delay> MAX_ALLOWED_DELAY) {
            std::cerr << "E: subscriber cannot keep up, aborting. Delay=" <<delay<< std::endl;
            break;
        }
        // Work for 1 msec plus some random additional time
        s_sleep(1000*(1+within(2)));
    }
    Exit = true;
    return (NULL);
}


// ---------------------------------------------------------------------
// This is our server task
// It publishes a time-stamped message to its pub socket every 1ms.

static void *
publisher () {
    zmq::context_t context (1);

    // Prepare publisher
    zmq::socket_t publisher(context, ZMQ_PUB);
    publisher.bind("tcp://*:5556");

    std::stringstream ss;

    while (!Exit) {
        // Send current clock (msecs) to subscribers
        ss.str("");
        ss << s_clock();
        s_send (publisher, ss.str());

        s_sleep(1);
    }
    return 0;
}


// This main thread simply starts a client, and a server, and then
// waits for the client to croak.
//
int main (void)
{
    std::thread server_thread(&publisher);
    std::thread client_thread(&subscriber);
    client_thread.join();
    server_thread.join();
    return 0;
}

package guide;

import java.util.Random;

//  Suicidal Snail

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZThread;
import org.zeromq.ZThread.IAttachedRunnable;

public class suisnail
{
    private static final long MAX_ALLOWED_DELAY = 1000; //  msecs
    private static Random rand = new Random(System.currentTimeMillis());

    //  This is our subscriber. It connects to the publisher and subscribes to
    //  everything. It sleeps for a short time between messages to simulate
    //  doing too much work. If a message is more than one second late, it
    //  croaks.
    private static class Subscriber implements IAttachedRunnable
    {
        @Override
        public void run(Object[] args, ZContext ctx, Socket pipe)
        {
            //  Subscribe to everything
            Socket subscriber = ctx.createSocket(SocketType.SUB);
            subscriber.subscribe(ZMQ.SUBSCRIPTION_ALL);
            subscriber.connect("tcp://localhost:5556");

            //  Get and process messages
            while (true) {
                String string = subscriber.recvStr();
                System.out.printf("%s\n", string);
                long clock = Long.parseLong(string);

                //  Suicide snail logic
                if (System.currentTimeMillis() - clock > MAX_ALLOWED_DELAY) {
                    System.err.println(
                        "E: subscriber cannot keep up, aborting"
                    );
                    break;
                }
                //  Work for 1 msec plus some random additional time
                try {
                    Thread.sleep(1000 + rand.nextInt(2000));
                }
                catch (InterruptedException e) {
                    break;
                }
            }
            pipe.send("gone and died");
        }
    }

    //  .split publisher task
    //  This is our publisher task. It publishes a time-stamped message to its
    //  PUB socket every millisecond:
    private static class Publisher implements IAttachedRunnable
    {
        @Override
        public void run(Object[] args, ZContext ctx, Socket pipe)
        {
            //  Prepare publisher
            Socket publisher = ctx.createSocket(SocketType.PUB);
            publisher.bind("tcp://*:5556");

            while (true) {
                //  Send current clock (msecs) to subscribers
                String string = String.format("%d", System.currentTimeMillis());
                publisher.send(string);
                String signal = pipe.recvStr(ZMQ.DONTWAIT);
                if (signal != null) {
                    break;
                }
                try {
                    Thread.sleep(1);
                }
                catch (InterruptedException e) {
                }
            }
        }
    }

    //  .split main task
    //  The main task simply starts a client and a server, and then waits for
    //  the client to signal that it has died:
    public static void main(String[] args) throws Exception
    {
        try (ZContext ctx = new ZContext()) {
            Socket pubpipe = ZThread.fork(ctx, new Publisher());
            Socket subpipe = ZThread.fork(ctx, new Subscriber());
            subpipe.recvStr();
            pubpipe.send("break");
            Thread.sleep(100);
        }
    }
}

--
--  Suicidal Snail
--
--  Author: Robert G. Jakabosky <bobby@sharedrealm.com>
--
require"zmq"
require"zmq.threads"
require"zhelpers"

--  ---------------------------------------------------------------------
--  This is our subscriber
--  It connects to the publisher and subscribes to everything. It
--  sleeps for a short time between messages to simulate doing too
--  much work. If a message is more than 1 second late, it croaks.

local subscriber = [[
    require"zmq"
    require"zhelpers"

    local MAX_ALLOWED_DELAY    = 1000    --  msecs

    local context = zmq.init(1)

    --  Subscribe to everything
    local subscriber = context:socket(zmq.SUB)
    subscriber:connect("tcp://localhost:5556")
    subscriber:setopt(zmq.SUBSCRIBE, "", 0)

    --  Get and process messages
    while true do
        local msg = subscriber:recv()
        local clock = tonumber(msg)

        --  Suicide snail logic
        if (s_clock () - clock > MAX_ALLOWED_DELAY) then
            fprintf (io.stderr, "E: subscriber cannot keep up, aborting\n")
            break
        end
        --  Work for 1 msec plus some random additional time
        s_sleep (1 + randof (2))
    end
    subscriber:close()
    context:term()
]]

--  ---------------------------------------------------------------------
--  This is our server task
--  It publishes a time-stamped message to its pub socket every 1ms.

local publisher = [[
    require"zmq"
    require"zhelpers"

    local context = zmq.init(1)

    --  Prepare publisher
    local publisher = context:socket(zmq.PUB)
    publisher:bind("tcp://*:5556")

    while true do
        --  Send current clock (msecs) to subscribers
        publisher:send(tostring(s_clock()))
        s_sleep (1);            --  1msec wait
    end
    publisher:close()
    context:term()
]]

--  This main thread simply starts a client, and a server, and then
--  waits for the client to croak.
--

local server_thread = zmq.threads.runstring(nil, publisher)
server_thread:start(true)

local client_thread = zmq.threads.runstring(nil, subscriber)
client_thread:start()
client_thread:join()

<?php
/* Suicidal Snail
 *
 *  @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/

/*  ---------------------------------------------------------------------
 * This is our subscriber
 * It connects to the publisher and subscribes to everything. It
 * sleeps for a short time between messages to simulate doing too
 * much work. If a message is more than 1 second late, it croaks.
 */
define("MAX_ALLOWED_DELAY", 100); // msecs

function subscriber()
{
    $context = new ZMQContext();

    // Subscribe to everything
    $subscriber = new ZMQSocket($context, ZMQ::SOCKET_SUB);
    $subscriber->connect("tcp://localhost:5556");
    $subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, "");

    //  Get and process messages
    while (true) {
        $clock = $subscriber->recv();
        //  Suicide snail logic
        if (microtime(true)*100 - $clock*100 > MAX_ALLOWED_DELAY) {
            echo "E: subscriber cannot keep up, aborting", PHP_EOL;
            break;
        }

        //  Work for 1 msec plus some random additional time
        usleep(1000 + rand(0, 1000));
    }
}

/* ---------------------------------------------------------------------
 * This is our server task
 * It publishes a time-stamped message to its pub socket every 1ms.
 */
function publisher()
{
    $context = new ZMQContext();

    //  Prepare publisher
    $publisher = new ZMQSocket($context, ZMQ::SOCKET_PUB);
    $publisher->bind("tcp://*:5556");

    while (true) {
        //  Send current clock (msecs) to subscribers
        $publisher->send(microtime(true));
        usleep(1000); //  1msec wait
    }
}

/*
 * This main thread simply starts a client, and a server, and then
 * waits for the client to croak.
 */
$pid = pcntl_fork();
if ($pid == 0) {
    publisher();
    exit();
}

$pid = pcntl_fork();
if ($pid == 0) {
    subscriber();
    exit();
}

"""
Suicidal Snail

Author: Min RK <benjaminrk@gmail.com>
"""
from __future__ import print_function
import sys
import threading
import time
from pickle import dumps, loads
import random

import zmq

from zhelpers import zpipe

# ---------------------------------------------------------------------
# This is our subscriber
# It connects to the publisher and subscribes to everything. It
# sleeps for a short time between messages to simulate doing too
# much work. If a message is more than 1 second late, it croaks.

MAX_ALLOWED_DELAY = 1.0    # secs

def subscriber(pipe):
    # Subscribe to everything
    ctx = zmq.Context.instance()
    sub = ctx.socket(zmq.SUB)
    sub.setsockopt(zmq.SUBSCRIBE, b'')
    sub.connect("tcp://localhost:5556")

    # Get and process messages
    while True:
        clock = loads(sub.recv())
        # Suicide snail logic
        if (time.time() - clock > MAX_ALLOWED_DELAY):
            print("E: subscriber cannot keep up, aborting", file=sys.stderr)
            break

        # Work for 1 msec plus some random additional time
        time.sleep(1e-3 * (1+2*random.random()))
    pipe.send(b"gone and died")


# ---------------------------------------------------------------------
# This is our server task
# It publishes a time-stamped message to its pub socket every 1ms.

def publisher(pipe):
    # Prepare publisher
    ctx = zmq.Context.instance()
    pub = ctx.socket(zmq.PUB)
    pub.bind("tcp://*:5556")

    while True:
        # Send current clock (secs) to subscribers
        pub.send(dumps(time.time()))
        try:
            signal = pipe.recv(zmq.DONTWAIT)
        except zmq.ZMQError as e:
            if e.errno == zmq.EAGAIN:
                # nothing to recv
                pass
            else:
                raise
        else:
            # received break message
            break
        time.sleep(1e-3)            # 1msec wait


# This main thread simply starts a client, and a server, and then
# waits for the client to signal it's died.

def main():
    ctx = zmq.Context.instance()
    pub_pipe, pub_peer = zpipe(ctx)
    sub_pipe, sub_peer = zpipe(ctx)

    pub_thread = threading.Thread(target=publisher, args=(pub_peer,))
    pub_thread.daemon=True
    pub_thread.start()
    sub_thread = threading.Thread(target=subscriber, args=(sub_peer,))
    sub_thread.daemon=True
    sub_thread.start()
    # wait for sub to finish
    sub_pipe.recv()
    # tell pub to halt
    pub_pipe.send(b"break")
    time.sleep(0.1)

if __name__ == '__main__':
    main()

#
#  Suicidal Snail
#

package require zmq

if {[llength $argv] == 0} {
    set argv [list driver]
} elseif {[llength $argv] != 1} {
    puts "Usage: suisnail.tcl <driver|sub|pub>"
    exit 1
}

lassign $argv what
set MAX_ALLOWED_DELAY   1000    ;#  msecs

set tclsh [info nameofexecutable]
expr {srand([pid])}

switch -exact -- $what {
    sub {
	#  This is our subscriber
	#  It connects to the publisher and subscribes to everything. It
	#  sleeps for a short time between messages to simulate doing too
	#  much work. If a message is more than 1 second late, it croaks.
	zmq context context
	zmq socket subpipe context PAIR
	subpipe connect "ipc://subpipe.ipc"
	#  Subscribe to everything
	zmq socket subscriber context SUB
	subscriber setsockopt SUBSCRIBE ""
	subscriber connect "tcp://localhost:5556"

	#  Get and process messages
	while {1} {
	    set string [subscriber recv]
	    puts "$string (delay = [expr {[clock milliseconds] - $string}])"

	    if {[clock milliseconds] - $string > $::MAX_ALLOWED_DELAY} {
		puts stderr "E: subscriber cannot keep up, aborting"
		break
	    }

	    after [expr {1+int(rand()*2)}]
	}

	subpipe send "gone and died"

	subscriber close
	subpipe close
	context term
    }
    pub {
	#  This is our server task
	#  It publishes a time-stamped message to its pub socket every 1ms.
	zmq context context
	zmq socket pubpipe context PAIR
	pubpipe connect "ipc://pubpipe.ipc"

	#  Prepare publisher
	zmq socket publisher context PUB
	publisher bind "tcp://*:5556"

	while {1} {
	    #  Send current clock (msecs) to subscribers
	    publisher send [clock milliseconds]

	    if {"POLLIN" in [pubpipe getsockopt EVENTS]} {
		break
	    }

	    after 1 ;# 1msec wait
	}

	publisher close
	pubpipe close
	context term
    }
    driver {
	zmq context context
	zmq socket pubpipe context PAIR
	pubpipe bind "ipc://pubpipe.ipc"
	zmq socket subpipe context PAIR
	subpipe bind "ipc://subpipe.ipc"
	puts "Start publisher, output redirected to publisher.log"
	exec $tclsh suisnail.tcl pub > publisher.log 2>@1 &
	puts "Start subscriber, output redirected to subscriber.log"
	exec $tclsh suisnail.tcl sub > subscriber.log 2>@1 &
	subpipe recv
	pubpipe send "break"
	after 100
	pubpipe close
	subpipe close
	context term
    }
}

//  Clone server Model One

#include "kvsimple.c"

int main (void)
{
    //  Prepare our context and publisher socket
    zctx_t *ctx = zctx_new ();
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (publisher, "tcp://*:5556");
    zclock_sleep (200);

    zhash_t *kvmap = zhash_new ();
    int64_t sequence = 0;
    srandom ((unsigned) time (NULL));

    while (!zctx_interrupted) {
        //  Distribute as key-value message
        kvmsg_t *kvmsg = kvmsg_new (++sequence);
        kvmsg_fmt_key  (kvmsg, "%d", randof (10000));
        kvmsg_fmt_body (kvmsg, "%d", randof (1000000));
        kvmsg_send     (kvmsg, publisher);
        kvmsg_store   (&kvmsg, kvmap);
    }
    printf (" Interrupted\n%d messages out\n", (int) sequence);
    zhash_destroy (&kvmap);
    zctx_destroy (&ctx);
    return 0;
}

#include <iostream>
#include <unordered_map>
#include "kvsimple.hpp"

using namespace std;

int main() {
	// Prepare our context and publisher socket
	zmq::context_t ctx(1);
	zmq::socket_t publisher(ctx, ZMQ_PUB);
	publisher.bind("tcp://*:5555");
	s_sleep(5000); // Sleep for a short while to allow connections to be established

	// Initialize key-value map and sequence
	unordered_map<string,string> kvmap;
	int64_t sequence = 0;
	srand(time(NULL));

	s_catch_signals();
	while (!s_interrupted) {
		// Distribute as key-value message
		string key = to_string(within(10000));
		string body = to_string(within(1000000));
		kvmsg kv(key, sequence, (unsigned char *)body.c_str());
		kv.send(publisher); // Send key-value message
		// Store key-value pair in map
		kvmap[key] = body;
		sequence++;

		// Sleep for a short while before sending the next message
		s_sleep(1000);
	}

	cout << "Interrupted" << endl;
	cout << sequence << " messages out" << endl;
	return 0;
}

package guide;

import java.nio.ByteBuffer;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;

import org.zeromq.SocketType;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZContext;

/**
 * 
 * Clone server model 1
 * @author Danish Shrestha <dshrestha06@gmail.com>
 *
 */
public class clonesrv1
{
    private static AtomicLong sequence = new AtomicLong();

    public void run()
    {
        try (ZContext ctx = new ZContext()) {
            Socket publisher = ctx.createSocket(SocketType.PUB);
            publisher.bind("tcp://*:5556");

            try {
                Thread.sleep(200);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }

            Random random = new Random();

            while (true) {
                long currentSequenceNumber = sequence.incrementAndGet();
                int key = random.nextInt(10000);
                int body = random.nextInt(1000000);

                ByteBuffer b = ByteBuffer.allocate(4);
                b.asIntBuffer().put(body);

                kvsimple kvMsg = new kvsimple(
                    key + "", currentSequenceNumber, b.array()
                );
                kvMsg.send(publisher);
                System.out.println("sending " + kvMsg);
            }
        }
    }

    public static void main(String[] args)
    {
        new clonesrv1().run();
    }
}

"""
Clone server Model One

"""

import random
import time

import zmq

from kvsimple import KVMsg

def main():
    # Prepare our context and publisher socket
    ctx = zmq.Context()
    publisher = ctx.socket(zmq.PUB)

    publisher.bind("tcp://*:5556")
    time.sleep(0.2)

    sequence = 0
    random.seed(time.time())
    kvmap = {}

    try:
        while True:
            # Distribute as key-value message
            sequence += 1
            kvmsg = KVMsg(sequence)
            kvmsg.key = "%d" % random.randint(1,10000)
            kvmsg.body = "%d" % random.randint(1,1000000)
            kvmsg.send(publisher)
            kvmsg.store(kvmap)
    except KeyboardInterrupt:
        print " Interrupted\n%d messages out" % sequence

if __name__ == '__main__':
    main()

#
#  Clone server Model One
#

lappend auto_path .
package require KVSimple

#  Prepare our context and publisher socket
zmq context context
set pub [zmq socket publisher context PUB]
$pub bind "tcp://*:5556"
after 200

set sequence 0
expr srand([pid])

while {1} {
    #  Distribute as key-value message
    set kvmsg [KVSimple new [incr sequence]]
    $kvmsg set_key [expr {int(rand()*10000)}]
    $kvmsg set_body [expr {int(rand()*1000000)}]
    $kvmsg send $pub
    $kvmsg store kvmap
    puts [$kvmsg dump]
    after 500
}

$pub close
context term

//  Clone client Model One

#include "kvsimple.c"

int main (void)
{
    //  Prepare our context and updates socket
    zctx_t *ctx = zctx_new ();
    void *updates = zsocket_new (ctx, ZMQ_SUB);
    zsocket_set_subscribe (updates, "");
    zsocket_connect (updates, "tcp://localhost:5556");

    zhash_t *kvmap = zhash_new ();
    int64_t sequence = 0;

    while (true) {
        kvmsg_t *kvmsg = kvmsg_recv (updates);
        if (!kvmsg)
            break;          //  Interrupted
        kvmsg_store (&kvmsg, kvmap);
        sequence++;
    }
    printf (" Interrupted\n%d messages in\n", (int) sequence);
    zhash_destroy (&kvmap);
    zctx_destroy (&ctx);
    return 0;
}

#include <iostream>
#include <unordered_map>
#include "kvsimple.hpp"

using namespace std;

int main() {
	// Prepare our context and updates socket
	zmq::context_t ctx(1);
	zmq::socket_t updates(ctx, ZMQ_SUB);
	updates.set(zmq::sockopt::subscribe, ""); // Subscribe to all messages
	updates.connect("tcp://localhost:5555");

	// Initialize key-value map and sequence
	unordered_map<string, string> kvmap;
	int64_t sequence = 0;

	while (true) {
		// Receive key-value message
		auto update_kv_msg = kvmsg::recv(updates);
		if (!update_kv_msg) {
			cout << "Interrupted" << endl;
			return 0;
		}

		// Convert message to string and extract key-value pair
		string key = update_kv_msg->key();
		string value = (char *)update_kv_msg->body().c_str();
		cout << key << " --- " << value << endl;

		// Store key-value pair in map
		kvmap[key] = value;
		sequence++;
	}
	return 0;
}

package guide;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;

/**
 * Clone client model 1
 * @author Danish Shrestha <dshrestha06@gmail.com>
 *
 */
public class clonecli1
{
    private static Map<String, kvsimple> kvMap    = new HashMap<String, kvsimple>();
    private static AtomicLong            sequence = new AtomicLong();

    public void run()
    {
        try (ZContext ctx = new ZContext()) {
            Socket subscriber = ctx.createSocket(SocketType.SUB);
            subscriber.connect("tcp://localhost:5556");
            subscriber.subscribe(ZMQ.SUBSCRIPTION_ALL);

            while (true) {
                kvsimple kvMsg = kvsimple.recv(subscriber);
                if (kvMsg == null)
                    break;

                clonecli1.kvMap.put(kvMsg.getKey(), kvMsg);
                System.out.println("receiving " + kvMsg);
                sequence.incrementAndGet();
            }
        }
    }

    public static void main(String[] args)
    {
        new clonecli1().run();
    }
}

"""
Clone Client Model One

Author: Min RK <benjaminrk@gmail.com>

"""

import random
import time

import zmq

from kvsimple import KVMsg

def main():
    # Prepare our context and publisher socket
    ctx = zmq.Context()
    updates = ctx.socket(zmq.SUB)
    updates.linger = 0
    updates.setsockopt(zmq.SUBSCRIBE, '')
    updates.connect("tcp://localhost:5556")

    kvmap = {}
    sequence = 0

    while True:
        try:
            kvmsg = KVMsg.recv(updates)
        except:
            break # Interrupted
        kvmsg.store(kvmap)
        sequence += 1
    print "Interrupted\n%d messages in" % sequence


if __name__ == '__main__':
    main()

#
# Clone client Model One
#

lappend auto_path .
package require KVSimple

zmq context context
set upd [zmq socket updates context SUB]
$upd setsockopt SUBSCRIBE ""
$upd connect "tcp://localhost:5556"
after 200

while {1} {
    set kvmsg [KVSimple new]
    $kvmsg recv $upd
    $kvmsg store kvmap
    puts [$kvmsg dump]
}

$upd close
context term

//  kvsimple class - key-value message class for example applications

#include "kvsimple.h"
#include "zlist.h"

//  Keys are short strings
#define KVMSG_KEY_MAX   255

//  Message is formatted on wire as 3 frames:
//  frame 0: key (0MQ string)
//  frame 1: sequence (8 bytes, network order)
//  frame 2: body (blob)
#define FRAME_KEY       0
#define FRAME_SEQ       1
#define FRAME_BODY      2
#define KVMSG_FRAMES    3

//  The kvmsg class holds a single key-value message consisting of a
//  list of 0 or more frames:

struct _kvmsg {
    //  Presence indicators for each frame
    int present [KVMSG_FRAMES];
    //  Corresponding 0MQ message frames, if any
    zmq_msg_t frame [KVMSG_FRAMES];
    //  Key, copied into safe C string
    char key [KVMSG_KEY_MAX + 1];
};

//  .split constructor and destructor
//  Here are the constructor and destructor for the class:

//  Constructor, takes a sequence number for the new kvmsg instance:
kvmsg_t *
kvmsg_new (int64_t sequence)
{
    kvmsg_t
        *self;

    self = (kvmsg_t *) zmalloc (sizeof (kvmsg_t));
    kvmsg_set_sequence (self, sequence);
    return self;
}

//  zhash_free_fn callback helper that does the low level destruction:
void
kvmsg_free (void *ptr)
{
    if (ptr) {
        kvmsg_t *self = (kvmsg_t *) ptr;
        //  Destroy message frames if any
        int frame_nbr;
        for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++)
            if (self->present [frame_nbr])
                zmq_msg_close (&self->frame [frame_nbr]);

        //  Free object itself
        free (self);
    }
}

//  Destructor
void
kvmsg_destroy (kvmsg_t **self_p)
{
    assert (self_p);
    if (*self_p) {
        kvmsg_free (*self_p);
        *self_p = NULL;
    }
}

//  .split recv method
//  This method reads a key-value message from socket, and returns a new
//  {{kvmsg}} instance:

kvmsg_t *
kvmsg_recv (void *socket)
{
    assert (socket);
    kvmsg_t *self = kvmsg_new (0);

    //  Read all frames off the wire, reject if bogus
    int frame_nbr;
    for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) {
        if (self->present [frame_nbr])
            zmq_msg_close (&self->frame [frame_nbr]);
        zmq_msg_init (&self->frame [frame_nbr]);
        self->present [frame_nbr] = 1;
        if (zmq_msg_recv (&self->frame [frame_nbr], socket, 0) == -1) {
            kvmsg_destroy (&self);
            break;
        }
        //  Verify multipart framing
        int rcvmore = (frame_nbr < KVMSG_FRAMES - 1)? 1: 0;
        if (zsocket_rcvmore (socket) != rcvmore) {
            kvmsg_destroy (&self);
            break;
        }
    }
    return self;
}

//  .split send method
//  This method sends a multiframe key-value message to a socket:

void
kvmsg_send (kvmsg_t *self, void *socket)
{
    assert (self);
    assert (socket);

    int frame_nbr;
    for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) {
        zmq_msg_t copy;
        zmq_msg_init (&copy);
        if (self->present [frame_nbr])
            zmq_msg_copy (&copy, &self->frame [frame_nbr]);
        zmq_msg_send (&copy, socket, 
            (frame_nbr < KVMSG_FRAMES - 1)? ZMQ_SNDMORE: 0);
        zmq_msg_close (&copy);
    }
}

//  .split key methods
//  These methods let the caller get and set the message key, as a
//  fixed string and as a printf formatted string:

char *
kvmsg_key (kvmsg_t *self)
{
    assert (self);
    if (self->present [FRAME_KEY]) {
        if (!*self->key) {
            size_t size = zmq_msg_size (&self->frame [FRAME_KEY]);
            if (size > KVMSG_KEY_MAX)
                size = KVMSG_KEY_MAX;
            memcpy (self->key,
                zmq_msg_data (&self->frame [FRAME_KEY]), size);
            self->key [size] = 0;
        }
        return self->key;
    }
    else
        return NULL;
}

void
kvmsg_set_key (kvmsg_t *self, char *key)
{
    assert (self);
    zmq_msg_t *msg = &self->frame [FRAME_KEY];
    if (self->present [FRAME_KEY])
        zmq_msg_close (msg);
    zmq_msg_init_size (msg, strlen (key));
    memcpy (zmq_msg_data (msg), key, strlen (key));
    self->present [FRAME_KEY] = 1;
}

void
kvmsg_fmt_key (kvmsg_t *self, char *format, ...)
{
    char value [KVMSG_KEY_MAX + 1];
    va_list args;

    assert (self);
    va_start (args, format);
    vsnprintf (value, KVMSG_KEY_MAX, format, args);
    va_end (args);
    kvmsg_set_key (self, value);
}

//  .split sequence methods
//  These two methods let the caller get and set the message sequence number:

int64_t
kvmsg_sequence (kvmsg_t *self)
{
    assert (self);
    if (self->present [FRAME_SEQ]) {
        assert (zmq_msg_size (&self->frame [FRAME_SEQ]) == 8);
        byte *source = zmq_msg_data (&self->frame [FRAME_SEQ]);
        int64_t sequence = ((int64_t) (source [0]) << 56)
                         + ((int64_t) (source [1]) << 48)
                         + ((int64_t) (source [2]) << 40)
                         + ((int64_t) (source [3]) << 32)
                         + ((int64_t) (source [4]) << 24)
                         + ((int64_t) (source [5]) << 16)
                         + ((int64_t) (source [6]) << 8)
                         +  (int64_t) (source [7]);
        return sequence;
    }
    else
        return 0;
}

void
kvmsg_set_sequence (kvmsg_t *self, int64_t sequence)
{
    assert (self);
    zmq_msg_t *msg = &self->frame [FRAME_SEQ];
    if (self->present [FRAME_SEQ])
        zmq_msg_close (msg);
    zmq_msg_init_size (msg, 8);

    byte *source = zmq_msg_data (msg);
    source [0] = (byte) ((sequence >> 56) & 255);
    source [1] = (byte) ((sequence >> 48) & 255);
    source [2] = (byte) ((sequence >> 40) & 255);
    source [3] = (byte) ((sequence >> 32) & 255);
    source [4] = (byte) ((sequence >> 24) & 255);
    source [5] = (byte) ((sequence >> 16) & 255);
    source [6] = (byte) ((sequence >> 8)  & 255);
    source [7] = (byte) ((sequence)       & 255);

    self->present [FRAME_SEQ] = 1;
}

//  .split message body methods
//  These methods let the caller get and set the message body as a
//  fixed string and as a printf formatted string:

byte *
kvmsg_body (kvmsg_t *self)
{
    assert (self);
    if (self->present [FRAME_BODY])
        return (byte *) zmq_msg_data (&self->frame [FRAME_BODY]);
    else
        return NULL;
}

void
kvmsg_set_body (kvmsg_t *self, byte *body, size_t size)
{
    assert (self);
    zmq_msg_t *msg = &self->frame [FRAME_BODY];
    if (self->present [FRAME_BODY])
        zmq_msg_close (msg);
    self->present [FRAME_BODY] = 1;
    zmq_msg_init_size (msg, size);
    memcpy (zmq_msg_data (msg), body, size);
}

void
kvmsg_fmt_body (kvmsg_t *self, char *format, ...)
{
    char value [255 + 1];
    va_list args;

    assert (self);
    va_start (args, format);
    vsnprintf (value, 255, format, args);
    va_end (args);
    kvmsg_set_body (self, (byte *) value, strlen (value));
}

//  .split size method
//  This method returns the body size of the most recently read message,
//  if any exists:

size_t
kvmsg_size (kvmsg_t *self)
{
    assert (self);
    if (self->present [FRAME_BODY])
        return zmq_msg_size (&self->frame [FRAME_BODY]);
    else
        return 0;
}

//  .split store method
//  This method stores the key-value message into a hash map, unless
//  the key and value are both null. It nullifies the {{kvmsg}} reference
//  so that the object is owned by the hash map, not the caller:

void
kvmsg_store (kvmsg_t **self_p, zhash_t *hash)
{
    assert (self_p);
    if (*self_p) {
        kvmsg_t *self = *self_p;
        assert (self);
        if (self->present [FRAME_KEY]
        &&  self->present [FRAME_BODY]) {
            zhash_update (hash, kvmsg_key (self), self);
            zhash_freefn (hash, kvmsg_key (self), kvmsg_free);
        }
        *self_p = NULL;
    }
}

//  .split dump method
//  This method prints the key-value message to stderr for
//  debugging and tracing:

void
kvmsg_dump (kvmsg_t *self)
{
    if (self) {
        if (!self) {
            fprintf (stderr, "NULL");
            return;
        }
        size_t size = kvmsg_size (self);
        byte  *body = kvmsg_body (self);
        fprintf (stderr, "[seq:%" PRId64 "]", kvmsg_sequence (self));
        fprintf (stderr, "[key:%s]", kvmsg_key (self));
        fprintf (stderr, "[size:%zd] ", size);
        int char_nbr;
        for (char_nbr = 0; char_nbr < size; char_nbr++)
            fprintf (stderr, "%02X", body [char_nbr]);
        fprintf (stderr, "\n");
    }
    else
        fprintf (stderr, "NULL message\n");
}

//  .split test method
//  It's good practice to have a self-test method that tests the class; this
//  also shows how it's used in applications:

int
kvmsg_test (int verbose)
{
    kvmsg_t
        *kvmsg;

    printf (" * kvmsg: ");

    //  Prepare our context and sockets
    zctx_t *ctx = zctx_new ();
    void *output = zsocket_new (ctx, ZMQ_DEALER);
    int rc = zmq_bind (output, "ipc://kvmsg_selftest.ipc");
    assert (rc == 0);
    void *input = zsocket_new (ctx, ZMQ_DEALER);
    rc = zmq_connect (input, "ipc://kvmsg_selftest.ipc");
    assert (rc == 0);

    zhash_t *kvmap = zhash_new ();

    //  Test send and receive of simple message
    kvmsg = kvmsg_new (1);
    kvmsg_set_key  (kvmsg, "key");
    kvmsg_set_body (kvmsg, (byte *) "body", 4);
    if (verbose)
        kvmsg_dump (kvmsg);
    kvmsg_send (kvmsg, output);
    kvmsg_store (&kvmsg, kvmap);

    kvmsg = kvmsg_recv (input);
    if (verbose)
        kvmsg_dump (kvmsg);
    assert (streq (kvmsg_key (kvmsg), "key"));
    kvmsg_store (&kvmsg, kvmap);

    //  Shutdown and destroy all objects
    zhash_destroy (&kvmap);
    zctx_destroy (&ctx);

    printf ("OK\n");
    return 0;
}

/*  =====================================================================
 *  kvsimple - simple key-value message class for example applications
 *  ===================================================================== */

#ifndef __KVSIMPLE_HPP_INCLUDED__
#define __KVSIMPLE_HPP_INCLUDED__

#include "zhelpers.hpp"
#include "zmq.hpp"
#include <cstdint>
#include <iostream>
#include <optional>
#include <sstream>
#include <string>
#include <unordered_map>

using ustring = std::basic_string<unsigned char>;
struct kvmsg {
  kvmsg(std::string key, int64_t sequence, ustring body);
  kvmsg() = default;
  //  Reads key-value message from socket, returns new kvmsg instance.
  static std::optional<kvmsg> recv(zmq::socket_t &socket);
  //  Send key-value message to socket; any empty frames are sent as such.
  void send(zmq::socket_t &socket);

  //  Return key from last read message, if any, else NULL
  std::string key() const;
  //  Return sequence nbr from last read message, if any
  int64_t sequence() const;
  //  Return body from last read message, if any, else NULL
  ustring body() const;
  //  Return body size from last read message, if any, else zero
  size_t size() const;

  //  Set message key as provided
  void set_key(std::string key);
  //  Set message sequence number
  void set_sequence(int64_t sequence);
  //  Set message body
  void set_body(ustring body);

  //  Dump message to stderr, for debugging and tracing
  std::string to_string();

  //  Runs self test of class
  static bool test(int verbose);

private:
  static constexpr uint32_t kvmsg_key_max = 255;
  static constexpr uint32_t frame_key = 0;
  static constexpr uint32_t frame_seq = 1;
  static constexpr uint32_t frame_body = 2;
  static constexpr uint32_t kvmsg_frames = 3;

  std::string key_;
  ustring body_;
  int64_t sequence_{};
};

namespace {
std::optional<zmq::message_t> receive_message(zmq::socket_t &socket) {
  zmq::message_t message(0);
  message.rebuild(0);
  try {
    if (!socket.recv(message, zmq::recv_flags::none)) {
      return {};
    }
  } catch (zmq::error_t &error) {
    std::cerr << "E: " << error.what() << std::endl;
    return {};
  }
  return message;
}
} // namespace

kvmsg::kvmsg(std::string key, int64_t sequence, ustring body)
    : key_(key), body_(body), sequence_(sequence) {}

//  Reads key-value message from socket, returns new kvmsg instance.
std::optional<kvmsg> kvmsg::recv(zmq::socket_t &socket) {
  auto key_message = receive_message(socket);
  if (!key_message)
    return {};
  kvmsg msg;
  msg.set_key(
      std::string((char *)(*key_message).data(), (*key_message).size()));
  auto sequence_message = receive_message(socket);
  msg.set_sequence(*(int64_t *)(*sequence_message).data());
  if (!sequence_message)
    return {};
  auto body_message = receive_message(socket);
  if (!body_message)
    return {};
  msg.set_body(
      ustring((unsigned char *)(*body_message).data(), (*body_message).size()));
  return msg;
}

//  Send key-value message to socket; any empty frames are sent as such.
void kvmsg::send(zmq::socket_t &socket) {
  {
    zmq::message_t message;
    message.rebuild(key_.size());
    std::memcpy(message.data(), key_.c_str(), key_.size());
    socket.send(message, zmq::send_flags::sndmore);
  }
  {
    zmq::message_t message;
    message.rebuild(sizeof(sequence_));
    std::memcpy(message.data(), (void *)&sequence_, sizeof(sequence_));
    socket.send(message, zmq::send_flags::sndmore);
  }
  {
    zmq::message_t message;
    message.rebuild(body_.size());
    std::memcpy(message.data(), body_.c_str(), body_.size());
    socket.send(message, zmq::send_flags::none);
  }
}

//  Return key from last read message, if any, else NULL
std::string kvmsg::key() const { return key_; }
//  Return sequence nbr from last read message, if any
int64_t kvmsg::sequence() const { return sequence_; }
//  Return body from last read message, if any, else NULL
ustring kvmsg::body() const { return body_; }
//  Return body size from last read message, if any, else zero
size_t kvmsg::size() const { return body_.size(); }
//  Set message key as provided
void kvmsg::set_key(std::string key) { key_ = key; }
//  Set message sequence number
void kvmsg::set_sequence(int64_t sequence) { sequence_ = sequence; }
//  Set message body
void kvmsg::set_body(ustring body) { body_ = body; }

std::string kvmsg::to_string() {
  std::stringstream ss;
  ss << "key=" << key_ << ",sequence=" << sequence_ << ",body=";
  s_dump_message(ss, body_);
  return ss.str();
}
//  Dump message to stderr, for debugging and tracing

//  Runs self test of class
bool kvmsg::test(int verbose) {

  zmq::context_t context;
  zmq::socket_t output(context, ZMQ_DEALER);
  output.bind("ipc://kvmsg_selftest.ipc");
  zmq::socket_t input(context, ZMQ_DEALER);
  input.connect("ipc://kvmsg_selftest.ipc");

  kvmsg message("key", 1, (unsigned char *)"body");
  if (verbose) {
    std::cout << message.to_string()<<std::endl;
  }
  message.send(output);
  std::unordered_map<std::string, kvmsg> kvmap;
  kvmap["key"] = message;
  auto input_message_opt = kvmsg::recv(input);
  if (!input_message_opt)
    return false;
  assert((*input_message_opt).key() == "key");
  assert((*input_message_opt).sequence() == 1);
  assert((*input_message_opt).body() == (unsigned char *)"body");
  if (verbose) {
    std::cout << (*input_message_opt).to_string()<<std::endl;
  }

  return true;
}

// Main routine for running the basic test

//int main() {
// std::cout << (kvmsg::test(1) ? "SUCCESS" : "FAILURE") << std::endl;
// return 0;
//}

#endif //  Included

package guide;

import java.nio.ByteBuffer;
import java.util.Arrays;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;

/**
 * 
 * A simple getKey value message class
 * @author Danish Shrestha <dshrestha06@gmail.com>
 *
 */
public class kvsimple
{
    private final String key;
    private long         sequence;
    private final byte[] body;

    public kvsimple(String key, long sequence, byte[] body)
    {
        this.key = key;
        this.sequence = sequence;
        this.body = body; // clone if needed
    }

    public String getKey()
    {
        return key;
    }

    public long getSequence()
    {
        return sequence;
    }

    public void setSequence(long sequence)
    {
        this.sequence = sequence;
    }

    public byte[] getBody()
    {
        return body;
    }

    public void send(Socket publisher)
    {

        publisher.send(key.getBytes(ZMQ.CHARSET), ZMQ.SNDMORE);

        ByteBuffer bb = ByteBuffer.allocate(8);
        bb.asLongBuffer().put(sequence);
        publisher.send(bb.array(), ZMQ.SNDMORE);

        publisher.send(body, 0);
    }

    public static kvsimple recv(Socket updates)
    {
        byte[] data = updates.recv(0);
        if (data == null || !updates.hasReceiveMore())
            return null;
        String key = new String(data, ZMQ.CHARSET);
        data = updates.recv(0);
        if (data == null || !updates.hasReceiveMore())
            return null;
        Long sequence = ByteBuffer.wrap(data).getLong();
        byte[] body = updates.recv(0);
        if (body == null || updates.hasReceiveMore())
            return null;

        return new kvsimple(key, sequence, body);
    }

    @Override
    public String toString()
    {
        return "kvsimple [getKey=" + key + ", getSequence=" + sequence + ", body=" + Arrays.toString(body) + "]";
    }

    @Override
    public int hashCode()
    {
        final int prime = 31;
        int result = 1;
        result = prime * result + Arrays.hashCode(body);
        result = prime * result + ((key == null) ? 0 : key.hashCode());
        result = prime * result + (int) (sequence ^ (sequence >>> 32));
        return result;
    }

    @Override
    public boolean equals(Object obj)
    {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        kvsimple other = (kvsimple) obj;
        if (!Arrays.equals(body, other.body))
            return false;
        if (key == null) {
            if (other.key != null)
                return false;
        }
        else if (!key.equals(other.key))
            return false;
        if (sequence != other.sequence)
            return false;
        return true;
    }

}

"""
=====================================================================
kvsimple - simple key-value message class for example applications

Author: Min RK <benjaminrk@gmail.com>

"""

import struct # for packing integers
import sys

import zmq

class KVMsg(object):
    """
    Message is formatted on wire as 3 frames:
    frame 0: key (0MQ string)
    frame 1: sequence (8 bytes, network order)
    frame 2: body (blob)
    """
    key = None # key (string)
    sequence = 0 # int
    body = None # blob

    def __init__(self, sequence, key=None, body=None):
        assert isinstance(sequence, int)
        self.sequence = sequence
        self.key = key
        self.body = body

    def store(self, dikt):
        """Store me in a dict if I have anything to store"""
        # this seems weird to check, but it's what the C example does
        if self.key is not None and self.body is not None:
            dikt[self.key] = self

    def send(self, socket):
        """Send key-value message to socket; any empty frames are sent as such."""
        key = '' if self.key is None else self.key
        seq_s = struct.pack('!l', self.sequence)
        body = '' if self.body is None else self.body
        socket.send_multipart([ key, seq_s, body ])

    @classmethod
    def recv(cls, socket):
        """Reads key-value message from socket, returns new kvmsg instance."""
        key, seq_s, body = socket.recv_multipart()
        key = key if key else None
        seq = struct.unpack('!l',seq_s)[0]
        body = body if body else None
        return cls(seq, key=key, body=body)

    def dump(self):
        if self.body is None:
            size = 0
            data='NULL'
        else:
            size = len(self.body)
            data=repr(self.body)
        print >> sys.stderr, "[seq:{seq}][key:{key}][size:{size}] {data}".format(
            seq=self.sequence,
            key=self.key,
            size=size,
            data=data,
        )

# ---------------------------------------------------------------------
# Runs self test of class

def test_kvmsg (verbose):
    print " * kvmsg: ",

    # Prepare our context and sockets
    ctx = zmq.Context()
    output = ctx.socket(zmq.DEALER)
    output.bind("ipc://kvmsg_selftest.ipc")
    input = ctx.socket(zmq.DEALER)
    input.connect("ipc://kvmsg_selftest.ipc")

    kvmap = {}
    # Test send and receive of simple message
    kvmsg = KVMsg(1)
    kvmsg.key = "key"
    kvmsg.body = "body"
    if verbose:
        kvmsg.dump()
    kvmsg.send(output)
    kvmsg.store(kvmap)

    kvmsg2 = KVMsg.recv(input)
    if verbose:
        kvmsg2.dump()
    assert kvmsg2.key == "key"
    kvmsg2.store(kvmap)

    assert len(kvmap) == 1 # shouldn't be different

    print "OK"

if __name__ == '__main__':
    test_kvmsg('-v' in sys.argv)

# =====================================================================
# kvsimple - simple key-value message class for example applications
# =====================================================================

lappend auto_path .

package require TclOO
package require zmq
package require mdp

package provide KVSimple 1.0

#  Keys are short strings
set KVMSG_KEY_MAX   255

#  Message is formatted on wire as 3 frames:
#  frame 0: key (0MQ string)
#  frame 1: sequence (8 bytes, network order)
#  frame 2: body (blob)
set FRAME_KEY       0
set FRAME_SEQ       1
set FRAME_BODY      2
set KVMSG_FRAMES    3

oo::class create KVSimple {

    variable frame key

    #  Constructor, sets sequence as provided
    constructor {{isequence 0}} {
	set frame [list]
	my set_sequence $isequence
    }

    destructor {
    }

    #  Reads key-value message from socket
    method recv {socket} {
	set frame [list]
	#  Read all frames off the wire
	for {set frame_nbr 0} {$frame_nbr < $::KVMSG_FRAMES} {incr frame_nbr} {
	    lappend frame [$socket recv]
	    #  Verify multipart framing
	    if {![$socket getsockopt RCVMORE]} {
		break
	    }
	}
    }

    #  Send key-value message to socket; any empty frames are sent as such.
    method send {socket} {
	for {set frame_nbr 0} {$frame_nbr < $::KVMSG_FRAMES} {incr frame_nbr} {
	    if {$frame_nbr == ($::KVMSG_FRAMES - 1)} {
		$socket send [lindex $frame $frame_nbr]
	    } else {
		$socket sendmore [lindex $frame $frame_nbr]
	    }
	}
    }

    #  Return key from last read message, if any, else NULL
    method key {} {
	if {[llength $frame] > $::FRAME_KEY} {
	    if {![info exists key]} {
		set size [string length [lindex $frame $::FRAME_KEY]]
		if {$size > $::KVMSG_KEY_MAX} {
		    set size $::KVMSG_KEY_MAX
		}
		set key [string range [lindex $frame $::FRAME_KEY] 0 [expr {$size - 1}]]
	    }
	    return $key
	} else {
	    return {}
	}
    }

    #  Return sequence nbr from last read message, if any
    method sequence {} {
	if {[llength $frame] > $::FRAME_SEQ} {
	    set s [lindex $frame $::FRAME_SEQ]
	    if {[string length $s] != 8} {
		error "sequence frame must have length 8"
	    }
	    binary scan [lindex $frame $::FRAME_SEQ] W r
	    return $r
	} else {
	    return 0
	}
    }

    #  Return body from last read message, if any, else NULL
    method body {} {
	if {[llength $frame] > $::FRAME_BODY} {
	    return [lindex $frame $::FRAME_BODY]
	} else {
	    return {}
	}
    }

    #  Return body size from last read message, if any, else zero
    method size {} {
	if {[llength $frame] > $::FRAME_BODY} {
	    return [string length [lindex $frame $::FRAME_BODY]]
	} else {
	    return {}
	}
    }

    #  Set message key as provided
    method set_key {ikey} {
	while {[llength $frame] <= $::FRAME_KEY} {
	    lappend frame {}
	}
	lset frame $::FRAME_KEY $ikey
    }

    #  Set message sequence number
    method set_sequence {isequence} {
	while {[llength $frame] <= $::FRAME_SEQ} {
	    lappend frame {}
	}
	set sequence [binary format W $isequence]
	lset frame $::FRAME_SEQ $sequence
    }

    #  Set message body
    method set_body {ibody} {
	while {[llength $frame] <= $::FRAME_KEY} {
	    lappend frame {}
	}
	lset frame $::FRAME_BODY $ibody
    }

    #  Set message key using printf format
    method fmt_key {format args} {
	my set_key [format $format {*}$args]
    }

    #  Set message body using printf format
    method fmt_body {format args} {
	my set_body [format $format {*}$args]
    }

    #  Store entire kvmsg into hash map, if key/value are set
    #  Nullifies kvmsg reference, and destroys automatically when no longer
    #  needed.
    method store {hashnm} {
	upvar $hashnm hash
	if {[info exists hash([my key])]} {
	    $hash([my key]) destroy
	}
	set hash([my key]) [self]
    }

    #  Dump message to stderr, for debugging and tracing
    method dump {} {
	set rt ""
	append rt [format {[seq:%lld]} [my sequence]]
	append rt [format {[key:%s]} [my key]]
	append rt [format {[size:%d] } [my size]]
	set size [my size]
	set body [my body]
	for {set i 0} {$i < $size} {incr i} {
	    set c [lindex $body $i]
	    if {[string is ascii $c]} {
		append rt $c
	    } else {
		append rt [binary scan H2 $c]
	    }
	}
	return $rt
    }
}

namespace eval ::KVSimpleTest {
    proc test {verbose} {

	puts -nonewline " * kvmsg: "

	#  Prepare our context and sockets
	zmq context context
	set os [zmq socket output context DEALER]
	output bind "ipc://kvmsg_selftest.ipc"
	set is [zmq socket input context DEALER]
	input connect "ipc://kvmsg_selftest.ipc"

	#  Test send and receive of simple message
	set kvmsg [KVSimple new 1]
	$kvmsg set_key "key"
	$kvmsg set_body "body"
	if {$verbose} {
	    puts [$kvmsg dump]
	}

	$kvmsg send $os
	$kvmsg store kvmap

	$kvmsg recv $is
	if {$verbose} {
	    puts [$kvmsg dump]
	}
	if {[$kvmsg key] ne "key"} {
	    error "Unexpected key: [$kvmsg key]"
	}
	$kvmsg store kvmap

	#  Shutdown and destroy all objects
	input close
	output close
	context term

	puts "OK"
    }
}


#::KVSimpleTest::test 1

//  Clone server - Model Two

//  Lets us build this source without creating a library
#include "kvsimple.c"

static int s_send_single (const char *key, void *data, void *args);
static void state_manager (void *args, zctx_t *ctx, void *pipe);

int main (void)
{
    //  Prepare our context and sockets
    zctx_t *ctx = zctx_new ();
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (publisher, "tcp://*:5557");

    int64_t sequence = 0;
    srandom ((unsigned) time (NULL));

    //  Start state manager and wait for synchronization signal
    void *updates = zthread_fork (ctx, state_manager, NULL);
    free (zstr_recv (updates));

    while (!zctx_interrupted) {
        //  Distribute as key-value message
        kvmsg_t *kvmsg = kvmsg_new (++sequence);
        kvmsg_fmt_key  (kvmsg, "%d", randof (10000));
        kvmsg_fmt_body (kvmsg, "%d", randof (1000000));
        kvmsg_send     (kvmsg, publisher);
        kvmsg_send     (kvmsg, updates);
        kvmsg_destroy (&kvmsg);
    }
    printf (" Interrupted\n%d messages out\n", (int) sequence);
    zctx_destroy (&ctx);
    return 0;
}

//  Routing information for a key-value snapshot
typedef struct {
    void *socket;           //  ROUTER socket to send to
    zframe_t *identity;     //  Identity of peer who requested state
} kvroute_t;

//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int
s_send_single (const char *key, void *data, void *args)
{
    kvroute_t *kvroute = (kvroute_t *) args;
    //  Send identity of recipient first
    zframe_send (&kvroute->identity,
        kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);
    kvmsg_t *kvmsg = (kvmsg_t *) data;
    kvmsg_send (kvmsg, kvroute->socket);
    return 0;
}

//  .split state manager
//  The state manager task maintains the state and handles requests from
//  clients for snapshots:

static void
state_manager (void *args, zctx_t *ctx, void *pipe)
{
    zhash_t *kvmap = zhash_new ();

    zstr_send (pipe, "READY");
    void *snapshot = zsocket_new (ctx, ZMQ_ROUTER);
    zsocket_bind (snapshot, "tcp://*:5556");

    zmq_pollitem_t items [] = {
        { pipe, 0, ZMQ_POLLIN, 0 },
        { snapshot, 0, ZMQ_POLLIN, 0 }
    };
    int64_t sequence = 0;       //  Current snapshot version number
    while (!zctx_interrupted) {
        int rc = zmq_poll (items, 2, -1);
        if (rc == -1 && errno == ETERM)
            break;              //  Context has been shut down

        //  Apply state update from main thread
        if (items [0].revents & ZMQ_POLLIN) {
            kvmsg_t *kvmsg = kvmsg_recv (pipe);
            if (!kvmsg)
                break;          //  Interrupted
            sequence = kvmsg_sequence (kvmsg);
            kvmsg_store (&kvmsg, kvmap);
        }
        //  Execute state snapshot request
        if (items [1].revents & ZMQ_POLLIN) {
            zframe_t *identity = zframe_recv (snapshot);
            if (!identity)
                break;          //  Interrupted

            //  Request is in second frame of message
            char *request = zstr_recv (snapshot);
            if (streq (request, "ICANHAZ?"))
                free (request);
            else {
                printf ("E: bad request, aborting\n");
                break;
            }
            //  Send state snapshot to client
            kvroute_t routing = { snapshot, identity };

            //  For each entry in kvmap, send kvmsg to client
            zhash_foreach (kvmap, s_send_single, &routing);

            //  Now send END message with sequence number
            printf ("Sending state shapshot=%d\n", (int) sequence);
            zframe_send (&identity, snapshot, ZFRAME_MORE);
            kvmsg_t *kvmsg = kvmsg_new (sequence);
            kvmsg_set_key  (kvmsg, "KTHXBAI");
            kvmsg_set_body (kvmsg, (byte *) "", 0);
            kvmsg_send     (kvmsg, snapshot);
            kvmsg_destroy (&kvmsg);
        }
    }
    zhash_destroy (&kvmap);
}

#include "kvsimple.hpp"
#include <thread>

static int s_send_snapshot(std::unordered_map<std::string, kvmsg>& kvmap, zmq::socket_t* snapshot);
static void state_manager(zmq::context_t* ctx);

// simulate zthread_fork, create attached thread and return the pipe socket
std::pair<std::thread, zmq::socket_t> zthread_fork(zmq::context_t& ctx, void (*thread_func)(zmq::context_t*)) {
    // create the pipe socket for the main thread to communicate with its child thread
    zmq::socket_t pipe(ctx, ZMQ_PAIR);
    pipe.connect("inproc://state_manager");

    // start child thread
    std::thread t(thread_func, &ctx);

    return std::make_pair(std::move(t), std::move(pipe));
}

int main(void) {
    // Prepare our context and socket
    zmq::context_t ctx(1);
    zmq::socket_t publisher(ctx, ZMQ_PUB);
    publisher.bind("tcp://*:5557");

    int64_t sequence = 0;

    //  Start state manager and wait for synchronization signal
    auto [state_manager_thread, state_manager_pipe] = zthread_fork(ctx, state_manager);
    zmq::message_t sync_msg;
    state_manager_pipe.recv(sync_msg);

    s_catch_signals();
    while(!s_interrupted) {
        kvmsg msg = kvmsg("key", ++sequence, (unsigned char *)"value");
        msg.set_key(std::to_string(within(10000)));
        msg.set_body((unsigned char *)std::to_string(within(1000000)).c_str());
        msg.send(publisher);
        msg.send(state_manager_pipe);
        s_sleep(500);
    }
    std::cout << " Interrupted\n" <<  sequence << " messages out\n" << std::endl;
    kvmsg msg("END", sequence, (unsigned char *)"");
    msg.send(state_manager_pipe); 
    state_manager_thread.join();
    return 0;
}

//  Routing information for a key-value snapshot
typedef struct {
    zmq::socket_t *socket; //  ROUTER socket to send to
    std::string identity; //  Identity of peer who requested state
} kvroute_t;

//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int s_send_snapshot(std::unordered_map<std::string, kvmsg>& kvmap, kvroute_t& kvroute) {
    for (auto& kv : kvmap) {
        s_sendmore(*kvroute.socket, kvroute.identity);
        kv.second.send(*kvroute.socket);
    }
    return 0;
}

//  .split state manager
//  The state manager task maintains the state and handles requests from
//  clients for snapshots:
static void state_manager(zmq::context_t *ctx) {
    std::unordered_map<std::string, kvmsg> kvmap;
    zmq::socket_t pipe(*ctx, ZMQ_PAIR);
    pipe.bind("inproc://state_manager");
    s_send(pipe, std::string("READY"));

    zmq::socket_t snapshot(*ctx, ZMQ_ROUTER);
    snapshot.bind("tcp://*:5556");

    zmq::pollitem_t items[] = {
        {pipe, 0, ZMQ_POLLIN, 0},
        {snapshot, 0, ZMQ_POLLIN, 0}
    };
    int64_t sequence = 0;
    while(true) {
        zmq::poll(&items[0], 2, -1);
        if (items[0].revents & ZMQ_POLLIN) {
            auto msg = kvmsg::recv(pipe);
            if (!msg || msg->key() == "END") {
                break;
            }
            sequence = msg->sequence();
            kvmap[msg->key()] = *msg;
        }
        //  Execute state snapshot request
        if (items[1].revents & ZMQ_POLLIN) {
            std::string identity = s_recv(snapshot);
            std::string request = s_recv(snapshot);
            if (request != "ICANHAZ?") {
                std::cerr << "E: bad request, aborting\n";
                break;
            }
            //  Send state snapshot to client
            kvroute_t kvroute = {&snapshot, identity};

            //  For each entry in kvmap, send kvmsg to client
            s_send_snapshot(kvmap, kvroute);

            //  Now send END message with sequence number
            std::cout << "sending state snapshot=" << sequence << std::endl;
            s_sendmore(snapshot, identity);
            kvmsg msg("KTHXBAI", sequence, (unsigned char *)"");
            msg.send(snapshot);
        }
    }
}

package guide;

import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZThread;
import org.zeromq.ZThread.IAttachedRunnable;

/**
 * Clone server Model Two
 *
 * @author Danish Shrestha <dshrestha06@gmail.com>
 *
 */
public class clonesrv2
{

    public void run()
    {
        try (ZContext ctx = new ZContext()) {
            Socket publisher = ctx.createSocket(SocketType.PUB);
            publisher.bind("tcp://*:5557");

            Socket updates = ZThread.fork(ctx, new StateManager());

            Random random = new Random();
            long sequence = 0;
            while (!Thread.currentThread().isInterrupted()) {
                long currentSequenceNumber = ++sequence;
                int key = random.nextInt(10000);
                int body = random.nextInt(1000000);

                ByteBuffer b = ByteBuffer.allocate(4);
                b.asIntBuffer().put(body);

                kvsimple kvMsg = new kvsimple(
                    key + "", currentSequenceNumber, b.array()
                );
                kvMsg.send(publisher);
                kvMsg.send(updates); // send a message to State Manager thread.

                try {
                    Thread.sleep(1000);
                }
                catch (InterruptedException e) {
                }
            }

            System.out.printf(" Interrupted\n%d messages out\n", sequence);
        }
    }

    public static class StateManager implements IAttachedRunnable
    {
        private static Map<String, kvsimple> kvMap = new LinkedHashMap<String, kvsimple>();

        @Override
        public void run(Object[] args, ZContext ctx, Socket pipe)
        {
            pipe.send("READY"); // optional

            Socket snapshot = ctx.createSocket(SocketType.ROUTER);
            snapshot.bind("tcp://*:5556");

            Poller poller = ctx.createPoller(2);
            poller.register(pipe, ZMQ.Poller.POLLIN);
            poller.register(snapshot, ZMQ.Poller.POLLIN);

            long stateSequence = 0;
            while (!Thread.currentThread().isInterrupted()) {
                if (poller.poll() < 0)
                    break; //  Context has been shut down

                // apply state updates from main thread
                if (poller.pollin(0)) {
                    kvsimple kvMsg = kvsimple.recv(pipe);
                    if (kvMsg == null)
                        break;
                    StateManager.kvMap.put(kvMsg.getKey(), kvMsg);
                    stateSequence = kvMsg.getSequence();
                }

                // execute state snapshot request
                if (poller.pollin(1)) {
                    byte[] identity = snapshot.recv(0);
                    if (identity == null)
                        break;
                    String request = new String(snapshot.recv(0), ZMQ.CHARSET);

                    if (!request.equals("ICANHAZ?")) {
                        System.out.println("E: bad request, aborting");
                        break;
                    }

                    Iterator<Entry<String, kvsimple>> iter = kvMap.entrySet().iterator();
                    while (iter.hasNext()) {
                        Entry<String, kvsimple> entry = iter.next();
                        kvsimple msg = entry.getValue();
                        System.out.println("Sending message " + entry.getValue().getSequence());
                        this.sendMessage(msg, identity, snapshot);
                    }

                    // now send end message with getSequence number
                    System.out.println("Sending state snapshot = " + stateSequence);
                    snapshot.send(identity, ZMQ.SNDMORE);
                    kvsimple message = new kvsimple("KTHXBAI", stateSequence, ZMQ.MESSAGE_SEPARATOR);
                    message.send(snapshot);
                }
            }
        }

        private void sendMessage(kvsimple msg, byte[] identity, Socket snapshot)
        {
            snapshot.send(identity, ZMQ.SNDMORE);
            msg.send(snapshot);
        }
    }

    public static void main(String[] args)
    {
        new clonesrv2().run();
    }
}

"""
Clone server Model Two

Author: Min RK <benjaminrk@gmail.com>
"""
import random
import threading
import time

import zmq

from kvsimple import KVMsg
from zhelpers import zpipe

def main():
    # Prepare our context and publisher socket
    ctx = zmq.Context()
    publisher = ctx.socket(zmq.PUB)
    publisher.bind("tcp://*:5557")

    updates, peer = zpipe(ctx)

    manager_thread = threading.Thread(target=state_manager, args=(ctx,peer))
    manager_thread.daemon=True
    manager_thread.start()


    sequence = 0
    random.seed(time.time())

    try:
        while True:
            # Distribute as key-value message
            sequence += 1
            kvmsg = KVMsg(sequence)
            kvmsg.key = "%d" % random.randint(1,10000)
            kvmsg.body = "%d" % random.randint(1,1000000)
            kvmsg.send(publisher)
            kvmsg.send(updates)
    except KeyboardInterrupt:
        print " Interrupted\n%d messages out" % sequence

# simple struct for routing information for a key-value snapshot
class Route:
    def __init__(self, socket, identity):
        self.socket = socket # ROUTER socket to send to
        self.identity = identity # Identity of peer who requested state

def send_single(key, kvmsg, route):
    """Send one state snapshot key-value pair to a socket

    Hash item data is our kvmsg object, ready to send
    """
    # Send identity of recipient first
    route.socket.send(route.identity, zmq.SNDMORE)
    kvmsg.send(route.socket)


def state_manager(ctx, pipe):
    """This thread maintains the state and handles requests from clients for snapshots.
    """
    kvmap = {}
    pipe.send("READY")
    snapshot = ctx.socket(zmq.ROUTER)
    snapshot.bind("tcp://*:5556")

    poller = zmq.Poller()
    poller.register(pipe, zmq.POLLIN)
    poller.register(snapshot, zmq.POLLIN)

    sequence = 0       # Current snapshot version number
    while True:
        try:
            items = dict(poller.poll())
        except (zmq.ZMQError, KeyboardInterrupt):
            break # interrupt/context shutdown

        # Apply state update from main thread
        if pipe in items:
            kvmsg = KVMsg.recv(pipe)
            sequence = kvmsg.sequence
            kvmsg.store(kvmap)
        # Execute state snapshot request
        if snapshot in items:
            msg = snapshot.recv_multipart()
            identity = msg[0]
            request = msg[1]
            if request == "ICANHAZ?":
                pass
            else:
                print "E: bad request, aborting\n",
                break

            # Send state snapshot to client
            route = Route(snapshot, identity)

            # For each entry in kvmap, send kvmsg to client
            for k,v in kvmap.items():
                send_single(k,v,route)

            # Now send END message with sequence number
            print "Sending state shapshot=%d\n" % sequence,
            snapshot.send(identity, zmq.SNDMORE)
            kvmsg = KVMsg(sequence)
            kvmsg.key = "KTHXBAI"
            kvmsg.body = ""
            kvmsg.send(snapshot)

if __name__ == '__main__':
    main()

#
#  Clone server Model Two
#

lappend auto_path .
package require KVSimple

if {[llength $argv] == 0} {
    set argv "pub"
} elseif {[llength $argv] != 1} {
    puts "Usage: clonesrv2.tcl <pub|upd>"
    exit 1
}

lassign $argv what

set tclsh [info nameofexecutable]
expr srand([pid])

switch -exact -- $what {
    pub {
	#  Prepare our context and publisher socket
	zmq context context
	set pub [zmq socket publisher context PUB]
	$pub bind "tcp://*:5557"
	set upd [zmq socket updates context PAIR]
	$upd bind "ipc://updates.ipc"

	set sequence 0

	#  Start state manager and wait for synchronization signal
	exec $tclsh clonesrv2.tcl upd > upd.log 2>@1 &
	$upd recv

	while {1} {
	    #  Distribute as key-value message
	    set kvmsg [KVSimple new [incr sequence]]
	    $kvmsg set_key [expr {int(rand()*10000)}]
	    $kvmsg set_body [expr {int(rand()*1000000)}]
	    $kvmsg send $pub
	    $kvmsg send $upd
	    puts [$kvmsg dump]
	    after 500
	}

	$pub close
	$upd close
	context term
    }
    upd {
	zmq context context
	set upd [zmq socket updates context PAIR]
	$upd connect "ipc://updates.ipc"
	$upd send "READY"

	set snp [zmq socket snapshot context ROUTER]
	$snp bind "tcp://*:5556"

	set sequence 0 ;# Current snapshot version number

        #  Apply state update from main thread
	proc apply_state_update {upd} {
	    global kvmap sequence
	    set kvmsg [KVSimple new]
	    $kvmsg recv $upd
	    set sequence [$kvmsg sequence]
	    $kvmsg store kvmap
	}

        #  Execute state snapshot request
	proc execute_state_snapshot_request {snp} {
	    global kvmap sequence

	    set identity [$snp recv]

	    #  Request is in second frame of message
	    set request [$snp recv]
	    if {$request ne "ICANHAZ?"} {
		puts "E: bad request, aborting"
		exit 1
	    }
	    #  Send state snapshot to client
            #  For each entry in kvmap, send kvmsg to client
	    foreach {key value} [array get kvmap] {
		#  Send one state snapshot key-value pair to a socket
		#  Hash item data is our kvmsg object, ready to send
		$snp sendmore $identity
		$value send $snp
	    }

            #  Now send END message with sequence number
	    puts "Sending state snapshot=$sequence"
	    $snp sendmore $identity
	    set kvmsg [KVSimple new $sequence]
	    $kvmsg set_key "KTHXBAI"
	    $kvmsg set_body ""
	    $kvmsg send $snp
	    $kvmsg destroy
	}

	$upd readable [list apply_state_update $upd]
	$snp readable [list execute_state_snapshot_request $snp]
	vwait forever

	$upd close
	$snp close
	context term
    }
}

//  Clone client - Model Two

//  Lets us build this source without creating a library
#include "kvsimple.c"

int main (void)
{
    //  Prepare our context and subscriber
    zctx_t *ctx = zctx_new ();
    void *snapshot = zsocket_new (ctx, ZMQ_DEALER);
    zsocket_connect (snapshot, "tcp://localhost:5556");
    void *subscriber = zsocket_new (ctx, ZMQ_SUB);
    zsocket_set_subscribe (subscriber, "");
    zsocket_connect (subscriber, "tcp://localhost:5557");

    zhash_t *kvmap = zhash_new ();

    //  Get state snapshot
    int64_t sequence = 0;
    zstr_send (snapshot, "ICANHAZ?");
    while (true) {
        kvmsg_t *kvmsg = kvmsg_recv (snapshot);
        if (!kvmsg)
            break;          //  Interrupted
        if (streq (kvmsg_key (kvmsg), "KTHXBAI")) {
            sequence = kvmsg_sequence (kvmsg);
            printf ("Received snapshot=%d\n", (int) sequence);
            kvmsg_destroy (&kvmsg);
            break;          //  Done
        }
        kvmsg_store (&kvmsg, kvmap);
    }
    //  Now apply pending updates, discard out-of-sequence messages
    while (!zctx_interrupted) {
        kvmsg_t *kvmsg = kvmsg_recv (subscriber);
        if (!kvmsg)
            break;          //  Interrupted
        if (kvmsg_sequence (kvmsg) > sequence) {
            sequence = kvmsg_sequence (kvmsg);
            kvmsg_store (&kvmsg, kvmap);
        }
        else
            kvmsg_destroy (&kvmsg);
    }
    zhash_destroy (&kvmap);
    zctx_destroy (&ctx);
    return 0;
}

#include "kvsimple.hpp"

int main(void) {
    zmq::context_t ctx(1);
    zmq::socket_t snapshot(ctx, ZMQ_DEALER);
    snapshot.connect("tcp://localhost:5556");
    zmq::socket_t subscriber(ctx, ZMQ_SUB);
    subscriber.set(zmq::sockopt::subscribe, "");
    subscriber.connect("tcp://localhost:5557");

    std::unordered_map<std::string, kvmsg> kvmap;

    //  Get state snapshot
    int64_t sequence = 0;
    s_send(snapshot, std::string("ICANHAZ?"));
    while (true) {
        auto kv = kvmsg::recv(snapshot);
        if (!kv) break;
        if (kv->key() == "KTHXBAI") {
            sequence = kv->sequence();
            std::cout << "Received snapshot=" << sequence << std::endl;
            break;
        }
        kvmap[kv->key()] = *kv;
    }
    //  Now apply pending updates, discard out-of-sequence messages
    while(true) {
        auto kv = kvmsg::recv(subscriber);
        if (!kv) break;
        if (kv->sequence() > sequence) {
            sequence = kv->sequence();
            kvmap[kv->key()] = *kv;
            std::cout << "Received update=" << sequence << std::endl;
        }
    }
    return 0;
}

package guide;

import java.util.HashMap;
import java.util.Map;

import org.zeromq.SocketType;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZContext;

/**
 * Clone client Model Two
 * 
 * @author Danish Shrestha <dshrestha06@gmail.com>
 * 
 */
public class clonecli2
{
    private static Map<String, kvsimple> kvMap = new HashMap<String, kvsimple>();

    public void run()
    {
        try (ZContext ctx = new ZContext()) {
            Socket snapshot = ctx.createSocket(SocketType.DEALER);
            snapshot.connect("tcp://localhost:5556");

            Socket subscriber = ctx.createSocket(SocketType.SUB);
            subscriber.connect("tcp://localhost:5557");
            subscriber.subscribe(ZMQ.SUBSCRIPTION_ALL);

            // get state snapshot
            snapshot.send("ICANHAZ?".getBytes(ZMQ.CHARSET), 0);
            long sequence = 0;
            while (true) {
                kvsimple kvMsg = kvsimple.recv(snapshot);
                if (kvMsg == null)
                    break;
                sequence = kvMsg.getSequence();
                if ("KTHXBAI".equalsIgnoreCase(kvMsg.getKey())) {
                    System.out.println("Received snapshot = " + kvMsg.getSequence());
                    break; // done
                }

                System.out.println("receiving " + kvMsg.getSequence());
                clonecli2.kvMap.put(kvMsg.getKey(), kvMsg);
            }

            // now apply pending updates, discard out-of-getSequence messages
            while (true) {
                kvsimple kvMsg = kvsimple.recv(subscriber);

                if (kvMsg == null)
                    break;

                if (kvMsg.getSequence() > sequence) {
                    sequence = kvMsg.getSequence();
                    System.out.println("receiving " + sequence);
                    clonecli2.kvMap.put(kvMsg.getKey(), kvMsg);
                }
            }
        }
    }

    public static void main(String[] args)
    {
        new clonecli2().run();
    }
}

"""
Clone client Model Two

Author: Min RK <benjaminrk@gmail.com>

"""

import time

import zmq

from kvsimple import KVMsg

def main():

    # Prepare our context and subscriber
    ctx = zmq.Context()
    snapshot = ctx.socket(zmq.DEALER)
    snapshot.linger = 0
    snapshot.connect("tcp://localhost:5556")
    subscriber = ctx.socket(zmq.SUB)
    subscriber.linger = 0
    subscriber.setsockopt(zmq.SUBSCRIBE, '')
    subscriber.connect("tcp://localhost:5557")

    kvmap = {}

    # Get state snapshot
    sequence = 0
    snapshot.send("ICANHAZ?")
    while True:
        try:
            kvmsg = KVMsg.recv(snapshot)
        except:
            break;          # Interrupted

        if kvmsg.key == "KTHXBAI":
            sequence = kvmsg.sequence
            print "Received snapshot=%d" % sequence
            break          # Done
        kvmsg.store(kvmap)

    # Now apply pending updates, discard out-of-sequence messages
    while True:
        try:
            kvmsg = KVMsg.recv(subscriber)
        except:
            break          # Interrupted
        if kvmsg.sequence > sequence:
            sequence = kvmsg.sequence
            kvmsg.store(kvmap)

if __name__ == '__main__':
    main()

#
# Clone client Model Two
#

lappend auto_path .
package require KVSimple

#  Prepare our context and subscriber
zmq context context
set snp [zmq socket snapshot context DEALER]
$snp connect "tcp://localhost:5556"
set sub [zmq socket subscriber context SUB]
$sub setsockopt SUBSCRIBE ""
$sub connect "tcp://localhost:5557"

# Get state snapshot
set sequence 0
$snp send "ICANHAZ?"
while {1} {
    set kvmsg [KVSimple new]
    $kvmsg recv $snp
    if {[$kvmsg key] eq "KTHXBAI"} {
	set sequence [$kvmsg sequence]
	puts "Received snapshot=$sequence"
	$kvmsg destroy
	break
    }
    $kvmsg store kvmap
}

#  Now apply pending updates, discard out-of-sequence messages
while {1} {
    set kvmsg [KVSimple new]
    $kvmsg recv $sub
    puts [$kvmsg dump]
    if {[$kvmsg sequence] > $sequence} {
	puts "    store"
	$kvmsg store kvmap
    } else {
	puts "    ignore"
	$kvmsg destroy
    }
}

$snp close
$sub close
context term

//  Clone server - Model Three

//  Lets us build this source without creating a library
#include "kvsimple.c"

//  Routing information for a key-value snapshot
typedef struct {
    void *socket;           //  ROUTER socket to send to
    zframe_t *identity;     //  Identity of peer who requested state
} kvroute_t;

//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int
s_send_single (const char *key, void *data, void *args)
{
    kvroute_t *kvroute = (kvroute_t *) args;
    //  Send identity of recipient first
    zframe_send (&kvroute->identity,
        kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);
    kvmsg_t *kvmsg = (kvmsg_t *) data;
    kvmsg_send (kvmsg, kvroute->socket);
    return 0;
}

int main (void)
{
    //  Prepare our context and sockets
    zctx_t *ctx = zctx_new ();
    void *snapshot = zsocket_new (ctx, ZMQ_ROUTER);
    zsocket_bind (snapshot, "tcp://*:5556");
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (publisher, "tcp://*:5557");
    void *collector = zsocket_new (ctx, ZMQ_PULL);
    zsocket_bind (collector, "tcp://*:5558");

    //  .split body of main task
    //  The body of the main task collects updates from clients and
    //  publishes them back out to clients:

    int64_t sequence = 0;
    zhash_t *kvmap = zhash_new ();

    zmq_pollitem_t items [] = {
        { collector, 0, ZMQ_POLLIN, 0 },
        { snapshot, 0, ZMQ_POLLIN, 0 }
    };
    while (!zctx_interrupted) {
        int rc = zmq_poll (items, 2, 1000 * ZMQ_POLL_MSEC);

        //  Apply state update sent from client
        if (items [0].revents & ZMQ_POLLIN) {
            kvmsg_t *kvmsg = kvmsg_recv (collector);
            if (!kvmsg)
                break;          //  Interrupted
            kvmsg_set_sequence (kvmsg, ++sequence);
            kvmsg_send (kvmsg, publisher);
            kvmsg_store (&kvmsg, kvmap);
            printf ("I: publishing update %5d\n", (int) sequence);
        }
        //  Execute state snapshot request
        if (items [1].revents & ZMQ_POLLIN) {
            zframe_t *identity = zframe_recv (snapshot);
            if (!identity)
                break;          //  Interrupted

            //  Request is in second frame of message
            char *request = zstr_recv (snapshot);
            if (streq (request, "ICANHAZ?"))
                free (request);
            else {
                printf ("E: bad request, aborting\n");
                break;
            }
            //  Send state snapshot to client
            kvroute_t routing = { snapshot, identity };

            //  For each entry in kvmap, send kvmsg to client
            zhash_foreach (kvmap, s_send_single, &routing);

            //  Now send END message with sequence number
            printf ("I: sending shapshot=%d\n", (int) sequence);
            zframe_send (&identity, snapshot, ZFRAME_MORE);
            kvmsg_t *kvmsg = kvmsg_new (sequence);
            kvmsg_set_key  (kvmsg, "KTHXBAI");
            kvmsg_set_body (kvmsg, (byte *) "", 0);
            kvmsg_send     (kvmsg, snapshot);
            kvmsg_destroy (&kvmsg);
        }
    }
    printf (" Interrupted\n%d messages handled\n", (int) sequence);
    zhash_destroy (&kvmap);
    zctx_destroy (&ctx);

    return 0;
}

#include "kvsimple.hpp"

//  Routing information for a key-value snapshot
typedef struct {
    zmq::socket_t *socket; //  ROUTER socket to send to
    std::string identity; //  Identity of peer who requested state
} kvroute_t;

//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int s_send_snapshot(std::unordered_map<std::string, kvmsg>& kvmap, kvroute_t& kvroute) {
    for (auto& kv : kvmap) {
        s_sendmore(*kvroute.socket, kvroute.identity);
        kv.second.send(*kvroute.socket);
    }
    return 0;
}

int main(void) {
    //  Prepare our context and sockets
    zmq::context_t ctx(1);
    zmq::socket_t snapshot(ctx, ZMQ_ROUTER);
    snapshot.bind("tcp://*:5556");
    zmq::socket_t publisher(ctx, ZMQ_PUB);
    publisher.bind("tcp://*:5557");
    zmq::socket_t collector(ctx, ZMQ_PULL);
    collector.bind("tcp://*:5558");

    //  .split body of main task
    //  The body of the main task collects updates from clients and
    //  publishes them back out to clients:
    std::unordered_map<std::string, kvmsg> kvmap;
    int64_t sequence = 0;

    zmq::pollitem_t items[] = {
        {collector, 0, ZMQ_POLLIN, 0},
        {snapshot, 0, ZMQ_POLLIN, 0}
    };

    s_catch_signals();
    while(!s_interrupted) {
        try {
            zmq::poll(items, 2, -1);
        } catch (const zmq::error_t& e) {
            break; //  Interrupted
        }
        //  Apply state update sent from client
        if (items[0].revents & ZMQ_POLLIN) {
            auto msg = kvmsg::recv(collector);
            if (!msg) {
                break; //  Interrupted
            }
            msg->set_sequence(++sequence);
            kvmap[msg->key()] = *msg;
            msg->send(publisher);
            std::cout << "I: publishing update " << sequence << std::endl;
        }
        //  Execute state snapshot request
        if (items[1].revents & ZMQ_POLLIN) {
            std::string identity = s_recv(snapshot);
            std::string request = s_recv(snapshot);
            if (request != "ICANHAZ?") {
                std::cerr << "E: bad request, aborting\n";
                break;
            }
            //  Send state snapshot to client
            kvroute_t kvroute = {&snapshot, identity};

            //  For each entry in kvmap, send kvmsg to client
            s_send_snapshot(kvmap, kvroute);

            //  Now send END message with sequence number
            std::cout << "I: sending state snapshot=" << sequence << std::endl;
            s_sendmore(snapshot, identity);
            kvmsg msg("KTHXBAI", sequence, (unsigned char *)"");
            msg.send(snapshot);
        }
    }
    std::cout << "Interrupted\n" << sequence << " messages handled\n";
    return 0;
}

package guide;

import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

/**
 * Clone server Model Three
 * @author Danish Shrestha <dshrestha06@gmail.com>
 *
 */
public class clonesrv3
{
    private static Map<String, kvsimple> kvMap = new LinkedHashMap<String, kvsimple>();

    public void run()
    {
        try (ZContext ctx = new ZContext()) {
            Socket snapshot = ctx.createSocket(SocketType.ROUTER);
            snapshot.bind("tcp://*:5556");

            Socket publisher = ctx.createSocket(SocketType.PUB);
            publisher.bind("tcp://*:5557");

            Socket collector = ctx.createSocket(SocketType.PULL);
            collector.bind("tcp://*:5558");

            Poller poller = ctx.createPoller(2);
            poller.register(collector, Poller.POLLIN);
            poller.register(snapshot, Poller.POLLIN);

            long sequence = 0;
            while (!Thread.currentThread().isInterrupted()) {
                if (poller.poll(1000) < 0)
                    break; //  Context has been shut down

                // apply state updates from main thread
                if (poller.pollin(0)) {
                    kvsimple kvMsg = kvsimple.recv(collector);
                    if (kvMsg == null) //  Interrupted
                        break;
                    kvMsg.setSequence(++sequence);
                    kvMsg.send(publisher);
                    clonesrv3.kvMap.put(kvMsg.getKey(), kvMsg);
                    System.out.printf("I: publishing update %5d\n", sequence);
                }

                // execute state snapshot request
                if (poller.pollin(1)) {
                    byte[] identity = snapshot.recv(0);
                    if (identity == null)
                        break; //  Interrupted
                    String request = snapshot.recvStr();

                    if (!request.equals("ICANHAZ?")) {
                        System.out.println("E: bad request, aborting");
                        break;
                    }

                    Iterator<Entry<String, kvsimple>> iter = kvMap.entrySet()
                                                                  .iterator();
                    while (iter.hasNext()) {
                        Entry<String, kvsimple> entry = iter.next();
                        kvsimple msg = entry.getValue();
                        System.out.println(
                            "Sending message " + entry.getValue().getSequence()
                        );
                        this.sendMessage(msg, identity, snapshot);
                    }

                    // now send end message with getSequence number
                    System.out.println("Sending state snapshot = " + sequence);
                    snapshot.send(identity, ZMQ.SNDMORE);
                    kvsimple message = new kvsimple(
                        "KTHXBAI", sequence, ZMQ.SUBSCRIPTION_ALL
                    );
                    message.send(snapshot);
                }
            }

            System.out.printf(" Interrupted\n%d messages handled\n", sequence);
        }
    }

    private void sendMessage(kvsimple msg, byte[] identity, Socket snapshot)
    {
        snapshot.send(identity, ZMQ.SNDMORE);
        msg.send(snapshot);
    }

    public static void main(String[] args)
    {
        new clonesrv3().run();
    }
}

"""
Clone server Model Three

Author: Min RK <benjaminrk@gmail.com
"""

import zmq

from kvsimple import KVMsg

# simple struct for routing information for a key-value snapshot
class Route:
    def __init__(self, socket, identity):
        self.socket = socket # ROUTER socket to send to
        self.identity = identity # Identity of peer who requested state

def send_single(key, kvmsg, route):
    """Send one state snapshot key-value pair to a socket"""
    # Send identity of recipient first
    route.socket.send(route.identity, zmq.SNDMORE)
    kvmsg.send(route.socket)

def main():
    # context and sockets
    ctx = zmq.Context()
    snapshot = ctx.socket(zmq.ROUTER)
    snapshot.bind("tcp://*:5556")
    publisher = ctx.socket(zmq.PUB)
    publisher.bind("tcp://*:5557")
    collector = ctx.socket(zmq.PULL)
    collector.bind("tcp://*:5558")

    sequence = 0
    kvmap = {}

    poller = zmq.Poller()
    poller.register(collector, zmq.POLLIN)
    poller.register(snapshot, zmq.POLLIN)
    while True:
        try:
            items = dict(poller.poll(1000))
        except:
            break           # Interrupted

        # Apply state update sent from client
        if collector in items:
            kvmsg = KVMsg.recv(collector)
            sequence += 1
            kvmsg.sequence = sequence
            kvmsg.send(publisher)
            kvmsg.store(kvmap)
            print "I: publishing update %5d" % sequence

        # Execute state snapshot request
        if snapshot in items:
            msg = snapshot.recv_multipart()
            identity = msg[0]
            request = msg[1]
            if request == "ICANHAZ?":
                pass
            else:
                print "E: bad request, aborting\n",
                break

            # Send state snapshot to client
            route = Route(snapshot, identity)

            # For each entry in kvmap, send kvmsg to client
            for k,v in kvmap.items():
                send_single(k,v,route)

            # Now send END message with sequence number
            print "Sending state shapshot=%d\n" % sequence,
            snapshot.send(identity, zmq.SNDMORE)
            kvmsg = KVMsg(sequence)
            kvmsg.key = "KTHXBAI"
            kvmsg.body = ""
            kvmsg.send(snapshot)

    print " Interrupted\n%d messages handled" % sequence


if __name__ == '__main__':
    main()