//  Decentralized chat example
//
#include <czmq.h>

static void
listener_task (void *args, zctx_t *ctx, void *pipe)
{
    void *listener = zsocket_new (ctx, ZMQ_SUB);
    int address;
    for (address = 1; address < 255; address++) {
        int rc = zsocket_connect (listener,
            "tcp://%s%d:9000", (char *) args, address);
        assert (rc == 0);
    }
    zsocket_set_subscribe (listener, "");
    while (!zctx_interrupted) {
        char *message = zstr_recv (listener);
        if (message) {
            printf ("%s", message);
            free (message);
        }
    }
}

int main (int argc, char *argv [])
{
    char *dot = argc == 4? strrchr (argv [1], '.'): NULL;
    if (dot)
        dot [1] = 0;        //  Cut string after dot
    else {
        puts ("Usage: dechat ipaddress interface username");
        puts ("Example: dechat 192.168.55.123 eth0 joe");
        exit (0);
    }
    zctx_t *ctx = zctx_new ();
    zthread_fork (ctx, listener_task, argv [1]);

    void *broadcaster = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (broadcaster, "tcp://%s:9000", argv [2]);
    while (!zctx_interrupted) {
        char message [1024];
        if (!fgets (message, 1024, stdin))
            break;
        zstr_sendf (broadcaster, "%s: %s", argv [3], message);
    }
    zctx_destroy (&ctx);
    return 0;
}

import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;
import org.zeromq.ZThread;
import zmq.ZError;

import java.util.Scanner;

/**
 * Decentralized chat example
 */
public class Dechat {
  public static void main(String[] args) {
    if (args.length < 3) {
      System.out.println("Usage: dechat ipaddress interface username");
      System.out.println("Example: dechat 192.168.55.123 localhost joe");
      System.exit(0);
    }
    ZContext ctx = new ZContext();

    //cut string after dot
    String addressWithoutLastPart = args[0].substring(0, args[0].lastIndexOf('.'));
    ZThread.fork(ctx, new ListenerTask(), addressWithoutLastPart);
    ZMQ.Socket broadcaster = ctx.createSocket(ZMQ.PUB);
    broadcaster.bind(String.format("tcp://%s:9000", args[1]));
    Scanner scanner = new Scanner(System.in);
    while (!Thread.currentThread().isInterrupted()) {
      String line = scanner.nextLine();
      if (line.isEmpty())
        break;
      broadcaster.send(String.format("%s: %s", args[2], line));
    }
    ctx.destroy();
  }

  static class ListenerTask implements ZThread.IAttachedRunnable {

    @Override
    public void run(Object[] args, ZContext ctx, ZMQ.Socket pipe) {
      ZMQ.Socket listener = ctx.createSocket(ZMQ.SUB);
      int address;
      for (address = 1; address < 255; address++)
        listener.connect(String.format("tcp://%s.%d:9000", args[0], address));

      listener.subscribe(ZMQ.SUBSCRIPTION_ALL);
      while (!Thread.currentThread().isInterrupted()) {
        String message;
        try {
          message = listener.recvStr();
        } catch (ZMQException e) {
          if (e.getErrorCode() == ZError.ETERM)
            break;
          e.printStackTrace();
          break;
        }
        if (!message.isEmpty())
          System.out.println(message);
      }
    }
  }
}

"""Decentralized chat example"""
try:
    raw_input
except NameError:
    raw_input = input

import argparse
import os
from threading import Thread

# dependency, not in stdlib
from netifaces import interfaces, ifaddresses, AF_INET

import zmq


def listen(masked):
    """listen for messages

    masked is the first three parts of an IP address:

        192.168.1

    The socket will connect to all of X.Y.Z.{1-254}.
    """
    ctx = zmq.Context.instance()
    listener = ctx.socket(zmq.SUB)
    for last in range(1, 255):
        listener.connect("tcp://{0}.{1}:9000".format(masked, last))

    listener.setsockopt(zmq.SUBSCRIBE, b'')
    while True:
        try:
            print(listener.recv_string())
        except (KeyboardInterrupt, zmq.ContextTerminated):
            break


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("interface", type=str, help="the network interface",
                        choices=interfaces(),
                        )
    parser.add_argument("user", type=str, default=os.environ['USER'],
                        nargs='?',
                        help="Your username",
                        )
    args = parser.parse_args()
    inet = ifaddresses(args.interface)[AF_INET]
    addr = inet[0]['addr']
    masked = addr.rsplit('.', 1)[0]

    ctx = zmq.Context.instance()

    listen_thread = Thread(target=listen, args=(masked,))
    listen_thread.start()

    bcast = ctx.socket(zmq.PUB)
    bcast.bind("tcp://%s:9000" % args.interface)
    print("starting chat on %s:9000 (%s.*)" % (args.interface, masked))
    while True:
        try:
            msg = raw_input()
            bcast.send_string("%s: %s" % (args.user, msg))
        except KeyboardInterrupt:
            break
    bcast.close(linger=0)
    ctx.term()


if __name__ == '__main__':
    main()

//  UDP ping command
//  Model 1, does UDP work inline

#include <czmq.h>
#define PING_PORT_NUMBER 9999
#define PING_MSG_SIZE    1
#define PING_INTERVAL    1000  //  Once per second

static void
derp (char *s)
{
    perror (s);
    exit (1);
}

