This chapter's contract-owning documents — the PUB, SUB, XPUB, and XSUB specs own the contract. This chapter shows that contract through language examples.

1. Overview¶

The Publish-Subscribe pattern distributes messages based on topics. zlink provides two levels: basic PUB/SUB and advanced XPUB/XSUB.

Socket	Role	Characteristics
PUB	Publisher	Broadcasts to all subscribers. Cannot receive.
SUB	Subscriber	Topic prefix match filtering. Cannot send.
XPUB	Advanced Publisher	PUB + can receive subscription events
XSUB	Advanced Subscriber	Receives all messages without local filtering

Valid socket combinations: - PUB → SUB, PUB → XSUB - XPUB → SUB, XPUB → XSUB

SUB vs XSUB — Key Difference¶

Both SUB and XSUB send subscription info to the upstream PUB via zlink_set_subscription(). The public API usage is identical. The difference is whether the local filter engine is on or off.

	SUB (filtered)	XSUB (unfiltered)
With subscriptions	Receives only matching messages	Receives all messages (no filter check)
No subscriptions	Receives nothing	Receives all messages
`""` empty subscription	Receives all (matches every topic)	Already receives all without subscribing
Use case	Normal subscriber	Proxy/relay (pass-through)

Common confusion: "If I subscribe SUB with "", isn't it the same as XSUB?" → Both receive all messages in practice, but SUB still evaluates its local byte-prefix filter on every message while XSUB never applies one. Also, SUB with no subscriptions receives nothing, while XSUB with no subscriptions still receives everything.

Why XSUB/XPUB in the proxy pattern:

%%{init: {'flowchart': {'nodeSpacing': 32, 'rankSpacing': 40, 'padding': 8, 'wrappingWidth': 180}, 'themeVariables': {'fontSize': '18px'}}}%%
flowchart LR
  PUB -- data --> XSUB
  XSUB == proxy ==> XPUB
  XPUB -- data --> SUB
  SUB -. subscribe .-> XPUB
  XPUB -. propagate .-> XSUB

XSUB passes all messages from PUB without applying its own filter.
XPUB exposes SUB subscription events via zlink_xpub_recv(), allowing the proxy to inject subscription management logic (filtering, logging, authorization, etc.).
Plain SUB/PUB cannot build this relay structure.

%%{init: {'flowchart': {'nodeSpacing': 32, 'rankSpacing': 40, 'padding': 8, 'wrappingWidth': 180}, 'themeVariables': {'fontSize': '18px'}}}%%
flowchart LR
  PUB --> SUB1["SUB 1 (weather)"]
  PUB --> SUB2["SUB 2 (sports)"]

Part I: PUB/SUB¶

2. PUB/SUB Basic Usage¶

Publisher (PUB)¶

zlink_publish() submits the complete payload-part array as one message record from a PUB or XPUB socket.

void *pub = zlink_socket(ctx, ZLINK_SOCKET_PUB);
zlink_bind(pub, "tcp://*:5556");

/* Publish message -- dropped for a subscriber whose queue is at HWM */
zlink_msg_t part;
zlink_msg_init_size(&part, 14);
memcpy(zlink_msg_data(&part), "weather: sunny", 14);
zlink_submit_result_t rc = zlink_publish(
    pub, "weather", &part, 1, ZLINK_SEND_FLAGS_NONE);

Subscriber (SUB)¶

zlink_subscribe() receives a topic and its complete payload record in one call. There is no callback surface; the intended pattern is to observe ZLINK_POLLIN from a poller and pull with this function.

void *sub = zlink_socket(ctx, ZLINK_SOCKET_SUB);
zlink_connect(sub, "tcp://127.0.0.1:5556");

/* Subscribe to topic -- set after connect */
zlink_set_subscription(sub, "weather");

char topic[256];
size_t topic_len = 0;
zlink_msg_t parts[8];
size_t part_count = 0;
zlink_recv_result_t rc = zlink_subscribe(
    sub, NULL, topic, sizeof(topic), &topic_len,
    parts, 8, &part_count, ZLINK_RECV_FLAGS_NONE);
if (rc == ZLINK_RECV_OK) {
    printf("Topic: %.*s, Data: %.*s\n",
           (int)topic_len, topic,
           (int)zlink_msg_size(&parts[0]), (char *)zlink_msg_data(&parts[0]));
    zlink_multipart_close(parts, part_count);
}
/* other rc values: ZLINK_RECV_NO_DATA (EAGAIN), TERMINATED, BUFFER_TOO_SMALL */

Reference: core/tests/integration/test_pubsub.cpp -- empty subscription ("") → receives all messages

Sending and Receiving Summary¶

Socket	Direction	Send/Receive API	Notes
PUB	Send only	`zlink_publish()`	Cannot receive
SUB	Receive only	`zlink_subscribe()`	Topic + payload array returned separately
XPUB	Bidirectional	`zlink_publish()` / `zlink_xpub_recv()`	Receives subscription events
XSUB	Receive only	`zlink_subscribe()`	No local filter; receives all

Note: zlink_send() / zlink_recv() return ZLINK_SUBMIT_NOT_SUPPORTED / ZLINK_RECV_NOT_SUPPORTED on all 4 PUB/SUB sockets. Use zlink_publish() for publishing and zlink_subscribe() for receiving.

