Overview

zmq wraps ZeroMQ — a high-performance asynchronous messaging library. ZeroMQ sockets abstract over TCP, IPC, and in-process transport, with no broker required. It is used in distributed systems, microservices, HPC, and real-time data pipelines.

Messaging Patterns

REQ / REP — RPC

{ === SERVER (REP) === }
uses zmq;
var ctx := NewContext;
var sock := NewSocket(ctx, ZMQ_REP);
Bind(sock, BindAll(5555));
while True do begin
  var msg := Receive(sock);
  WriteLn('Got: ', msg);
  Send(sock, 'Hello from server!');
end;

{ === CLIENT (REQ) === }
uses zmq;
var ctx := NewContext;
var sock := NewSocket(ctx, ZMQ_REQ);
Connect(sock, TCPEndpoint('localhost', 5555));

Send(sock, 'ping');
var reply := Receive(sock);
WriteLn('Reply: ', reply);

CloseSocket(sock);
FreeContext(ctx);

PUB / SUB — Broadcast

{ === PUBLISHER === }
var pub := NewSocket(ctx, ZMQ_PUB);
Bind(pub, BindAll(5556));

// Prefix = topic; subscribers filter on this
Send(pub, 'weather Madrid 22C sunny');
Send(pub, 'sports Barcelona 3-1 Real');

{ === SUBSCRIBER === }
var sub := NewSocket(ctx, ZMQ_SUB);
Connect(sub, TCPEndpoint('localhost', 5556));
Subscribe(sub, 'weather');   // only receive weather messages

while True do
  WriteLn(Receive(sub));

PUSH / PULL — Work Queue

{ === VENTILATOR (PUSH) === }
var push := NewSocket(ctx, ZMQ_PUSH);
Bind(push, BindAll(5557));
for I := 0 to 99 do
  Send(push, 'task:' + IntToStr(I));

{ === WORKER (PULL) === }
var pull := NewSocket(ctx, ZMQ_PULL);
Connect(pull, TCPEndpoint('localhost', 5557));
SetReceiveTimeout(pull, 5000);   // 5 seconds

var task := Receive(pull);
while task <> '' do begin
  ProcessTask(task);
  task := Receive(pull);
end;

Multipart Messages

// Send frames as a single atomic message
SendMultipart(sock, ['envelope', 'metadata', 'payload']);

// Receive all frames
var frames := ReceiveMultipart(sock);
WriteLn('Envelope: ', frames[0]);
WriteLn('Payload:  ', frames[2]);

Non-blocking & Polling

// Poll with timeout — True if message is ready
if Poll(sock, 1000) then
  var msg := Receive(sock)
else
  WriteLn('No message in 1 second');

// Non-blocking receive — empty string if nothing
var msg := ReceiveNoWait(sock);
if msg <> '' then ProcessMsg(msg);