int main (void)
{
    zctx_t *ctx = zctx_new ();

    //  Create UDP socket
    int fd;
    if ((fd = socket (AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1)
        derp ("socket");

    //  Ask operating system to let us do broadcasts from socket
    int on = 1;
    if (setsockopt (fd, SOL_SOCKET, SO_BROADCAST, &on, sizeof (on)) == -1)
        derp ("setsockopt (SO_BROADCAST)");

    //  Bind UDP socket to local port so we can receive pings
    struct sockaddr_in si_this = { 0 };
    si_this.sin_family = AF_INET;
    si_this.sin_port = htons (PING_PORT_NUMBER);
    si_this.sin_addr.s_addr = htonl (INADDR_ANY);
    if (bind (fd, &si_this, sizeof (si_this)) == -1)
        derp ("bind");
    
    byte buffer [PING_MSG_SIZE];
    
    //  .split main ping loop
    //  We use {{zmq_poll}} to wait for activity on the UDP socket, because
    //  this function works on non-0MQ file handles. We send a beacon
    //  once a second, and we collect and report beacons that come in
    //  from other nodes:
    
    zmq_pollitem_t pollitems [] = {{ NULL, fd, ZMQ_POLLIN, 0 }};
    //  Send first ping right away
    uint64_t ping_at = zclock_time ();
    
    while (!zctx_interrupted) {
        long timeout = (long) (ping_at - zclock_time ());
        if (timeout < 0)
            timeout = 0;
        if (zmq_poll (pollitems, 1, timeout * ZMQ_POLL_MSEC) == -1)
            break;              //  Interrupted

        //  Someone answered our ping
        if (pollitems [0].revents & ZMQ_POLLIN) {
            struct sockaddr_in si_that;
            socklen_t si_len = sizeof (struct sockaddr_in);
            ssize_t size = recvfrom (fd, buffer, PING_MSG_SIZE, 0,
                                     &si_that, &si_len);
            if (size == -1)
                derp ("recvfrom");
            printf ("Found peer %s:%d\n",
                inet_ntoa (si_that.sin_addr), ntohs (si_that.sin_port));
        }
        if (zclock_time () >= ping_at) {
            //  Broadcast our beacon
            puts ("Pinging peers...");
            buffer [0] = '!';
            struct sockaddr_in si_that = si_this;
            inet_aton ("255.255.255.255", &si_that.sin_addr);
            if (sendto (fd, buffer, PING_MSG_SIZE, 0,
                        &si_that, sizeof (struct sockaddr_in)) == -1)
                derp ("sendto");
            ping_at = zclock_time () + PING_INTERVAL;
        }
    }
    close (fd);
    zctx_destroy (&ctx);
    return 0;
}

/*******************************************************************************
* Author      : James Chapman
* License     : MIT/X11 (http://directory.fsf.org/wiki/License:X11)
* Description : Simple UDP ping example
********************************************************************************/

#include <atomic>
#include <iostream>
#include <string>
#include <thread>

#ifdef _WIN32
#include <WS2tcpip.h>
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#endif // _WIN32

#include "zmq.hpp"

#ifdef _WIN32
#pragma comment(lib, "Ws2_32.lib")
#endif // _WIN32

#define PING_PORT_NUMBER    9999
#define PING_MSG_SIZE       2
#define PING_INTERVAL       500
#define SOCKET_POLL_TIMEOUT 3000

#define INFO_OUT(MSG)  std::cout << "[INFO]   " << " " << (MSG) << std::endl
#define ERROR_OUT(MSG) std::cerr << "[ERROR]  " << " " << (MSG) << std::endl

#ifndef _WIN32
#define SOCKET int
#define INVALID_SOCKET (SOCKET)(~0)
#define SOCKET_ERROR (SOCKET)(~1)
#define NO_ERROR 0
#endif // _WIN32

std::atomic<bool> g_threadInterupted(false);

/**
* Create a socket and use ZeroMQ to poll.
*/
void listener()
{
    #ifdef _WIN32
    WSADATA wsaData;
    #endif // _WIN32
    int nResult = 0;

    #ifdef _WIN32
    // Initialize Winsock
    nResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
    if (nResult != NO_ERROR)
    {
        ERROR_OUT("zmqListen : WSAStartup failed");
    }
    #endif // _WIN32

    // Create UDP socket
    SOCKET fdSocket;
    fdSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
    if (fdSocket == INVALID_SOCKET)
    {
        ERROR_OUT("zmqListen : Socket creation failed");
    }

    // Set up the sockaddr structure
    struct sockaddr_in saListen = {0};
    saListen.sin_family = AF_INET;
    saListen.sin_port = htons(PING_PORT_NUMBER);
    saListen.sin_addr.s_addr = htonl(INADDR_ANY);

    // Bind the socket
    nResult = bind(fdSocket, (sockaddr*)&saListen, sizeof(saListen));
    if (nResult != NO_ERROR)
    {
        ERROR_OUT("zmqListen : socket bind failed");
    }

    while (!g_threadInterupted)
    {
        // Poll socket for a message
        zmq::pollitem_t items[] = {{NULL, fdSocket, ZMQ_POLLIN, 0}};
        zmq::poll(&items[0], 1, SOCKET_POLL_TIMEOUT);

        // If we get a message, print the contents
        if (items[0].revents & ZMQ_POLLIN)
        {
            char recvBuf[PING_MSG_SIZE] = {0};
            #ifdef _WIN32
            int saSize = sizeof(struct sockaddr_in);
            #else
            socklen_t saSize = sizeof(struct sockaddr_in);
            #endif // _WIN32
            size_t size = recvfrom(fdSocket, recvBuf, PING_MSG_SIZE + 1, 0, (sockaddr*)&saListen, &saSize);
            {
                std::string ip(inet_ntoa(saListen.sin_addr));
                INFO_OUT("received: " + std::string(recvBuf) + " from " + ip);
            }
        }
    }

    #ifdef _WIN32
    closesocket(fdSocket);
    WSACleanup();
    #endif // _WIN32
}

/**
* Main function starts a listener thread and then sends out 5 broadcast pings
*/
int main()
{
    g_threadInterupted = false;

    // Start listener in a seperate thread
    std::thread listenerThread(listener);

    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    {
        #ifdef _WIN32
        WSADATA wsaData;
        #endif // _WIN32
        int nResult = 0;
        int nOptOffVal = 0;
        int nOptOnVal = 1;
        int nOptLen = sizeof(int);

        #ifdef _WIN32
        // Initialize Winsock
        nResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
        if (nResult != NO_ERROR)
        {
            ERROR_OUT("broadcast : WSAStartup failed");
        }
        #endif // _WIN32

        // Create UDP socket
        SOCKET fdSocket;
        fdSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
        if (fdSocket == INVALID_SOCKET)
        {
            ERROR_OUT("broadcast : socket creation failed");
        }

        // Ask operating system to let us do broadcasts from socket
        nResult = setsockopt(fdSocket, SOL_SOCKET, SO_BROADCAST, (char *)&nOptOnVal, nOptLen);
        if (nResult != NO_ERROR)
        {
            ERROR_OUT("broadcast : setsockopt SO_BROADCAST failed");
        }

        // Set up the sockaddr structure
        struct sockaddr_in saBroadcast = {0};
        saBroadcast.sin_family = AF_INET;
        saBroadcast.sin_port = htons(PING_PORT_NUMBER);
        saBroadcast.sin_addr.s_addr = htonl(INADDR_BROADCAST);

        // Broadcast 5 beacon messages
        for (int i = 0; i < 5; i++)
        {
            char buffer[PING_MSG_SIZE] = {0};
            strcpy(&buffer[0], "!");
            int bytes = sendto(fdSocket, buffer, PING_MSG_SIZE + 1, 0, (sockaddr*)&saBroadcast, sizeof(struct sockaddr_in));
            if (bytes == SOCKET_ERROR)
            {
                ERROR_OUT("broadcast : sendto failed");
            }

            std::this_thread::sleep_for(std::chrono::milliseconds(PING_INTERVAL));
        }

        #ifdef _WIN32
        closesocket(fdSocket);
        WSACleanup();
        #endif // _WIN32
    }

    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    g_threadInterupted = true;

    listenerThread.join();

    return 0;
}

import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;

/**
 * UDP ping command
 * Model 1, does UDP work inline
 */
public class Udpping1 {
  private static final int PING_PORT_NUMBER = 9999;
  private static final int PING_MSG_SIZE = 1;
  private static final int PING_INTERVAL = 1000; //  Once per second

  public static void main(String[] args) throws IOException {
    ZContext ctx = new ZContext();

    //  Create UDP socket
    DatagramChannel udpChannel = DatagramChannel.open();

    //  Ask operating system to let us do broadcasts from socket
    udpChannel.setOption(StandardSocketOptions.SO_BROADCAST, true);
    udpChannel.configureBlocking(false);
    //  Bind UDP socket to local port so we can receive pings
    udpChannel.bind(new InetSocketAddress(PING_PORT_NUMBER));
    ByteBuffer buffer = ByteBuffer.allocate(PING_MSG_SIZE);

    //  We use zmq_poll to wait for activity on the UDP socket, because
    //  this function works on non-0MQ file handles. We send a beacon
    //  once a second, and we collect and report beacons that come in
    //  from other nodes:
    PollItem[] pollItems = new PollItem[]{new PollItem(udpChannel, Poller.POLLIN)};
    //  Send first ping right away
    long pingAt = System.currentTimeMillis();

    while (!Thread.currentThread().isInterrupted()) {
      long timeout = pingAt - System.currentTimeMillis();
      if (timeout < 0)
        timeout = 0;
      if (ZMQ.poll(pollItems, 1, timeout) == -1)
        break;              //  Interrupted

      //  Someone answered our ping
      if (pollItems[0].isReadable()) {
        InetSocketAddress sourceAddr = (InetSocketAddress) udpChannel.receive(buffer);
        System.out.printf("Found peer %s:%d\n", sourceAddr.getAddress(), sourceAddr.getPort());
      }
      if (System.currentTimeMillis() >= pingAt) {
        //  Broadcast our beacon
        System.out.println("Pinging peers…");
        buffer.put(0, (byte) '!');
        udpChannel.send(buffer, new InetSocketAddress("255.255.255.255", PING_PORT_NUMBER));
        pingAt = System.currentTimeMillis() + PING_INTERVAL;
      }
    }
    udpChannel.close();
    ctx.destroy();
  }
}

#
# UDP ping command
# Model 1
#

import socket
import time

import zmq

#include <czmq.h>
PING_PORT_NUMBER = 9999
PING_MSG_SIZE    = 1
PING_INTERVAL    = 1  # Once per second

def main():

    # Create UDP socket
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)

    # Ask operating system to let us do broadcasts from socket
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)

    # Bind UDP socket to local port so we can receive pings
    sock.bind(('', PING_PORT_NUMBER))

    # main ping loop
    # We use zmq_poll to wait for activity on the UDP socket, since
    # this function works on non-0MQ file handles. We send a beacon
    # once a second, and we collect and report beacons that come in
    # from other nodes:

    poller = zmq.Poller()
    poller.register(sock, zmq.POLLIN)

    # Send first ping right away
    ping_at = time.time()

    while True:
        timeout = ping_at - time.time()
        if timeout < 0:
            timeout = 0
        try:
            events = dict(poller.poll(1000* timeout))
        except KeyboardInterrupt:
            print("interrupted")
            break

        # Someone answered our ping
        if sock.fileno() in events:
            msg, addrinfo = sock.recvfrom(PING_MSG_SIZE)
            print("Found peer %s:%d" % addrinfo)

        if time.time() >= ping_at:
            # Broadcast our beacon
            print ("Pinging peers...")
            sock.sendto(b'!', 0, ("255.255.255.255", PING_PORT_NUMBER))
            ping_at = time.time() + PING_INTERVAL