PUB / XPUB default: ZLINK_PUB_OPT_NODROP defaults to 0. When the HWM is reached, the message for that subscriber is silently dropped and zlink_publish() reports success. Callers that need backpressure instead of dropping must set ZLINK_PUB_OPT_NODROP to 1 explicitly.

When PUB's send queue is full (HWM), the default (ZLINK_PUB_OPT_NODROP=0) silently drops the message for that subscriber. For details, see Performance Guide.

3. Topic Filtering¶

Topic filtering in SUB sockets uses prefix matching.

Subscription Topic	Received Message	Match
`"weather"`	`"weather: sunny"`	O
`"weather"`	`"weathering storm"`	O
`"weather"`	`"sports: baseball"`	X
`""` (empty string)	All messages	O

Multiple Topic Subscriptions¶

/* Subscribe to multiple topics */
zlink_set_subscription(sub, "weather");
zlink_set_subscription(sub, "sports");

/* Unsubscribe */
zlink_unset_subscription(sub, "sports");

Empty Subscription (All Messages)¶

/* Subscribe with empty string -- receives all messages */
zlink_set_subscription(sub, "");

Reference: core/tests/integration/test_pubsub.cpp -- zlink_set_subscription(subscriber, "")

4. Message Format¶

zlink_publish() takes a topic and a payload-part array as separate parameters:

ZLINK_EXPORT zlink_submit_result_t zlink_publish (void *subject_,
                                                       const char *topic_id_,
                                                       zlink_msg_t *parts_,
                                                       size_t part_count_,
                                                       zlink_send_flags_t flags_);

For a multipart publish, place every part in array order and call zlink_publish() once. Core atomically submits the complete array as one publish record.

/* Publish: topic = "sensor:cpu", payload = 2 frames */
zlink_msg_t parts[2];
zlink_msg_init_size(&parts[0], 4);
memcpy(zlink_msg_data(&parts[0]), "host", 4);
zlink_msg_init_size(&parts[1], 2);
memcpy(zlink_msg_data(&parts[1]), "73", 2);

zlink_submit_result_t rc = zlink_publish(
    pub, "sensor:cpu", parts, 2, ZLINK_SEND_FLAGS_NONE);

/* One zlink_subscribe() call returns the complete payload array:
   topic         = "sensor:cpu"
   parts[0]      = "host"
   parts[1]      = "73"
   part_count    = 2 */

The topic is sent on the wire as the first frame. zlink_subscribe() separates the topic from the payload array on the receive side. Callers never need to assemble topic frames manually.

Note: Passing NULL as topic_id_ (zlink_publish(pub, NULL, &part, ...)) activates a compatibility path where the first message frame carries the topic according to the wire-prefix rule. This is not recommended. Always pass the topic_id_ parameter explicitly.

5. PUB/SUB Socket Options¶

SUB-Specific Functions¶

Function	Description
`zlink_set_subscription()`	Add topic subscription (prefix match)
`zlink_unset_subscription()`	Remove topic subscription

Common Options¶

Option	Type	Default	Description
`ZLINK_OPT_SNDHWM`	`uint64_t` bytes	automatic (fanout floor by default)	Default for PUB-family sockets. Recomputed within the same role budget as connections grow; `0` is unlimited
`ZLINK_OPT_RCVHWM`	`uint64_t` bytes	automatic (recv_ingress floor by default)	Default for SUB-family sockets. Recomputed within the same role budget as connections grow; `0` is unlimited
`ZLINK_OPT_LINGER`	int	-1	Wait time on close (ms)

6. PUB/SUB Usage Patterns¶

Pattern 1: Basic PUB/SUB¶

/* PUB */
void *pub = zlink_socket(ctx, ZLINK_SOCKET_PUB);
zlink_bind(pub, "tcp://*:5556");

/* SUB -- receive all messages */
void *sub = zlink_socket(ctx, ZLINK_SOCKET_SUB);
zlink_connect(sub, "tcp://127.0.0.1:5556");
zlink_set_subscription(sub, "");

msleep(100);  /* time for subscription to reach PUB */

zlink_msg_t msg;
zlink_msg_init_size(&msg, 4);
memcpy(zlink_msg_data(&msg), "test", 4);
zlink_publish(pub, NULL, &msg, 1, ZLINK_SEND_FLAGS_NONE);

/* zlink_subscribe() on sub returns "test" once the record arrives */

Reference: core/tests/integration/test_pubsub.cpp -- test_tcp()

Pattern 2: Multiple SUBs¶

Multiple SUBs connect to a single PUB. Each SUB receives only its own topics.

void *pub = zlink_socket(ctx, ZLINK_SOCKET_PUB);
zlink_bind(pub, "tcp://*:5556");

void *sub_weather = zlink_socket(ctx, ZLINK_SOCKET_SUB);
zlink_connect(sub_weather, "tcp://127.0.0.1:5556");
zlink_set_subscription(sub_weather, "weather");

void *sub_sports = zlink_socket(ctx, ZLINK_SOCKET_SUB);
zlink_connect(sub_sports, "tcp://127.0.0.1:5556");
zlink_set_subscription(sub_sports, "sports");

/* Only sub_weather receives weather, only sub_sports receives sports */

Pattern 3: Multiple PUBs → SUB¶

