ZeroMQ

Content:

  1. Introduction
  2. What is ZeroMQ
  3. How to use ZeroMQ
  4. Libraries
  5. Request-Reply
  6. Push-Pull
  7. Publisher-Subscriber
  8. Stream (plain socket)
  9. Conclusion

Introduction:

The purpose of this article is to give an introduction on how to use ZeroMQ with a little background information and some code examples.

ZeroMQ is not a traditional Message Queue like IBM MQSeries, Apache ActiveMQ, RabbitMQ etc..

Disclaimer: I am not an experienced ZeroMQ user, so I may have missed some things.

What is ZeroMQ:

Depending on point of view one can consider ZeroMQ one of:

I am most inclined towards the second view.

ZeroMQ project themselves say:

ZeroMQ (also known as ØMQ, 0MQ, or zmq) looks like an embeddable networking library but acts like a concurrency framework. It gives you sockets that carry atomic messages across various transports like in-process, inter-process, TCP, and multicast. You can connect sockets N-to-N with patterns like fan-out, pub-sub, task distribution, and request-reply. It's fast enough to be the fabric for clustered products. Its asynchronous I/O model gives you scalable multicore applications, built as asynchronous message-processing tasks. It has a score of language APIs and runs on most operating systems.

How to use ZeroMQ:

The main concepts in ZeroMQ are:

ZeroMQ support multiple transports;

tcp
TCP socket
inproc
intern within process using memory
ipc
IPC using Unix domain sockets [not available on all OS]
pgm
PGM datagrams [not available on all OS]
epgm
encapsulated PGM within UDP datagrams [not available on all OS]

The ZeroMQ API is transport agnostic - the transport is specified in the endpoint name. Endpoint names follow the syntax: 

transport://name

Example:

All the examples shown will use tcp transport.

ZeroMQ are designed for specific message patterns that requires different socket types (note that socket type here refers to ZeroMQ sockets and include all transports not just to TCP sockets):

Message pattern socket type one end socket type other end
Request-Reply REQ [sync]
DEALER [async]		 REP [sync]
ROUTER [async]
Publish-Subscribe PUB
XPUB		 SUB
XSUB
Pipeline PUSH PULL
Client-Server [draft] CLIENT SERVER
Radio-Dish [draft] RADIO DISH
Integration with non ZeroMQ STREAM N/A

Note that the message pattern socket type relationship is enforced by the software. Trying to do a message pattern with socket types intended for a different message pattern will not work.