if __name__ == '__main__':
    main()

//  UDP ping command
//  Model 2, uses separate UDP library

#include <czmq.h>
#include "udplib.c"
#define PING_PORT_NUMBER 9999
#define PING_MSG_SIZE    1
#define PING_INTERVAL    1000  //  Once per second

int main (void)
{
    zctx_t *ctx = zctx_new ();
    udp_t *udp = udp_new (PING_PORT_NUMBER);
    
    byte buffer [PING_MSG_SIZE];
    zmq_pollitem_t pollitems [] = {
        { NULL, udp_handle (udp), ZMQ_POLLIN, 0 }
    };
    //  Send first ping right away
    uint64_t ping_at = zclock_time ();
    
    while (!zctx_interrupted) {
        long timeout = (long) (ping_at - zclock_time ());
        if (timeout < 0)
            timeout = 0;
        if (zmq_poll (pollitems, 1, timeout * ZMQ_POLL_MSEC) == -1)
            break;              //  Interrupted

        //  Someone answered our ping
        if (pollitems [0].revents & ZMQ_POLLIN)
            udp_recv (udp, buffer, PING_MSG_SIZE);
        
        if (zclock_time () >= ping_at) {
            puts ("Pinging peers...");
            buffer [0] = '!';
            udp_send (udp, buffer, PING_MSG_SIZE);
            ping_at = zclock_time () + PING_INTERVAL;
        }
    }
    udp_destroy (&udp);
    zctx_destroy (&ctx);
    return 0;
}