A SUB can connect to multiple PUBs. It receives messages from all PUBs via fair-queue.

void *sub = zlink_socket(ctx, ZLINK_SOCKET_SUB);
zlink_set_subscription(sub, "");
zlink_connect(sub, "tcp://pub1:5556");
zlink_connect(sub, "tcp://pub2:5557");

7. PUB/SUB Caveats¶

Slow Subscriber (HWM Handling)¶

By default PUB/XPUB run in lossy mode — ZLINK_PUB_OPT_NODROP defaults to 0. When a slow subscriber's send queue reaches the HWM, the message for that subscriber is silently dropped (no error returned) and zlink_publish() reports success. Delivery to the other subscribers is unaffected.

/* Default — the slow subscriber's copy is dropped on HWM, publish succeeds */
struct quote_tick tick = {.price_micros = 91450000000LL, .volume = 1420};
zlink_msg_t quote;
zlink_msg_init_size(&quote, sizeof(tick));
memcpy(zlink_msg_data(&quote), &tick, sizeof(tick));
zlink_publish(pub, "quotes.KRW-BTC", &quote, 1, ZLINK_SEND_FLAGS_DONTWAIT);

/* Raise the HWM to absorb bursts and reduce loss */
uint64_t hwm_bytes = 64 * 1024 * 1024;  /* HWM is bytes */
zlink_set_option(pub, ZLINK_OPT_SNDHWM, &hwm_bytes, sizeof(hwm_bytes));

NODROP Mode — Backpressure Instead of Drop¶

Setting ZLINK_PUB_OPT_NODROP to 1 makes zlink_publish() return ZLINK_SUBMIT_BACKPRESSURED on HWM instead of dropping, so the caller can react.

/* Enable NODROP mode (backpressure on HWM) */
int nodrop = 1;
zlink_set_pub_option(pub, ZLINK_PUB_OPT_NODROP, &nodrop, sizeof(nodrop));

struct quote_tick tick = {.price_micros = 91450000000LL, .volume = 1420};
zlink_msg_t quote;
zlink_msg_init_size(&quote, sizeof(tick));
memcpy(zlink_msg_data(&quote), &tick, sizeof(tick));
zlink_submit_result_t rc = zlink_publish(
    pub, "quotes.KRW-BTC", &quote, 1, ZLINK_SEND_FLAGS_DONTWAIT);
if (rc == ZLINK_SUBMIT_BACKPRESSURED) {
    /* HWM reached — the call already consumed quote; resubmit a retained
       retained copy of the complete record as a new array. */
}

This mode couples the publisher to its slowest subscriber: one full pipe stops delivery to every subscriber on the socket. Reliable delivery that must not depend on subscriber speed belongs on a request-reply socket, not on PUB/SUB.

Mode	Behavior on HWM	When to Use
Default (`NODROP=0`, lossy)	Silent drop — no error, message lost	Ordinary fanout (observation, notification, sensor, tick)
`NODROP=1`	Returns `ZLINK_SUBMIT_BACKPRESSURED` — caller controls	Loss is unacceptable and coupling to the slowest subscriber is acceptable

ZLINK_PUB_OPT_NODROP applies to both PUB and XPUB sockets (PUB is implemented on top of XPUB).

Late Joiner (Messages Lost Before Subscription)¶

Messages published before the subscription message from SUB reaches PUB are lost.

/* Time needed for subscription to propagate to PUB */
zlink_connect(sub, "tcp://127.0.0.1:5556");
zlink_set_subscription(sub, "topic");
msleep(100);  /* wait for subscription propagation */
/* Only messages published after this point can be received */

Direction Constraints¶

PUB/SUB each have their own dedicated API:

/* PUB: send via zlink_publish(). Cannot receive */
zlink_msg_t parts[4];
for (size_t i = 0; i < 4; ++i) {
    zlink_msg_init_size(&parts[i], 5);
    memcpy(zlink_msg_data(&parts[i]), "sunny", 5);
}
zlink_publish(pub, "weather", &parts[0], 1, ZLINK_SEND_FLAGS_NONE);  /* OK */