Internally ZeroMQ use a protocol ZMTP (ZeroMQ Message Transport Protocol, but that is encapsulated by the libraries.

Libraries:

ZeroMQ is usable from many programming languages and for some of them there are multiple libraries to choose from:

Above list is not complete. Check ZeroMQ documentation for complete list.

Request-Reply:

This is the message pattern where the request side does send-recv-send-recv-send-recv-... and the reply side does recv-send-recv-send-recv-send-...:

Request Reply model

This is the equivalent of a web service call.

Client: 

#include <stdio.h>

#include <czmq.h>

void test(const char *addr)
{   
    zsock_t *s;
    char *reqmsg;
    char *respmsg;
    s = zsock_new_req(addr);
    reqmsg = "C (high) client";
    zstr_send(s, reqmsg);
    respmsg = zstr_recv(s);
    printf("%s\n", respmsg);
    zstr_free(&respmsg);
    zsock_destroy(&s);
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

#include <stdio.h>
#include <string.h>

#include <zmq.h>

void test(const char *addr)
{   
    void *ctx;
    void *s;
    char *reqmsg;
    char respmsg[10000];
    int msglen;
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_REQ);;
    zmq_connect(s, addr);
    reqmsg = "C (low) client";
    zmq_send(s, reqmsg, strlen(reqmsg), 0);
    msglen = zmq_recv(s, respmsg, sizeof(respmsg), 0);
    respmsg[msglen] = 0;
    printf("%s\n", respmsg);
    zmq_close(s);
    zmq_ctx_destroy(ctx);
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

#include <iostream>
#include <string>

using namespace std;

#include <zmq.hpp>

using namespace zmq;

void test(string addr)
{
    context_t ctx;
    socket_t s(ctx, socket_type::req);
    s.connect(addr);
    string reqmsg = "C++ client";
    s.send(buffer(reqmsg), send_flags::none);
    message_t temp;
    s.recv(temp, recv_flags::none);
    string respmsg = temp.to_string();
    cout << respmsg << endl;
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

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

public class Cli4
{
    private static void test(String addr) {
        try (ZContext ctx = new ZContext()) {
            ZMQ.Socket s = ctx.createSocket(SocketType.REQ);
            s.connect(addr);
            String reqmsg = "Java client";
            s.send(reqmsg);
            String respmsg = s.recvStr();
            System.out.println(respmsg);
            s.close();
        }
    }
    public static void main(String[] args) throws Exception
    {
        test("tcp://localhost:10001");
        test("tcp://localhost:10002");
        test("tcp://localhost:10003");
        test("tcp://localhost:10004");
        test("tcp://localhost:10005");
        test("tcp://localhost:10006");
    }
}

import zmq

def test(addr):
    ctx = zmq.Context()
    s = ctx.socket(zmq.REQ)
    s.connect(addr)
    reqmsg = "Python client"
    s.send_string(reqmsg)
    respmsg = s.recv_string()
    print(respmsg)

test("tcp://localhost:10001")
test("tcp://localhost:10002")
test("tcp://localhost:10003")
test("tcp://localhost:10004")
test("tcp://localhost:10005")
test("tcp://localhost:10006")

using System;

using NetMQ;
using NetMQ.Sockets;

public class Cli
{
    private static void Test(string addr)
    {
        using (RequestSocket s = new RequestSocket())
        {
            s.Connect(addr);
            string reqmsg = "C# Client";
            s.SendFrame(reqmsg);
            string respmsg = s.ReceiveFrameString();
            Console.WriteLine(respmsg);
        }
    }
    public static void Main(string[] args)
    {
        Test("tcp://localhost:10001");
        Test("tcp://localhost:10002");
        Test("tcp://localhost:10003");
        Test("tcp://localhost:10004");
        Test("tcp://localhost:10005");
        Test("tcp://localhost:10006");
    }
}

Server: 

#include <string.h>

#include <czmq.h>

int main(void)
{
    zsock_t *s;
    char *reqmsg;
    char respmsg[10000];
    s = zsock_new_rep("tcp://*:10001");
    for(;;)
    {
        reqmsg = zstr_recv(s);
        strcpy(respmsg, reqmsg);
        strcat(respmsg, " - C (high) server");
        zstr_free(&reqmsg);
        zstr_send(s, respmsg);
    }
    return 0;
}

#include <string.h>

#include <zmq.h>

int main(void)
{
    void *ctx;
    void *s;
    char reqmsg[10000];
    char respmsg[10000];
    int msglen;
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_REP);
    zmq_bind(s, "tcp://*:10002");
    for(;;)
    {
        msglen = zmq_recv(s, reqmsg, sizeof(reqmsg), 0);
        reqmsg[msglen] = 0;
        strcpy(respmsg, reqmsg);
        strcat(respmsg, " - C (low) server");
        zmq_send(s, respmsg, strlen(respmsg), 0);
    }
    return 0;
}

#include <string>

using namespace std;

#include <zmq.hpp>

using namespace zmq;

int main(void)
{
    context_t ctx;
    socket_t s(ctx, socket_type::rep);
    s.bind("tcp://*:10003");
    for(;;)
    {
        message_t temp;
        s.recv(temp, recv_flags::none);
        string reqmsg = temp.to_string();
        string respmsg = reqmsg + " - C++ server";
        s.send(buffer(respmsg), send_flags::none);
    }
    return 0;
}

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

public class Srv4
{
    public static void main(String[] args) throws Exception
    {
        try (ZContext ctx = new ZContext()) {
            ZMQ.Socket s = ctx.createSocket(SocketType.REP);
            s.bind("tcp://*:10004");
            while(true) {
                String reqmsg = s.recvStr();
                String respmsg = reqmsg + " - Java server";
                s.send(respmsg);
            }
        }
    }
}

import zmq

ctx = zmq.Context()
s = ctx.socket(zmq.REP)
s.bind("tcp://*:10005")
while True:
    reqmsg = s.recv_string()
    respmsg = reqmsg + " - Python server"
    s.send_string(respmsg)

using System;

using NetMQ;
using NetMQ.Sockets;

public class Srv6
{
    public static void Main(string[] args)
    {
        using (ResponseSocket s = new ResponseSocket())
        {
            s.Bind("tcp://*:10006");
            while(true)
            {
                string reqmsg = s.ReceiveFrameString();
                string respmsg = reqmsg + " - C# server";
                s.SendFrame(respmsg);
            }
        }
    }
}

Push-Pull:

This is the message pattern where the push side does send-send-send-send-send-... and the pull side does recv-recv-recv-recv-recv-...:

Push Pull model

This is the equivalent of a traditional message queue send to queue and receive from queue.

Client: 

#include <czmq.h>

void test(const char *addr)
{   
    zsock_t *s;
    char *reqmsg;
    int i;
    s = zsock_new_push(addr);
    for(i = 0; i < 3; i++)
    {
        reqmsg = "C (high) client";
        zstr_send(s, reqmsg);
    }
    zmq_sleep(1);
    zsock_destroy(&s);
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

#include <string.h>

#include <zmq.h>

void test(const char *addr)
{   
    void *ctx;
    void *s;
    char *reqmsg;
    int i;
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_PUSH);;
    zmq_connect(s, addr);
    for(i = 0; i < 3; i++)
    {
        reqmsg = "C (low) client";
        zmq_send(s, reqmsg, strlen(reqmsg), 0);
    }
    zmq_sleep(1);
    zmq_close(s);
    zmq_ctx_destroy(ctx);
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

#include <iostream>
#include <string>

using namespace std;

#include <zmq.hpp>

using namespace zmq;

void test(string addr)
{
    context_t ctx;
    socket_t s(ctx, socket_type::push);
    s.connect(addr);
    for(int i = 0; i < 3; i++) 
    {
        string reqmsg = "C++ client";
        s.send(buffer(reqmsg), send_flags::none);
    }
    zmq_sleep(1);
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

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

public class Cli4
{
    private static void test(String addr) {
        try (ZContext ctx = new ZContext()) {
            ZMQ.Socket s = ctx.createSocket(SocketType.PUSH);
            s.connect(addr);
            for(int i = 0; i < 3; i++) {
                String reqmsg = "Java client";
                s.send(reqmsg);
            }
            try {
                Thread.sleep(1000);
            } catch(InterruptedException ex) {
            }
            s.close();
        }
    }
    public static void main(String[] args) throws Exception
    {
        test("tcp://localhost:10001");
        test("tcp://localhost:10002");
        test("tcp://localhost:10003");
        test("tcp://localhost:10004");
        test("tcp://localhost:10005");
        test("tcp://localhost:10006");
    }
}

import time

import zmq

def test(addr):
    ctx = zmq.Context()
    s = ctx.socket(zmq.PUSH)
    s.connect(addr)
    for i in range(3):
        reqmsg = "Python client"
        s.send_string(reqmsg)
    time.sleep(1)

test("tcp://localhost:10001")
test("tcp://localhost:10002")
test("tcp://localhost:10003")
test("tcp://localhost:10004")
test("tcp://localhost:10005")
test("tcp://localhost:10006")

using System;
using System.Threading;

using NetMQ;
using NetMQ.Sockets;

public class Cli
{
    private static void Test(string addr)
    {
        using (PushSocket s = new PushSocket())
        {
            s.Connect(addr);
            for(int i = 0; i < 3; i++)
            {
                string reqmsg = "C# Client";
                s.SendFrame(reqmsg);
            }
            Thread.Sleep(1000);
        }
    }
    public static void Main(string[] args)
    {
        Test("tcp://localhost:10001");
        Test("tcp://localhost:10002");
        Test("tcp://localhost:10003");
        Test("tcp://localhost:10004");
        Test("tcp://localhost:10005");
        Test("tcp://localhost:10006");
    }
}

Server: 

#include <stdio.h>

#include <czmq.h>

int main(void)
{
    zsock_t *s;
    char *reqmsg;
    s = zsock_new_pull("tcp://*:10001");
    for(;;)
    {
        reqmsg = zstr_recv(s);
        printf("C (high) server : %s\n", reqmsg);
        zstr_free(&reqmsg);
    }
    return 0;
}

#include <stdio.h>

#include <zmq.h>

int main(void)
{
    void *ctx;
    void *s;
    char reqmsg[10000];
    int msglen;
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_PULL);
    zmq_bind(s, "tcp://*:10002");
    for(;;)
    {
        msglen = zmq_recv(s, reqmsg, sizeof(reqmsg), 0);
        reqmsg[msglen] = 0;
        printf("C (low) server : %s\n", reqmsg);
    }
    return 0;
}

#include <iostream>
#include <string>

using namespace std;

#include <zmq.hpp>

using namespace zmq;

int main(void)
{
    context_t ctx;
    socket_t s(ctx, socket_type::pull);
    s.bind("tcp://*:10003");
    for(;;)
    {
        message_t temp;
        s.recv(temp, recv_flags::none);
        string reqmsg = temp.to_string();
        cout << "C++ server : " << reqmsg << endl;
    }
    return 0;
}

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

public class Srv4
{
    public static void main(String[] args) throws Exception
    {
        try (ZContext ctx = new ZContext()) {
            ZMQ.Socket s = ctx.createSocket(SocketType.PULL);
            s.bind("tcp://*:10004");
            while(true) {
                String reqmsg = s.recvStr();
                System.out.println("Java server : " + reqmsg);
            }
        }
    }
}

import zmq

ctx = zmq.Context()
s = ctx.socket(zmq.PULL)
s.bind("tcp://*:10005")
while True:
    reqmsg = s.recv_string()
    print("Python server : " + reqmsg)

using System;

using NetMQ;
using NetMQ.Sockets;

public class Srv6
{
    public static void Main(string[] args)
    {
        using (PullSocket s = new PullSocket())
        {
            s.Bind("tcp://*:10006");
            while(true)
            {
                string reqmsg = s.ReceiveFrameString();
                Console.WriteLine("C# server : " + reqmsg);
            }
        }
    }
}

Publisher-Subscriber:

This is the message pattern where one publisher side does send-send-send-... and multiple subscriber sides does recv-rev-recv-...:

Publisher Subscriber model

This is the equivalent of a traditional message queue publish to topic and subscribe to topic.

Client: 

#include <stdio.h>
#include <string.h>

#include <czmq.h>

#define TOPIC "1234"

void test(const char *addr)
{   
    zsock_t *s;
    char *respmsg, *respmsg2;
    int n;
    s = zsock_new_sub(addr, "");
    zsock_set_subscribe(s, TOPIC);
    n = 0;
    while(n < 3)
    {
        respmsg = zstr_recv(s);
        respmsg2 = respmsg + strlen(TOPIC) + 1;
        printf("C (high) client : %s\n", respmsg2);
        n++;
    }
    zstr_free(&respmsg);
    zsock_destroy(&s);
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

#include <stdio.h>
#include <string.h>

#include <zmq.h>

#define TOPIC "1234"

void test(const char *addr)
{   
    void *ctx;
    void *s;
    char respmsg[10000], *respmsg2;
    int n, msglen;
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_SUB);;
    zmq_connect(s, addr);
    zmq_setsockopt(s, ZMQ_SUBSCRIBE, TOPIC, 0);
    n = 0;
    while(n < 3)
    {
        msglen = zmq_recv(s, respmsg, sizeof(respmsg), 0);
        respmsg[msglen] = 0;
        respmsg2 = respmsg + strlen(TOPIC) + 1;
        printf("C (low) client : %s\n", respmsg2);
        n++;
    }
    zmq_close(s);
    zmq_ctx_destroy(ctx);
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

#include <iostream>
#include <string>

using namespace std;

#include <zmq.hpp>

using namespace zmq;

const string TOPIC = "1234";

void test(string addr)
{
    context_t ctx;
    socket_t s(ctx, socket_type::sub);
    s.connect(addr);
    s.set(sockopt::subscribe, TOPIC);
    int n = 0;
    while(n < 3) 
    {
        message_t temp;
        s.recv(temp, recv_flags::none);
        string respmsg = temp.to_string();
        respmsg = respmsg.substr(TOPIC.size() + 1);
        cout << "C++ client : " << respmsg << endl;
        n++;
    }
}

int main(void)
{
    test("tcp://localhost:10001");
    test("tcp://localhost:10002");
    test("tcp://localhost:10003");
    test("tcp://localhost:10004");
    test("tcp://localhost:10005");
    test("tcp://localhost:10006");
    return 0;
}

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

public class Cli4
{
    private static final String TOPIC = "1234";
    private static void test(String addr) {
        try (ZContext ctx = new ZContext()) {
            ZMQ.Socket s = ctx.createSocket(SocketType.SUB);
            s.connect(addr);
            s.subscribe(TOPIC);
            int n = 0;
            while(n < 3) {
                String respmsg = s.recvStr();
                respmsg = respmsg.substring(TOPIC.length() + 1);
                System.out.println("Java client : " + respmsg);
                n++;
            }
            s.close();
        }
    }
    public static void main(String[] args) throws Exception
    {
        test("tcp://localhost:10001");
        test("tcp://localhost:10002");
        test("tcp://localhost:10003");
        test("tcp://localhost:10004");
        test("tcp://localhost:10005");
        test("tcp://localhost:10006");
    }
}

import zmq

TOPIC = '1234'

def test(addr):
    ctx = zmq.Context()
    s = ctx.socket(zmq.SUB)
    s.connect(addr)
    s.subscribe(TOPIC)
    for i in range(3):
        respmsg = s.recv_string()
        respmsg = respmsg[len(TOPIC):]
        print('Python client : ' + respmsg)

test("tcp://localhost:10001")
test("tcp://localhost:10002")
test("tcp://localhost:10003")
test("tcp://localhost:10004")
test("tcp://localhost:10005")
test("tcp://localhost:10006")

using System;

using NetMQ;
using NetMQ.Sockets;

public class Cli
{
    private const string TOPIC = "1234";
    private static void Test(string addr)
    {
        using (SubscriberSocket s = new SubscriberSocket())
        {
            s.Connect(addr);
            s.Subscribe(TOPIC);
            for(int i = 0; i < 3; i++) {
                string respmsg = s.ReceiveFrameString();
                respmsg = respmsg.Substring(TOPIC.Length + 1);
                Console.WriteLine("C# Client : " + respmsg);
            }
        }
    }
    public static void Main(string[] args)
    {
        Test("tcp://localhost:10001");
        Test("tcp://localhost:10002");
        Test("tcp://localhost:10003");
        Test("tcp://localhost:10004");
        Test("tcp://localhost:10005");
        Test("tcp://localhost:10006");
    }
}

Server: 

#include <stdio.h>

#include <czmq.h>

#define TOPIC "1234"

int main(void)
{
    zsock_t *s;
    char respmsg[10000];
    int count;
    s = zsock_new_pub("tcp://*:10001");
    count = 0;
    for(;;)
    {
        count++;
        sprintf(respmsg, "%s C (high) server (%d)", TOPIC, count);
        zstr_send(s, respmsg);
        zmq_sleep(3);
    }
    return 0;
}

#include <stdio.h>
#include <string.h>

#include <zmq.h>

#define TOPIC "1234"

int main(void)
{
    void *ctx;
    void *s;
    char respmsg[10000];
    int count;
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_PUB);
    zmq_bind(s, "tcp://*:10002");
    count = 0;
    for(;;)
    {
        count++;
        sprintf(respmsg, "%s C (low) server (%d)", TOPIC, count);
        zmq_send(s, respmsg, strlen(respmsg), 0);
        zmq_sleep(3);
    }
    return 0;
}

#include <sstream>
#include <string>

using namespace std;

#include <zmq.hpp>

using namespace zmq;

const string TOPIC = "1234";

int main(void)
{
    context_t ctx;
    socket_t s(ctx, socket_type::pub);
    s.bind("tcp://*:10003");
    int count = 0;
    for(;;)
    {
        count++;
        stringstream temp;
        temp << TOPIC << " C++ server (" << count << ")";
        string respmsg = temp.str();
        s.send(buffer(respmsg), send_flags::none);
        zmq_sleep(3);
    }
    return 0;
}

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

public class Srv4
{
    private static final String TOPIC = "1234";
    public static void main(String[] args) throws Exception
    {
        try (ZContext ctx = new ZContext()) {
            ZMQ.Socket s = ctx.createSocket(SocketType.PUB);
            s.bind("tcp://*:10004");
            int count = 0;
            while(true) {
                count++;
                String respmsg = String.format("%s Java server (%d)", TOPIC, count);
                s.send(respmsg);
                try {
                   Thread.sleep(3000);
                } catch(InterruptedException ex) {
                }
            }
        }
    }
}

import time

import zmq

TOPIC = '1234'

ctx = zmq.Context()
s = ctx.socket(zmq.PUB)
s.bind("tcp://*:10005")
count = 0
while True:
    count = count + 1
    respmsg = "%s Python server (%d)" % (TOPIC, count)
    s.send_string(respmsg)
    time.sleep(3)

using System;
using System.Threading;

using NetMQ;
using NetMQ.Sockets;

public class Srv6
{
    private const string TOPIC = "1234";
    public static void Main(string[] args)
    {
        using (PublisherSocket s = new PublisherSocket())
        {
            s.Bind("tcp://*:10006");
            int count = 0;
            while(true)
            {
                count++;
                string respmsg = String.Format("{0} C# server ({1})", TOPIC, count);
                s.SendFrame(respmsg);
                Thread.Sleep(3000);
            }
        }
    }
}

Stream (plain socket):

ZeroMQ stream mode is a plain socket compatibility mode allowing ZeroMQ code to communicate with programs using plain sockets.

I only see the point of that for C, because the C socket API is a bit cumbersome.

But I don't even like ZeroMQ stream mode for that. Using ZeroMQ stream mode may require fewer lines than using socket API, but writing those lines becomes a guessing exercise, because the usage is rather obscure. IMHO - YMMV.

To demo I will implement the C code from:

using ZeroMQ in stream mode.

Utility functions (wrapzmq.c): 

void wrap_send(void *s, char *id, int idlen, char *buf, int buflen)
{
    zmq_send(s, id, idlen, ZMQ_SNDMORE); // send id
    int z = zmq_send(s, buf, buflen, 0);
}

typedef int (*moref)(char *buf, int n);

void wrap_recv_gen(void *s, char *buf, int buflen, moref more)
{
    char temp[256];
    int n = 0;
    while(n < buflen && more(buf, n))
    {
        while(zmq_recv(s, temp, sizeof(temp), 0) == 0); // recv id
        n += zmq_recv(s, buf + n, buflen - n, 0);
#ifdef DEBUG
        for(int i = 0; i < n; i++) printf(">%d : %d\n", i, buf[i]);
#endif
    }
}

void wrap_recv_gen_with_id(void *s, char *buf, int buflen, char *id, size_t *idlen, moref more)
{
    int n = 0;
    while(n < buflen && more(buf, n))
    {
        while((*idlen = zmq_recv(s, id, 256, 0)) == 0); // recv id
        n += zmq_recv(s, buf + n, buflen - n, 0);                     
#ifdef DEBUG
        for(int i = 0; i < n; i++) printf(">%d : %d\n", i, buf[i]);
#endif
    }
}

static int always(char *buf, int n)
{
    return 1;
}

void wrap_recv(void *s, char *buf, int buflen)
{
#ifdef DEBUG
    printf("recv %d bytes:\n", buflen);
#endif
    wrap_recv_gen(s, buf, buflen, always);
}

void wrap_recv_with_id(void *s, char *buf, int buflen, char *id, size_t *idlen)
{
#ifdef DEBUG
    printf("recv %d bytes:\n", buflen);
#endif
    wrap_recv_gen_with_id(s, buf, buflen, id, idlen, always);
}

static int notline(char *buf, int n)
{
    return n < 2 || buf[n-1] != '\n' || buf[n-2] != '\r';
}
    
void wrap_recv_line(void *s, char *buf, int buflen)
{
#ifdef DEBUG
    printf("recv line:\n");
#endif
    wrap_recv_gen(s, buf, buflen, notline);
}

void wrap_recv_line_with_id(void *s, char *buf, int buflen, char *id, size_t *idlen)
{
#ifdef DEBUG
    printf("recv line:\n");
#endif
    wrap_recv_gen_with_id(s, buf, buflen, id, idlen, notline);
}

static int notstr(char *buf, int n)
{
    return n < 1 || n < buf[0]; 
}

void wrap_recv_count(void *s, char *buf, int buflen)
{
#ifdef DEBUG
    printf("recv count:\n");
#endif
    wrap_recv_gen(s, buf, buflen, notstr);
}

void wrap_recv_count_with_id(void *s, char *buf, int buflen, char *id, size_t *idlen)
{
#ifdef DEBUG
    printf("recv count:\n");
#endif
    wrap_recv_gen_with_id(s, buf, buflen, id, idlen, notstr);
}

Client (binary protocol): 

#include <stdio.h>
#include <string.h>

#include <zmq.h>

#include "wrapzmq.c"

#pragma pack(push, 1)
typedef struct
{
    unsigned char len;
    char buf[256];
} cbuf;
#pragma pack(pop)

int main(void)
{
    void *ctx;
    void *s;
    char id[256];
    size_t idlen;
    long int a, b, c, tmp, tmp2[2];
    cbuf v, v2;
    /* create socket */
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_STREAM);
    /* connect to host */
    zmq_connect(s, "tcp://127.0.0.1:12345");
    idlen = sizeof(id);
    zmq_getsockopt(s, ZMQ_IDENTITY, id, &idlen);
    wrap_recv(s, 0, 0); // skip empty connect message
    /* send a and b */
    a = 123;
    b = 456;
    tmp2[0] = htonl(a);
    tmp2[1] = htonl(b);
    wrap_send(s, id, idlen, (char *)&tmp2[0], sizeof(tmp2));
    /* receive c and print */
    wrap_recv(s, (char *)&tmp, sizeof(tmp));
    c = ntohl(tmp);
    printf("%ld\n", c);
    /* send v */
    v.len = strlen("ABC");
    strcpy(v.buf, "ABC");
    wrap_send(s, id, idlen, (char *)&v, 1 + v.len);
    /* receive v2 and print */
    wrap_recv_count(s, (char *)&v2, sizeof(v2));
    v2.buf[v2.len] = 0;
    printf("%s\n", v2.buf);
    /* close socket */
    zmq_close(s);
    zmq_ctx_destroy(ctx);
    return 0;
}

Client (text protocol): 

#include <stdio.h>
#include <string.h>

#include <zmq.h>

#include "wrapzmq.c"

int main(void)
{
    void *ctx;
    void *s;
    char id[256];
    size_t idlen;
    long int a, b, c;
    char *v, v2[10000];
    char buf[10000];
    /* create socket */
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_STREAM);
    /* connect to host */
    zmq_connect(s, "tcp://127.0.0.1:12345");
    idlen = sizeof(id);
    zmq_getsockopt(s, ZMQ_IDENTITY, id, &idlen);
    wrap_recv(s, buf, 0); // skip empty connect message
    /* send a and b */
    a = 123;
    b = 456;
    sprintf(buf, "%ld %ld\r\n", a, b);
    wrap_send(s, id, idlen, buf, strlen(buf));
    /* receive c and print */
    wrap_recv_line(s, buf, sizeof(buf));
    sscanf(buf, "%ld\r\n", &c);
    printf("%ld\n", c);
    /* send v */
    v = "ABC";
    sprintf(buf, "%s\r\n", v);
    wrap_send(s, id, idlen, buf, strlen(buf));
    /* receive v2 and print */
    wrap_recv_line(s, buf, sizeof(buf));
    sscanf(buf, "%s\r\n", v2);
    printf("%s\n", v2);
    /* close socket */
    zmq_close(s);
    zmq_ctx_destroy(ctx);
    return 0;
}

Server (binary protocol): 

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <zmq.h>

#include "wrapzmq.c"

#pragma pack(push, 1)
typedef struct
{
    unsigned char len;
    char buf[256];
} cbuf;
#pragma pack(pop)

int main(void)
{
    void *ctx;
    void *s;
    char id[256];
    size_t idlen;
    long int a, b, c, tmp, tmp2[2];
    cbuf v, v2;
    /* create socket */
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_STREAM);
    /* bind */
    zmq_bind(s, "tcp://127.0.0.1:12345");
    idlen = sizeof(id);
    zmq_getsockopt(s, ZMQ_IDENTITY, id, &idlen);
    for(;;)
    {
        wrap_recv(s, 0, 0); // skip empty accept message
        /* read a and b */
        wrap_recv_with_id(s, (char *)&tmp2, sizeof(tmp2), id, &idlen);
        a = ntohl(tmp2[0]);
        b = ntohl(tmp2[1]);
        /* calculate c and send */
        c = a + b;
        tmp = htonl(c);
        wrap_send(s, id, idlen, (char *)&tmp, sizeof(tmp));
        /* read v */
        wrap_recv_count_with_id(s, (char *)&v, sizeof(v), id, &idlen);
        v.buf[v.len] = 0;
        /* calculate v2 and send */
        strcpy(v2.buf, v.buf);
        strcat(v2.buf, v.buf);
        v2.len = strlen(v2.buf);
        wrap_send(s, id, idlen, (char *)&v2, 1 + v2.len);
        wrap_send(s, id, idlen, 0, 0); // disconnect
    }
    /* close socket */
    zmq_close(s);
    zmq_ctx_destroy(ctx);
    return 0;
}