import org.zeromq.ZContext;
import org.zeromq.ZMQ;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;

/**
 * UDP ping command
 * Model 2, uses separate UDP library
 */
public class Udpping2 {
  private static final int PING_PORT_NUMBER = 9999;
  private static final int PING_MSG_SIZE = 1;
  private static final int PING_INTERVAL = 1000; //  Once per second

  public static void main(String[] args) throws IOException {
    ZContext ctx = new ZContext();
    Udplib udp = new Udplib(PING_PORT_NUMBER);

    ByteBuffer buffer = ByteBuffer.allocate(PING_MSG_SIZE);
    ZMQ.PollItem[] pollItems = new ZMQ.PollItem[]{
            new ZMQ.PollItem(udp.handle(), ZMQ.Poller.POLLIN)};
    //  Send first ping right away
    long pingAt = System.currentTimeMillis();

    while (!Thread.currentThread().isInterrupted()) {
      long timeout = pingAt - System.currentTimeMillis();
      if (timeout < 0)
        timeout = 0;
      if (ZMQ.poll(pollItems, 1, timeout) == -1)
        break;              //  Interrupted

      //  Someone answered our ping
      if (pollItems[0].isReadable()){
        SocketAddress sourceAddr = udp.recv(buffer);
        System.out.printf("new peer found: %s\n",sourceAddr);
      }

      if (System.currentTimeMillis() >= pingAt) {
        System.out.println("Pinging peers…");
        buffer.put(0, (byte) '!');
        udp.send(buffer);
        pingAt = System.currentTimeMillis() + PING_INTERVAL;
      }
    }
    udp.destroy();
    ctx.destroy();
  }
}

#
# UDP ping command
# Model 2
#

import time

import zmq
from udplib import UDP

PING_PORT_NUMBER = 9999
PING_MSG_SIZE    = 1
PING_INTERVAL    = 1  # Once per second

def main():

    udp = UDP(PING_PORT_NUMBER)

    poller = zmq.Poller()
    poller.register(udp.handle, zmq.POLLIN)

    # Send first ping right away
    ping_at = time.time()

    while True:
        timeout = ping_at - time.time()
        if timeout < 0:
            timeout = 0
        try:
            events = dict(poller.poll(1000* timeout))
        except KeyboardInterrupt:
            print("interrupted")
            break

        # Someone answered our ping
        if udp.handle.fileno() in events:
            udp.recv(PING_MSG_SIZE)

        if time.time() >= ping_at:
            # Broadcast our beacon
            print ("Pinging peers...")
            udp.send(b'!')
            ping_at = time.time() + PING_INTERVAL

if __name__ == '__main__':
    main()

//  UDP ping command
//  Model 3, uses abstract network interface

#include <czmq.h>
#include "interface.c"

int main (void)
{
    interface_t *interface = interface_new ();
    while (true) {
        zmsg_t *msg = interface_recv (interface);
        if (!msg)
            break;              //  Interrupted
        zmsg_dump (msg);
    }
    interface_destroy (&interface);
    return 0;
}

import org.zeromq.ZMsg;

/**
 * UDP ping command
 * Model 3, uses abstract network interface
 */