/* Using zlink_send() on PUB → returns ZLINK_SUBMIT_NOT_SUPPORTED */
zlink_send(pub, &parts[1], 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* SUB: receive via zlink_subscribe(). Cannot send/publish */
zlink_publish(sub, "weather", &parts[2], 1, ZLINK_SEND_FLAGS_NONE);  /* ZLINK_SUBMIT_NOT_SUPPORTED */
zlink_send(sub, &parts[3], 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);     /* ZLINK_SUBMIT_NOT_SUPPORTED */

Part II: XPUB/XSUB¶

8. XPUB/XSUB Overview¶

XPUB/XSUB are advanced publish-subscribe sockets that allow applications to handle subscription frames directly. They are used for building proxies/brokers, subscription monitoring, and Last-Value Caching.

SUB vs XSUB — Key Difference¶

	SUB	XSUB
Topic registration	`zlink_set_subscription()`	`zlink_set_subscription()` (same)
Message receive	`zlink_subscribe()` — filtered	`zlink_subscribe()` — no filter, receives all
Local filter	On — drops non-matching	Off — passes all messages
No subscriptions	Receives nothing	Receives all messages

XSUB is needed in proxies because it passes all messages through without applying its own filter. Topic registration is sent to upstream identically via zlink_set_subscription() on both.

PUB vs XPUB — Key Difference¶

	PUB	XPUB
Message publish	`zlink_publish()`	`zlink_publish()` (same)
Subscription events	Not exposed	`zlink_xpub_recv()`

XPUB can observe which clients subscribe to or unsubscribe from which topics.

XSUB/XPUB Roles in a Proxy¶

A proxy has two separate flows:

%%{init: {'flowchart': {'nodeSpacing': 32, 'rankSpacing': 40, 'padding': 8, 'wrappingWidth': 180}, 'themeVariables': {'fontSize': '18px'}}}%%
flowchart LR
  subgraph data ["Data flow (publish)"]
    direction LR
    P1[PUB] -- publish --> X1[XSUB] == proxy forward ==> X2[XPUB] -- deliver --> S1[SUB]
  end
  subgraph sub ["Subscription flow (reverse)"]
    direction RL
    S2[SUB] -. subscribe .-> X3[XPUB] -. propagate .-> X4[XSUB] -. register .-> P2[PUB]
  end

Data Flow¶

Step	Actor	Action	Note
1	PUB	`zlink_publish(pub, topic, ...)`	Publish data
2	proxy internal	XSUB internal recv → XPUB internal send	Handled by `zlink_proxy()`
3	SUB	`zlink_subscribe()`	Final consumption

Key point: The proxy data relay in zlink_proxy(xsub, xpub, NULL) uses internal recv/send paths, not the public zlink_send()/zlink_recv() API. Users never need to call XSUB recv → XPUB send directly.

Subscription Propagation Flow¶

Step	Actor	Action	API
1	SUB	Subscribe → arrives at XPUB via wire	`zlink_set_subscription(sub, "weather")`
2	proxy app	Receive subscription event from XPUB	`zlink_xpub_recv(xpub, ...)`
3	proxy app	Register on XSUB → propagates to PUB via wire	`zlink_set_subscription(xsub, "weather")`
4	PUB	Publish matching data	`zlink_publish(pub, "weather", ...)`
5	data flow	XSUB → XPUB → SUB	Handled by `zlink_proxy()`

zlink_set_subscription() sends subscription info upstream on the wire identically for both SUB and XSUB. Calling it on XSUB in a proxy is not because "XSUB can send" — the proxy app registers subscription events received from XPUB onto XSUB to propagate them upstream.

Why XSUB/XPUB?¶

Question	With SUB/PUB	With XSUB/XPUB
Data pass-through	SUB local filter on — must register subscriptions	XSUB local filter off — passes all
Subscription events	PUB does not expose them	XPUB exposes them via `zlink_xpub_recv()`
Proxy suitability	Proxy must manage topics itself	Relay-only — ideal for proxy

PUB/SUB Socket Public API Summary¶

Public API	PUB	SUB	XPUB	XSUB
`zlink_publish()`	OK	—	OK	—
`zlink_subscribe()`	—	OK	—	OK
`zlink_set_subscription()`	—	OK	—	OK
`zlink_xpub_recv()`	—	—	OK	—
Local filter	N/A	On	N/A	Off

zlink_send() / zlink_recv() return ZLINK_SUBMIT_NOT_SUPPORTED / ZLINK_RECV_NOT_SUPPORTED on all 4 PUB/SUB sockets. Use zlink_publish() for publishing and zlink_subscribe() for receiving.

Proxy patterns (built-in zlink_proxy(), manual proxy construction, ROUTER/DEALER broker) are covered in the Proxy Guide.

9. Subscription Frame Format¶

Subscription/unsubscription frames between XPUB/XSUB follow this format:

Byte	Meaning
`0x01` + topic	Subscription request
`0x00` + topic	Unsubscription request

/* Subscribe from XSUB */
zlink_set_subscription(xsub, "A");

/* Unsubscribe from XSUB */
zlink_unset_subscription(xsub, "A");

XPUB receives subscription frames with zlink_xpub_recv():

void *xpub = zlink_socket(ctx, ZLINK_SOCKET_XPUB);
zlink_bind(xpub, "tcp://*:5557");

const zlink_routing_id_t *source_rid = NULL;
int subscribed = 0;
char topic[256];
size_t topic_len = 0;

zlink_recv_result_t rc = zlink_xpub_recv(
  xpub, &source_rid, &subscribed, topic, sizeof(topic), &topic_len, ZLINK_RECV_FLAGS_NONE);

Reference: core/tests/integration/test_xpub_manual.cpp -- subscription1[] = {1, 'A'}, unsubscription1[] = {0, 'A'}

10. XPUB Socket Options¶

Option	Type	Default	Description
`ZLINK_PUB_OPT_MANUAL`	int	0	Enable manual subscription management mode
`ZLINK_PUB_OPT_VERBOSE`	int	0	Also surface duplicate subscribes for already-subscribed topics as events
`ZLINK_PUB_OPT_APPROVE_SUBSCRIBE`	binary	--	(MANUAL mode) Approve the topic subscription for the pipe of the most recently dequeued subscription event
`ZLINK_PUB_OPT_REJECT_SUBSCRIBE`	binary	--	(MANUAL mode) Reject the topic subscription for the pipe of the most recently dequeued subscription event

XPUB_MANUAL Mode¶

By default, XPUB processes SUB subscriptions automatically. In MANUAL mode, after receiving a subscription event with zlink_xpub_recv(), the application decides the actual subscription with ZLINK_PUB_OPT_APPROVE_SUBSCRIBE / ZLINK_PUB_OPT_REJECT_SUBSCRIBE through zlink_set_pub_option(). zlink_set_subscription() / zlink_unset_subscription() are SUB/XSUB-only and fail with ZLINK_CONFIG_INVALID_ARGUMENT on an XPUB handle.

/* Enable MANUAL mode */
int manual = 1;
zlink_set_pub_option(xpub, ZLINK_PUB_OPT_MANUAL, &manual, sizeof(manual));

/* zlink_xpub_recv() returns subscribed=1, topic="A"
   Then approve a transformed subscription for that subscriber's pipe: */
const char mapped_topic[] = "XA";
zlink_set_pub_option(xpub, ZLINK_PUB_OPT_APPROVE_SUBSCRIBE,
                     mapped_topic, sizeof(mapped_topic) - 1);

/* Publish */
zlink_msg_t msg_a;
zlink_msg_init_size(&msg_a, 1);
memcpy(zlink_msg_data(&msg_a), "A", 1);
zlink_publish(xpub, NULL, &msg_a, 1, ZLINK_SEND_FLAGS_NONE);   /* does not reach the subscriber */

zlink_msg_t msg_xa;
zlink_msg_init_size(&msg_xa, 2);
memcpy(zlink_msg_data(&msg_xa), "XA", 2);
zlink_publish(xpub, NULL, &msg_xa, 1, ZLINK_SEND_FLAGS_NONE);  /* subscriber receives this */

Reference: core/tests/integration/test_xpub_manual.cpp -- test_basic(): subscription request for A → transformed to B

11. XPUB/XSUB Usage Patterns¶

Pattern 1: Building a Proxy/Broker¶

Build a PUB/SUB proxy using XSUB (frontend) + XPUB (backend).

/* Proxy frontend: PUBs connect here */
void *xsub = zlink_socket(ctx, ZLINK_SOCKET_XSUB);
zlink_bind(xsub, "tcp://*:5556");

/* Proxy backend: SUBs connect here */
void *xpub = zlink_socket(ctx, ZLINK_SOCKET_XPUB);
zlink_bind(xpub, "tcp://*:5557");

/* Run proxy (forwards messages and subscriptions bidirectionally) */
zlink_proxy(xsub, xpub, NULL);

Pattern 2: MANUAL Mode Proxy (Subscription Transformation)¶

An advanced proxy that transforms or filters subscription requests.

int manual = 1;
zlink_set_pub_option(xpub, ZLINK_PUB_OPT_MANUAL, &manual, sizeof(manual));

for (;;) {
    const zlink_routing_id_t *source_rid = NULL;
    int subscribed = 0;
    char topic[256];
    size_t topic_len = 0;

    zlink_recv_result_t rc = zlink_xpub_recv(
      xpub, &source_rid, &subscribed, topic, sizeof(topic) - 1, &topic_len, ZLINK_RECV_FLAGS_NONE);
    if (rc != ZLINK_RECV_OK)
        break;
    topic[topic_len] = '\0';   /* topic bytes are binary-safe (no NUL) — set_subscription takes a C string */

    if (subscribed) {
        /* Approve the subscription for the pipe of the event just dequeued */
        zlink_set_pub_option(xpub, ZLINK_PUB_OPT_APPROVE_SUBSCRIBE, topic, topic_len);

        /* Propagate subscription upstream (XSUB) */
        zlink_set_subscription(xsub, topic);
    } else {
        /* Reject (remove) the subscription for that pipe */
        zlink_set_pub_option(xpub, ZLINK_PUB_OPT_REJECT_SUBSCRIBE, topic, topic_len);

        zlink_unset_subscription(xsub, topic);
    }
}

Reference: core/tests/integration/test_xpub_manual.cpp -- test_xpub_proxy_unsubscribe_on_disconnect()

Pattern 3: Subscription Monitoring¶

Use XPUB to observe which clients subscribe to which topics.

void *xpub = zlink_socket(ctx, ZLINK_SOCKET_XPUB);
zlink_bind(xpub, "tcp://*:5557");

for (;;) {
    const zlink_routing_id_t *source_rid = NULL;
    int subscribed = 0;
    char topic[256];
    size_t topic_len = 0;

    zlink_recv_result_t rc = zlink_xpub_recv(
      xpub, &source_rid, &subscribed, topic, sizeof(topic), &topic_len, ZLINK_RECV_FLAGS_NONE);
    if (rc != ZLINK_RECV_OK)
        break;
    printf("%s: %.*s\n", subscribed ? "New subscription" : "Unsubscription",
           (int) topic_len, topic);
}

Pattern 4: Automatic Unsubscribe on Subscriber Disconnect¶

When a SUB disconnects, an unsubscribe frame is automatically delivered to XPUB.

/* After SUB disconnects */
zlink_close(sub);

/* The next zlink_xpub_recv() returns
   subscribed=0 and the previously subscribed topic */

Reference: core/tests/integration/test_xpub_manual.cpp -- test_xpub_proxy_unsubscribe_on_disconnect()

12. Caveats¶

Subscription Propagation Timing¶

Subscription messages are propagated asynchronously. Messages published immediately after subscribing may not be received.

zlink_connect(sub, endpoint);
zlink_set_subscription(sub, "topic");
/* Publishing a "topic" message at this point may result in loss */
msleep(100);  /* wait for subscription propagation */
/* Messages published after this point can be received */

Subscription Management in XPUB MANUAL Mode¶

In MANUAL mode, if ZLINK_PUB_OPT_APPROVE_SUBSCRIBE is not applied after receiving a subscription event, that subscription is not registered. Subscriptions must be explicitly processed.

Multiple Subscribers → Single XPUB¶

When multiple SUBs subscribe to the same topic, the XPUB subscription is maintained until all SUBs have unsubscribed.

Reference: core/tests/integration/test_xpub_manual.cpp -- test_missing_subscriptions(): processing two subscribers sequentially to prevent omissions

← PAIR | DEALER →

Full language examples¶

C++C#/.NETJavaKotlinPythonNode/TypeScriptJavaScriptGoRust

zlink::context_t ctx;
zlink::xpub_socket_t publisher (ctx);
zlink::sub_socket_t subscriber (ctx);
zlink::socket_monitor_t pub_monitor = publisher.monitor_open ();
zlink::socket_monitor_t sub_monitor = subscriber.monitor_open ();

publisher.bind ("tcp://127.0.0.1:0");
const std::string endpoint = publisher.options ().last_endpoint ();
assert (!endpoint.empty ());
subscriber.connect (endpoint);
assert (detail::wait_connected (pub_monitor, sub_monitor));

const std::string topic = detail::k_pubsub_topic;
subscriber.set_subscription (topic);

zlink::subscription_event_t event;
assert (publisher.receive_subscription_event (event)
        == static_cast<int> (zlink::recv_result_t::ok));
assert (event.subscribed);
assert (event.topic == topic);

const std::string sent = detail::k_pubsub_payload;
zlink::message_t outbound = detail::make_message (sent);
publisher.publish (topic).message (outbound).submit ();

zlink::topic_message_t inbound;
assert (subscriber.subscribe (inbound) == static_cast<int> (zlink::recv_result_t::ok));
assert (inbound.topic () == topic);
assert (inbound.parts ().size () == 1);
const std::string received = inbound.parts ()[0].to_string ();
assert (received == detail::k_pubsub_payload);
inbound.close ();
std::printf ("[pubsub/recv] publish: \"%s/%s\" → subscribe: \"%s/%s\"\n", topic.c_str (),
             sent.c_str (), topic.c_str (), received.c_str ());
return 0;

if (!SampleSupport.IsNativeAvailable())
    return;

using var ctx = Zlink.CreateContext();
using var publisher = ctx.CreatePubSocket();
using var subscriber = ctx.CreateSubSocket();
string endpoint = SampleSupport.NewEndpoint("tcp", "sample");
using var publisherMonitor = publisher.MonitorOpen(SocketEvent.ConnectionReady);
using var subscriberMonitor = subscriber.MonitorOpen(SocketEvent.ConnectionReady);
publisher.Bind(endpoint);
subscriber.Connect(endpoint);
SampleSupport.WaitConnected(publisherMonitor, subscriberMonitor);
subscriber.SetSubscription("prices");

using (Message message = Message.From("101.25"))
    publisher.Publish("prices").Message(message).Submit();
string payload = SampleSupport.SubscribeUtf8(subscriber, out string topic, 2000);
Console.WriteLine(
    $"[pubsub/recv] publish: \"prices/101.25\" -> subscribe: \"{topic}/{payload}\"");

SampleSupport.ensureNative();
String endpoint = SampleSupport.tcpEndpoint();
String published = SampleSupport.PUBSUB_TOPIC + "/" + SampleSupport.PUBSUB_PAYLOAD;

try (Context ctx = Zlink.createContext();
     PubSocket pub = ctx.createPubSocket();
     SubSocket sub = ctx.createSubSocket();
     var pubMonitor = pub.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY);
     var subMonitor = sub.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY)) {
    pub.bind(endpoint);
    sub.setSubscription(SampleSupport.PUBSUB_TOPIC);
    sub.connect(endpoint);
    SampleSupport.waitPubSubReady(pubMonitor, subMonitor);

    try (Message payload = Message.from(SampleSupport.PUBSUB_PAYLOAD)) {
        pub.publish(SampleSupport.PUBSUB_TOPIC).message(payload).submit();
    }

    try (var received = new TopicMessage()) {
        if (!sub.subscribe(received, RecvFlags.NONE)) {
            throw new IllegalStateException("no pubsub delivery");
        }
        String value = received.topic() + "/"
            + received.singlePartOrThrow().toUtf8String();
        if (!published.equals(value)) {
            throw new IllegalStateException("unexpected delivery: " + value);
        }
        System.out.println("[pubsub/recv] publish: \"" + published
            + "\" \u2192 subscribe: \"" + value + "\"");
    }
}

