High water mark for PUB/SUB.

/**
 * Publisher.
 */
#include <zmq.h>
#include <chrono>
#include <thread>
#include <cstdio>
#include <cstring>
#include <cassert>

int main (void)
{
    printf("Starting PUB...\n");
    void* context = zmq_ctx_new();
    assert(context && "Failure in context.");
    void* socket = zmq_socket(context, ZMQ_PUB);
    assert(socket && "Failure in socket.");

    {
        auto const rc = zmq_bind(socket, "tcp://*:5555");
        assert(!rc && "Failure in bind.");
    }

    {
        int const hwm = 2;
        int const rc = zmq_setsockopt(socket, ZMQ_SNDHWM, &hwm, sizeof(int));
        assert(!rc && "Failure in setsocket.");
    }

    for (int i = 0; ; ++i) {
        char message[20] = "";
        sprintf(message, "pub_%d", i);
        auto const rc = zmq_send(socket, message, strlen(message), 0);
        assert(rc > 0 && "Failure in send.");
        printf("PUB sent: %s\n", message);
        int const delay = 200;
        std::this_thread::sleep_for(std::chrono::milliseconds(delay));
    }
    zmq_close(socket);
    zmq_ctx_destroy(context);
    return EXIT_SUCCESS;
}

/**
 * Subscriber (client).
 */
#include <zmq.h>
#include <chrono>
#include <thread>
#include <cassert>
#include <cstdlib>

int main(void)
{
    printf("Starting SUB...\n");
    void* context = zmq_ctx_new();
    assert(context && "Failure in context.");
    void* socket = zmq_socket(context, ZMQ_SUB);
    assert(socket && "Failure in socket.");

    {
        int const rc = zmq_connect(socket, "tcp://localhost:5555");
        assert(!rc && "Failure in connect.");
    }

    {
        int const rc = zmq_setsockopt(socket, ZMQ_SUBSCRIBE, "", 0);
        assert(!rc && "Failure in setsocket.");
    }

    {
        int const hwm = 2;
        int const rc = zmq_setsockopt(socket, ZMQ_RCVHWM, &hwm, sizeof(hwm));
        assert(!rc && "Failure in setsocket.");
    }

    while (true) {
        char buffer[256] = "";
        errno = 0;
        int const rc = zmq_recv(socket, buffer, sizeof(buffer), 0);
        assert(rc > 0 && "Failure in recv.");
        printf("SUB received: %s\n", buffer);
        int const delay = 1;
        std::this_thread::sleep_for(std::chrono::seconds(delay));
    }
    zmq_close(socket);
    zmq_ctx_destroy(context);
    return EXIT_SUCCESS;
}