public class Udpping3 {
  public static void main(String[] args) {
    Interface interfaceT = new Interface();
    while (true) {
      ZMsg msg = interfaceT.recv();
      if (msg == null)
        break;  //  Interrupted
      msg.dump();
    }
    interfaceT.destroy();
  }
}

"""UDP ping command
Model 3, uses abstract network interface
"""

from interface import Interface

def main():
    interface = Interface()
    while True:
        try:
            print(interface.recv())
        except KeyboardInterrupt:
            print("interrupted")
            break
    interface.stop()

if __name__ == '__main__':
    main()

//  Interface class for Chapter on Distributed Computing
//  This implements an "interface" to our network of nodes

#include <czmq.h>
#include <uuid/uuid.h>
#include "udplib.c"

//  =====================================================================
//  Synchronous part, works in our application thread

//  Structure of our class

typedef struct {
    zctx_t *ctx;                //  Our context wrapper
    void *pipe;                 //  Pipe through to agent
} interface_t;

//  This is the thread that handles our real interface class
static void
    interface_agent (void *args, zctx_t *ctx, void *pipe);

//  .split constructor and destructor
//  Here are the constructor and destructor for the interface class.
//  Note that the class has barely any properties; it is just an excuse
//  to start the background thread and a wrapper around {{zmsg_recv}}:

interface_t *
interface_new (void)
{
    interface_t
        *self;

    self = (interface_t *) zmalloc (sizeof (interface_t));
    self->ctx = zctx_new ();
    self->pipe = zthread_fork (self->ctx, interface_agent, NULL);
    return self;
}

void
interface_destroy (interface_t **self_p)
{
    assert (self_p);
    if (*self_p) {
        interface_t *self = *self_p;
        zctx_destroy (&self->ctx);
        free (self);
        *self_p = NULL;
    }
}

//  .split receive message
//  Here we wait for a message from the interface. This returns
//  us a {{zmsg_t}} object, or NULL if interrupted:

static zmsg_t *
interface_recv (interface_t *self)
{
    assert (self);
    zmsg_t *msg = zmsg_recv (self->pipe);
    return msg;
}

//  =====================================================================
//  Asynchronous part, works in the background

//  .split peer class
//  This structure defines each peer that we discover and track:

typedef struct {
    uuid_t uuid;                //  Peer's UUID as binary blob
    char *uuid_str;             //  UUID as printable string
    uint64_t expires_at;
} peer_t;

#define PING_PORT_NUMBER 9999
#define PING_INTERVAL    1000  //  Once per second
#define PEER_EXPIRY      5000  //  Five seconds and it's gone

//  Convert binary UUID to freshly allocated string

static char *
s_uuid_str (uuid_t uuid)
{
    char hex_char [] = "0123456789ABCDEF";
    char *string = zmalloc (sizeof (uuid_t) * 2 + 1);
    int byte_nbr;
    for (byte_nbr = 0; byte_nbr < sizeof (uuid_t); byte_nbr++) {
        string [byte_nbr * 2 + 0] = hex_char [uuid [byte_nbr] >> 4];
        string [byte_nbr * 2 + 1] = hex_char [uuid [byte_nbr] & 15];
    }
    return string;
}

//  .split peer constructor and destructor
//  We have a constructor and destructor for the peer class:

static peer_t *
peer_new (uuid_t uuid)
{
    peer_t *self = (peer_t *) zmalloc (sizeof (peer_t));
    memcpy (self->uuid, uuid, sizeof (uuid_t));
    self->uuid_str = s_uuid_str (self->uuid);
    return self;
}

//  Destroy peer object

static void
peer_destroy (peer_t **self_p)
{
    assert (self_p);
    if (*self_p) {
        peer_t *self = *self_p;
        free (self->uuid_str);
        free (self);
        *self_p = NULL;
    }
}

//  .split peer methods
//  These methods return the peer's UUID in binary format or
//  as a printable string:

static byte *
peer_uuid (peer_t *self)
{
    assert (self);
    return self->uuid;
}

static char *
peer_uuid_str (peer_t *self)
{
    assert (self);
    return self->uuid_str;
}

//  Just resets the peers expiry time; we call this method
//  whenever we get any activity from a peer.

static void
peer_is_alive (peer_t *self)
{
    assert (self);
    self->expires_at = zclock_time () + PEER_EXPIRY;
}

//  Peer hash calls this handler automatically whenever we delete
//  peer from agent peers, or destroy that hash table.

static void
peer_freefn (void *argument)
{
    peer_t *peer = (peer_t *) argument;
    peer_destroy (&peer);
}

//  .split agent class
//  This structure holds the context for our agent so we can
//  pass that around cleanly to methods that need it:

typedef struct {
    zctx_t *ctx;                //  CZMQ context
    void *pipe;                 //  Pipe back to application
    udp_t *udp;                 //  UDP object
    uuid_t uuid;                //  Our UUID as binary blob
    zhash_t *peers;             //  Hash of known peers, fast lookup
} agent_t;

//  .split agent constructor and destructor
//  Now we create the constructor and destructor for our agent. Each 
//  interface has one agent object, which implements its background thread:

static agent_t *
agent_new (zctx_t *ctx, void *pipe)
{
    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));
    self->ctx = ctx;
    self->pipe = pipe;
    self->udp = udp_new (PING_PORT_NUMBER);
    self->peers = zhash_new ();
    uuid_generate (self->uuid);
    return self;
}