SampleSupport.ensureNative()
val endpoint = SampleSupport.tcpEndpoint()
val published = "${SampleSupport.PUBSUB_TOPIC}/${SampleSupport.PUBSUB_PAYLOAD}"

Zlink.createContext().use { ctx ->
    ctx.createPubSocket().use { pub ->
        ctx.createSubSocket().use { sub ->
            pub.monitorOpen(MonitorEventType.CONNECTION_READY).use { pubMonitor ->
                sub.monitorOpen(MonitorEventType.CONNECTION_READY).use { subMonitor ->
                    pub.bind(endpoint)
                    sub.setSubscription(SampleSupport.PUBSUB_TOPIC)
                    sub.connect(endpoint)
                    SampleSupport.waitPubSubReady(pubMonitor, subMonitor)

                    Message.from(SampleSupport.PUBSUB_PAYLOAD).use { payload ->
                        pub.publish(SampleSupport.PUBSUB_TOPIC).message(payload).submit()
                    }

                    TopicMessage().use { received ->
                        check(sub.subscribe(received, RecvFlags.NONE)) { "no pubsub delivery" }
                        val value = "${received.topic()}/${received.singlePartOrThrow().toUtf8String()}"
                        check(published == value) { "unexpected delivery: $value" }
                        println("[pubsub/recv] publish: \"$published\" → subscribe: \"$value\"")
                    }
                }
            }
        }
    }
}