Server (text protocol): 

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <zmq.h>

#include "wrapzmq.c"

int main(void)
{
    void *ctx;
    void *s;
    char id[256];
    size_t idlen;
    long int a, b, c;
    char *buf, *v, *v2;
    int buflen;
    /* create socket */
    ctx = zmq_ctx_new();
    s = zmq_socket(ctx, ZMQ_STREAM);
    /* bind */
    zmq_bind(s, "tcp://127.0.0.1:12345");
    idlen = sizeof(id);
    zmq_getsockopt(s, ZMQ_IDENTITY, id, &idlen);
    for(;;)
    {
        wrap_recv(s, 0, 0); // skip empty accept message
        /* read a and b */
        buflen = 24;
        buf = malloc(buflen);
        wrap_recv_line_with_id(s, buf, buflen, id, &idlen);
        sscanf(buf, "%ld %ld\r\n", &a, &b);
        free(buf);
        /* calculate c and send */
        c = a + b;
        buflen = 13;
        buf = malloc(buflen);
        sprintf(buf, "%ld\r\n", c);
        wrap_send(s, id, idlen, buf, strlen(buf));
        free(buf);
        /* read v */
        buflen = 258;
        buf = malloc(buflen);
        wrap_recv_line_with_id(s, buf, buflen, id, &idlen);
        v = malloc(strlen(buf) + 1);
        sscanf(buf, "%s\r\n", v);
        free(buf);
        /* calculate v2 and send */
        v2 = malloc(2 * strlen(v) + 1);
        strcpy(v2, v);
        strcat(v2, v);
        buflen = strlen(v2) + 3;
        buf = malloc(buflen);
        sprintf(buf, "%s\r\n", v2);
        wrap_send(s, id, idlen, buf, strlen(buf));
        wrap_send(s, id, idlen, 0, 0); // disconnect
        free(buf);
        free(v);
        free(v2);
    }
    /* close socket */
    zmq_close(s);
    zmq_ctx_destroy(ctx);
    return 0;
}

Note that a ZeroMQ stream mode server is really a multiplexed server via the use of id, but to be able to serve multiple concurrent requests, then the application logic and protocol need to support it - my examples does not and only works without concurrent requests.

Conclusion:

ZeroMQ works great for what is intended for. And the library API's are easy to use.

But one need to understand how to use it correct because otherwise it won't work and it can be very difficult to troubleshoot.

So read some documentationa and view some examples before trying to get some code working.

Article history:

Version Date Description
1.0 July 16th 2023 Initial version
1.1 August 29th 2026 Add stream mode section

Other articles:

See list of all articles here

Comments:

Please send comments to Arne Vajhøj