static void
agent_destroy (agent_t **self_p)
{
    assert (self_p);
    if (*self_p) {
        agent_t *self = *self_p;
        zhash_destroy (&self->peers);
        udp_destroy (&self->udp);
        free (self);
        *self_p = NULL;
    }
}

//  .skip
//  Here we handle the different control messages from the frontend.

static int
agent_control_message (agent_t *self)
{
    //  Get the whole message off the pipe in one go
    zmsg_t *msg = zmsg_recv (self->pipe);
    char *command = zmsg_popstr (msg);
    if (command == NULL)
        return -1;      //  Interrupted

    //  We don't actually implement any control commands yet
    //  but if we did, this would be where we did it...
    //  if (streq (command, "EXAMPLE")) {
    //  }
    
    free (command);
    zmsg_destroy (&msg);
    return 0;
}
//  .until

//  .split handle beacon
//  This is how we handle a beacon coming into our UDP socket;
//  this may be from other peers or an echo of our own broadcast
//  beacon:

static int
agent_handle_beacon (agent_t *self)
{
    uuid_t uuid;
    ssize_t size = udp_recv (self->udp, uuid, sizeof (uuid_t));

    //  If we got a UUID and it's not our own beacon, we have a peer
    if (size == sizeof (uuid_t)
    &&  memcmp (uuid, self->uuid, sizeof (uuid))) {
        char *uuid_str = s_uuid_str (uuid);
        
        //  Find or create peer via its UUID string
        peer_t *peer = (peer_t *) zhash_lookup (self->peers, uuid_str);
        if (peer == NULL) {
            peer = peer_new (uuid);
            zhash_insert (self->peers, uuid_str, peer);
            zhash_freefn (self->peers, uuid_str, peer_freefn);
            
            //  Report peer joined the network
            zstr_sendm (self->pipe, "JOINED");
            zstr_send (self->pipe, uuid_str);
        }
        //  Any activity from the peer means it's alive
        peer_is_alive (peer);
        free (uuid_str);
    }
    return 0;
}

//  .split reap peers
//  This method checks one peer item for expiration; if the peer hasn't
//  sent us anything by now, it's "dead" and we can delete it:

static int
agent_reap_peer (const char *key, void *item, void *argument)
{
    agent_t *self = (agent_t *) argument;
    peer_t *peer = (peer_t *) item;
    if (zclock_time () >= peer->expires_at) {
        //  Report peer left the network
        zstr_sendm (self->pipe, "LEFT");
        zstr_send (self->pipe, peer_uuid_str (peer));
        zhash_delete (self->peers, peer_uuid_str (peer));
    }
    return 0;
}

//  .split agent main loop
//  This is the main loop for the background agent. It uses {{zmq_poll}}
//  to monitor the frontend pipe (commands from the API) and the
//  backend UDP handle (beacons):

static void
interface_agent (void *args, zctx_t *ctx, void *pipe)
{
    //  Create agent instance to pass around
    agent_t *self = agent_new (ctx, pipe);
    
    //  Send first beacon immediately
    uint64_t ping_at = zclock_time ();
    zmq_pollitem_t pollitems [] = {
        { self->pipe, 0, ZMQ_POLLIN, 0 },
        { 0, udp_handle (self->udp), ZMQ_POLLIN, 0 }
    };
    while (!zctx_interrupted) {
        long timeout = (long) (ping_at - zclock_time ());
        if (timeout < 0)
            timeout = 0;
        if (zmq_poll (pollitems, 2, timeout * ZMQ_POLL_MSEC) == -1)
            break;              //  Interrupted

        //  If we had activity on the pipe, go handle the control
        //  message. Current code never sends control messages.
        if (pollitems [0].revents & ZMQ_POLLIN)
            agent_control_message (self);

        //  If we had input on the UDP socket, go process that
        if (pollitems [1].revents & ZMQ_POLLIN)
            agent_handle_beacon (self);

        //  If we passed the 1-second mark, broadcast our beacon
        if (zclock_time () >= ping_at) {
            udp_send (self->udp, self->uuid, sizeof (uuid_t));
            ping_at = zclock_time () + PING_INTERVAL;
        }
        //  Delete and report any expired peers
        zhash_foreach (self->peers, agent_reap_peer, self);
    }
    agent_destroy (&self);
}

import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;
import org.zeromq.ZThread;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Interface class for Chapter on Distributed Computing
 * This implements an "interface" to our network of nodes
 */
public class Interface {
  ZContext ctx; //  Our context wrapper
  ZMQ.Socket pipe;


  public Interface() {
    ctx = new ZContext();
    pipe = ZThread.fork(ctx, new InterfaceAgent());
  }

  /**
   * Here we wait for a message from the interface.
   *
   * @return Zmsg object or null if interrupted
   */
  public ZMsg recv() {
    return ZMsg.recvMsg(pipe);
  }

  public void destroy() {
    this.ctx.destroy();
  }

  /**
   * Asynchronous part, works in the background
   */
  private static class InterfaceAgent implements ZThread.IAttachedRunnable {
    private static final int PING_PORT_NUMBER = 9999;
    private static final int PING_INTERVAL = 1000; //  Once per second
    private static final byte[] emptyUUID = new byte[16];
    private Udplib udp;                 //  UDP object
    private UUID uuid;                //  Our UUID as binary blob
    private Map<UUID, Peer> peers;
    private ZMQ.Socket pipe;                 //  Pipe back to application

    private static ByteBuffer serializeUUID(UUID obj) {
      ByteBuffer result = ByteBuffer.allocate(16);
      result.putLong(obj.getMostSignificantBits());
      result.putLong(obj.getLeastSignificantBits());
      result.flip();
      return result;
    }