_, endpoint = tcp_endpoint()

with zlink.create_context() as ctx:
    with zlink.create_xpub_socket(ctx) as publisher:
        with zlink.create_sub_socket(ctx) as subscriber:
            with publisher.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as publisher_monitor:
                with subscriber.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as subscriber_monitor:
                    publisher.bind(endpoint)
                    subscriber.connect(endpoint)
                    subscriber.set_subscription(b"prices")
                    wait_connected(publisher_monitor, subscriber_monitor)

            event = zlink.create_subscription_event()
            if not publisher.receive_subscription_event_into(event):
                raise AssertionError("missing subscription event")
            if not event.subscribed or event.topic != "prices":
                raise AssertionError("unexpected subscription event")
            publisher.publish(b"prices").message(b"101.25").submit()
            received = zlink.create_topic_message()
            assert subscriber.subscribe_into(received)
            with received:
                if received.topic != "prices":
                    raise AssertionError(f"unexpected pubsub topic: {received.topic!r}")
                if received.to_bytes_list() != [b"101.25"]:
                    raise AssertionError("unexpected pubsub payload")
            print('[pubsub/recv] publish: "prices/101.25" → subscribe: "prices/101.25"')

const endpoint = await tcpEndpoint();
const ctx = zlink.createContext();
const pub = zlink.createPubSocket(ctx);
const sub = zlink.createSubSocket(ctx);