    private static UUID deserializeUUID(ByteBuffer buffer) {
      long msb = buffer.getLong(0);
      long lsb = buffer.getLong(8);
      return new UUID(msb, lsb);
    }

    private void init(ZMQ.Socket pipe) {
      this.pipe = pipe;
      try {
        udp = new Udplib(PING_PORT_NUMBER);
      } catch (IOException e) {
        e.printStackTrace();
        System.exit(1);
      }
      peers = new ConcurrentHashMap<>();
      uuid = UUID.randomUUID();
      System.out.println("I'm " + uuid);
    }

    private void destroy() throws IOException {
      udp.destroy();
    }

    /**
     * Here we handle the different control messages from the frontend.
     */
    private void controlMsg() {
      //  Get the whole message off the pipe in one go
      ZMsg msg = ZMsg.recvMsg(pipe);
      String command = msg.popString();


      //  We don't actually implement any control commands yet
      //  but if we did, this would be where we did it…
      //  if("EXAMPLE".equals(command)) {
      //  }
      msg.destroy();
    }

    /**
     * This is how we handle a beacon coming into our UDP socket;
     * this may be from other peers or an echo of our own broadcast
     * beacon:
     */
    private void handleBeacon() {
      ByteBuffer uuidBuffer = ByteBuffer.allocate(16);
      try {
        udp.recv(uuidBuffer);
      } catch (IOException e) {
        e.printStackTrace();
        System.exit(1);
      }
      if (Arrays.equals(uuidBuffer.array(), emptyUUID))
        return; //nothing to read
      UUID uuid = deserializeUUID(uuidBuffer);
      //  If we got a UUID and it's not our own beacon, we have a peer
      if (!this.uuid.equals(uuid)) {
        //  Find or create peer via its UUID string
        Peer peer = peers.get(uuid);
        if (peer == null) {
          peer = new Peer(uuid);
          peers.put(uuid, peer);

          //  Report peer joined the network
          pipe.sendMore("JOINED");
          pipe.send(serializeUUID(uuid).array());
        }
        //  Any activity from the peer means it's alive
        peer.isAlive();
      }
    }

    /**
     * This method checks one peer peer for expiration; if the peer hasn't
     * sent us anything by now, it's "dead" and we can delete it:
     */
    private void reapPeer(Peer peer) {
      if (System.currentTimeMillis() >= peer.expiresAt) {
        //  Report peer left the network
        pipe.sendMore("LEFT");
        pipe.send(serializeUUID(peer.uuid).array());
        peers.remove(peer.uuid);
      }
    }

    /**
     * This is the main loop for the background agent. It uses zmq_poll
     * to monitor the frontend pipe (commands from the API) and the
     * backend UDP handle (beacons):
     */
    @Override
    public void run(Object[] args, ZContext ctx, ZMQ.Socket pipe) {
      init(pipe);

      //  Send first beacon immediately
      long pingAt = System.currentTimeMillis();
      ZMQ.PollItem[] pollItems = new ZMQ.PollItem[]{
              new ZMQ.PollItem(pipe, ZMQ.Poller.POLLIN),
              new ZMQ.PollItem(udp.handle(), ZMQ.Poller.POLLIN)
      };
      while (!Thread.currentThread().isInterrupted()) {
        long timeout = pingAt - System.currentTimeMillis();
        if (timeout < 0)
          timeout = 0;
        if (ZMQ.poll(pollItems, 2, timeout) == -1)
          break;              //  Interrupted

        //  If we had activity on the pipe, go handle the control
        //  message. Current code never sends control messages.
        if (pollItems[0].isReadable())
          this.controlMsg();

        //  If we had input on the UDP socket, go process that
        if (pollItems[1].isReadable())
          this.handleBeacon();

        //  If we passed the 1-second mark, broadcast our beacon
        if (System.currentTimeMillis() >= pingAt) {
          try {
            udp.send(serializeUUID(this.uuid));
          } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
          }
          pingAt = System.currentTimeMillis() + PING_INTERVAL;
        }
        //  Delete and report any expired peers
        for (Peer peer : peers.values())
          reapPeer(peer);

      }
      try {
        this.destroy();
      } catch (IOException e) {
        e.printStackTrace();
        System.exit(1);
      }
    }
  }

  /**
   * This class defines each peer that we discover and track:
   */
  private static class Peer {
    private static final int PEER_EXPIRY = 5000;  //  Five seconds and it's gone
    private UUID uuid;          //  Peer's UUID as binary blob
    private long expiresAt;

    private Peer(UUID uuid) {
      this.uuid = uuid;
    }

    private void isAlive() {
      expiresAt = System.currentTimeMillis() + PEER_EXPIRY;
    }
  }
}

"""Interface class for Chapter on Distributed Computing

This implements an "interface" to our network of nodes
"""

import time
import uuid
from threading import Thread

import zmq
from zmq.eventloop.ioloop import IOLoop, PeriodicCallback
from zmq.eventloop.zmqstream import ZMQStream

import udplib

# =====================================================================
# Synchronous part, works in our application thread

def pipe(ctx):
    """create an inproc PAIR pipe"""
    a = ctx.socket(zmq.PAIR)
    b = ctx.socket(zmq.PAIR)
    url = "inproc://%s" % uuid.uuid1()
    a.bind(url)
    b.connect(url)
    return a, b

class Interface(object):
    """Interface class.

    Just starts a UDP ping agent in a background thread."""
    ctx = None      # Our context
    pipe = None     # Pipe through to agent

    def __init__(self):
        self.ctx = zmq.Context()
        p0, p1 = pipe(self.ctx)
        self.agent = InterfaceAgent(self.ctx, p1)
        self.agent_thread = Thread(target=self.agent.start)
        self.agent_thread.start()
        self.pipe = p0

    def stop(self):
        self.pipe.close()
        self.agent.stop()
        self.ctx.term()

    def recv(self):
        """receive a message from our interface"""
        return self.pipe.recv_multipart()


# =====================================================================
# Asynchronous part, works in the background

PING_PORT_NUMBER    = 9999
PING_INTERVAL       = 1.0  # Once per second
PEER_EXPIRY         = 5.0  # Five seconds and it's gone
UUID_BYTES          = 32

class Peer(object):

    uuid = None
    expires_at = None

    def __init__(self, uuid):
        self.uuid = uuid
        self.is_alive()

    def is_alive(self):
        """Reset the peers expiry time

        Call this method whenever we get any activity from a peer.
        """
        self.expires_at = time.time() + PEER_EXPIRY


class InterfaceAgent(object):
    """This structure holds the context for our agent so we can
    pass that around cleanly to methods that need it
    """

    ctx = None                 # ZMQ context
    pipe = None                # Pipe back to application
    udp = None                 # UDP object
    uuid = None                # Our UUID as binary blob
    peers = None               # Hash of known peers, fast lookup


    def __init__(self, ctx, pipe, loop=None):
        self.ctx = ctx
        self.pipe = pipe
        if loop is None:
            loop = IOLoop.instance()
        self.loop = loop
        self.udp = udplib.UDP(PING_PORT_NUMBER)
        self.uuid = uuid.uuid4().hex.encode('utf8')
        self.peers = {}

    def stop(self):
        self.pipe.close()
        self.loop.stop()

    def __del__(self):
        try:
            self.stop()
        except:
            pass

    def start(self):
        loop = self.loop
        loop.add_handler(self.udp.handle.fileno(), self.handle_beacon, loop.READ)
        stream = ZMQStream(self.pipe, loop)
        stream.on_recv(self.control_message)
        pc = PeriodicCallback(self.send_ping, PING_INTERVAL * 1000, loop)
        pc.start()
        pc = PeriodicCallback(self.reap_peers, PING_INTERVAL * 1000, loop)
        pc.start()
        loop.start()

    def send_ping(self, *a, **kw):
        try:
            self.udp.send(self.uuid)
        except Exception as e:
            self.loop.stop()

    def control_message(self, event):
        """Here we handle the different control messages from the frontend."""
        print("control message: %s"%event)

    def handle_beacon(self, fd, event):
        uuid = self.udp.recv(UUID_BYTES)
        if uuid in self.peers:
            self.peers[uuid].is_alive()
        else:
            self.peers[uuid] = Peer(uuid)
            self.pipe.send_multipart([b'JOINED', uuid])

    def reap_peers(self):
        now = time.time()
        for peer in list(self.peers.values()):
            if peer.expires_at < now:
                print("reaping %s" % peer.uuid, peer.expires_at, now)
                self.peers.pop(peer.uuid)
                self.pipe.send_multipart([b'LEFT', peer.uuid])

//  Shows how to provoke EAGAIN when reaching HWM

#include <czmq.h>

int main (void) {
    zctx_t *ctx = zctx_new ();
    
    void *mailbox = zsocket_new (ctx, ZMQ_DEALER);
    zsocket_set_sndhwm (mailbox, 4);
    zsocket_set_sndtimeo (mailbox, 0);
    zsocket_connect (mailbox, "tcp://localhost:9876");

    int count;
    for (count = 0; count < 10; count++) {
        printf ("Sending message %d\n", count);
        int rc = zstr_sendf (mailbox, "message %d", count);
        if (rc == -1) {
            printf ("%s\n", strerror (errno));
            break;
        }
    }
    zctx_destroy (&ctx);
    return 0;
}

import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;
import zmq.ZError;

/**
 * Shows how to provoke EAGAIN when reaching HWM
 */
public class Eagain {
  public static void main(String[] args) {
    ZContext ctx = new ZContext();
    ZMQ.Socket mailbox = ctx.createSocket(ZMQ.DEALER);
    mailbox.setSndHWM(4);
    mailbox.setSendTimeOut(0);
    mailbox.connect("tcp://localhost:9876");

    int count;
    for (count = 0; count < 10; count++) {
      System.out.printf("Sending message %d\n", count);
      try {
        mailbox.send(String.format("message %d", count));
        if (mailbox.base().errno() == ZError.EAGAIN) {
          System.out.printf("error code: EAGAIN\n");
          break;
        }
      } catch (ZMQException e) {
        System.out.printf("error code: %s", ZError.toString(e.getErrorCode()));
        break;
      }
    }
    ctx.destroy();
  }
}

"""Shows how to provoke EAGAIN when reaching HWM"""

import zmq

def main():
    ctx = zmq.Context.instance()
    mailbox = ctx.socket(zmq.DEALER)
    mailbox.sndhwm = 4
    mailbox.sndtimeo = 0
    mailbox.connect("tcp://localhost:9876")

    for count in range(10):
        print("Sending message %i" % count)
        try:
            mailbox.send(b"message %i" % count)
        except zmq.Again as e:
            print(e)
            break

    ctx.destroy(linger=0)

if __name__ == '__main__':
    main()