try {
  const pubMonitor = pub.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const subMonitor = sub.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    pub.bind(endpoint);
    sub.connect(endpoint);
    pubMonitor.recv();
    subMonitor.recv();
  } finally {
    pubMonitor.close();
    subMonitor.close();
  }

  const topic = 'prices';
  const sent = '101.25';
  sub.setSubscription(topic);
  const deadline = Date.now() + 5000;
  const received = new zlink.TopicMessage();
  let hasReceived = false;
  while (Date.now() < deadline) {
    pub.publish(topic).message(Buffer.from(sent)).submit();
    try {
      if (sub.subscribe(received, zlink.RecvFlags.DontWait)) {
        hasReceived = true;
        break;
      }
    } catch (error) {
      if (!(error instanceof zlink.RecvError && error.result === zlink.RecvResult.NoData)) {
        throw error;
      }
    }
    await new Promise((resolve) => setTimeout(resolve, 25));
  }
  assert.equal(hasReceived, true);
  try {
    const recv = received.parts[0].data().toString();
    assert.equal(received.topic, topic);
    assert.equal(recv, sent);
    console.log(`[pubsub/recv] publish: "${topic}/${sent}" \u2192 subscribe: "${topic}/${recv}"`);
  } finally {
    received.close();
  }
} finally {
  sub.close();
  pub.close();
  ctx.close();
}

const port = await reservePort();
const endpoint = `tcp://127.0.0.1:${port}`;
const ctx = zlink.createContext();
const pub = zlink.createPubSocket(ctx);
const sub = zlink.createSubSocket(ctx);

try {
  const pubMonitor = pub.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const subMonitor = sub.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    pub.bind(endpoint);
    sub.connect(endpoint);
    pubMonitor.recv();
    subMonitor.recv();
  } finally {
    pubMonitor.close();
    subMonitor.close();
  }

  const topic = 'prices';
  const sent = '101.25';
  sub.setSubscription(topic);
  const deadline = Date.now() + 5000;
  const received = new zlink.TopicMessage();
  let hasReceived = false;
  while (Date.now() < deadline) {
    pub.publish(topic).message(Buffer.from(sent)).submit();
    try {
      if (sub.subscribe(received, zlink.RecvFlags.DontWait)) {
        hasReceived = true;
        break;
      }
    } catch (error) {
      if (!(error instanceof zlink.RecvError && error.result === zlink.RecvResult.NoData)) {
        throw error;
      }
    }
    await new Promise((resolve) => setTimeout(resolve, 25));
  }
  assert.equal(hasReceived, true);
  try {
    const recv = received.parts[0].data().toString();
    assert.equal(received.topic, topic);
    assert.equal(recv, sent);
    console.log(`[pubsub/recv] publish: "${topic}/${sent}" → subscribe: "${topic}/${recv}"`);
  } finally {
    received.close();
  }
} finally {
  sub.close();
  pub.close();
  ctx.close();
}

ctx, err := zlink.NewContext()
samplecommon.Must(err)
defer ctx.Close()

publisher, err := ctx.XPubSocket()
samplecommon.Must(err)
defer publisher.Close()
subscriber, err := ctx.SubSocket()
samplecommon.Must(err)
defer subscriber.Close()

pubMon := samplecommon.OpenMonitor(publisher)
defer pubMon.Close()
subMon := samplecommon.OpenMonitor(subscriber)
defer subMon.Close()

endpoint := samplecommon.UniqueTCP("pubsub-recv")
samplecommon.Must(publisher.Bind(endpoint))
samplecommon.Must(subscriber.Connect(endpoint))
samplecommon.WaitConnected(pubMon, subMon)

topic := "prices"
samplecommon.Must(subscriber.SetSubscription(topic))
var event zlink.SubscriptionEvent
_, err = publisher.ReceiveSubscriptionEvent(&event, zlink.RecvFlagsNone)
samplecommon.Must(err)
if !event.Subscribed() || event.Topic() != topic {
    samplecommon.Must(fmt.Errorf("unexpected subscription event"))
}

payload := "101.25"
_, err = publisher.Publish(topic).Message(samplecommon.Message(payload)).Submit(context.Background())
samplecommon.Must(err)

var message zlink.TopicMessage
_, err = subscriber.Subscribe(&message, zlink.RecvFlagsNone)
samplecommon.Must(err)
defer message.Close()
part, err := message.SinglePartOrError()
samplecommon.Must(err)
if !bytes.Equal(part.Data(), []byte(payload)) {
    samplecommon.Must(fmt.Errorf("unexpected payload %q", string(part.Data())))
}

fmt.Printf("[pubsub/recv] publish: %q -> subscribe: %q\n", topic+"/"+payload, message.Topic()+"/"+string(part.Data()))

let ctx = Context::new().expect("context creation failed");
let endpoint = sample_support::tcp_endpoint();

let pub_sock = ctx.pub_socket().expect("pub socket failed");
let sub_sock = ctx.sub_socket().expect("sub socket failed");

sub_sock
    .set_subscription("prices")
    .expect("set_subscription failed");

let pub_mon = SocketMonitor::open(&pub_sock).expect("pub monitor open failed");
let sub_mon = SocketMonitor::open(&sub_sock).expect("sub monitor open failed");

pub_sock.bind(&endpoint).expect("bind failed");
sub_sock.connect(&endpoint).expect("connect failed");

sample_support::wait_connected(&[&pub_mon, &sub_mon]);
drop(pub_mon);
drop(sub_mon);

let msg = Message::try_from(b"101.25").expect("message failed");
pub_sock
    .publish("prices")
    .message(msg)
    .submit()
    .expect("publish failed");

let mut topic_msg = TopicMessage::empty();
assert!(
    sub_sock
        .subscribe(&mut topic_msg, RecvFlags::NONE)
        .expect("subscribe recv failed")
);
assert_eq!(topic_msg.topic(), "prices");
assert_eq!(topic_msg.parts()[0].as_str().unwrap(), "101.25");
println!(
    "[pubsub/recv] publish: \"{}/101.25\" → subscribe: \"{}/{}\"",
    topic_msg.topic(),
    topic_msg.topic(),
    topic_msg.parts()[0].as_str().unwrap()
);

← PAIR | DEALER →

PUB/SUB/XPUB/XSUB Publish-Subscribe¶

1. Overview¶

SUB vs XSUB — Key Difference¶

Part I: PUB/SUB¶

2. PUB/SUB Basic Usage¶

Publisher (PUB)¶

Subscriber (SUB)¶

Sending and Receiving Summary¶

3. Topic Filtering¶

Multiple Topic Subscriptions¶

Empty Subscription (All Messages)¶

4. Message Format¶

5. PUB/SUB Socket Options¶

SUB-Specific Functions¶

Common Options¶

6. PUB/SUB Usage Patterns¶

Pattern 1: Basic PUB/SUB¶

Pattern 2: Multiple SUBs¶

Pattern 3: Multiple PUBs → SUB¶

7. PUB/SUB Caveats¶

Slow Subscriber (HWM Handling)¶

NODROP Mode — Backpressure Instead of Drop¶

Late Joiner (Messages Lost Before Subscription)¶

Direction Constraints¶

Part II: XPUB/XSUB¶

8. XPUB/XSUB Overview¶

SUB vs XSUB — Key Difference¶

PUB vs XPUB — Key Difference¶

XSUB/XPUB Roles in a Proxy¶

Data Flow¶

Subscription Propagation Flow¶

Why XSUB/XPUB?¶

PUB/SUB Socket Public API Summary¶

9. Subscription Frame Format¶

10. XPUB Socket Options¶

XPUB_MANUAL Mode¶

11. XPUB/XSUB Usage Patterns¶

Pattern 1: Building a Proxy/Broker¶

Pattern 2: MANUAL Mode Proxy (Subscription Transformation)¶

Pattern 3: Subscription Monitoring¶

Pattern 4: Automatic Unsubscribe on Subscriber Disconnect¶

12. Caveats¶

Subscription Propagation Timing¶

Subscription Management in XPUB MANUAL Mode¶

Multiple Subscribers → Single XPUB¶

Full language examples¶