Socket Index | Previous: XPUB | Next: DEALER

Socket — XSUB¶

What this chapter defines — The public contract of the XSUB socket (a SUB that sends subscriptions as messages).

1. XSUB Overview¶

XSUB is an extended subscriber socket that supports subscription forwarding. XSUB supports the same subscribe/unsubscribe and topic receive APIs as SUB, but does not filter received messages with its own filter matching. A connected PUB or XPUB receives XSUB's subscription messages and performs the filtering on the publisher side. XSUB delivers every message that actually arrives to the application.

This document defines the public contract for registering, removing, and querying subscriptions on XSUB and for receiving the topic and complete payload record. Its intended audience is developers who map this contract to the C API and each language binding.

The following documents own the related contracts.

Related contract	Defining document
Common socket options, receive model, and common functions (`zlink_set_option`, etc.)	Socket Common
SUB socket subscription behavior (subscription APIs shared with XSUB)	SUB
Publisher-side subscription event observation and manual subscription management	XPUB

2. Subscription Behavior¶

XSUB subscriptions operate on topic filters.

The application registers a topic filter with zlink_set_subscription. XSUB increments a reference count when the same filter is registered more than once.
XSUB updates its local reference count first and then sends the subscribe message to each connected upstream pipe. The publisher-side PUB or XPUB is responsible for filtering and selects only messages whose topics match the filter byte prefix. If a pipe cannot accept the write because of SNDHWM, the subscribe message for that pipe is dropped while zlink_set_subscription still succeeds — that pipe never receives the subscription. XPUB owns the contract for observing and manually managing subscription events.
XSUB does not apply its own filter matching. It delivers every message that actually arrives so that the application can receive the complete record with zlink_subscribe.
zlink_unset_subscription decrements the reference count of a registered subscription. XSUB sends an upstream unsubscribe message only when the last registration is removed. Removing an unregistered filter sends no message upstream.

%%{init: {'sequence': {'actorFontSize': '18px', 'messageFontSize': '18px', 'noteFontSize': '18px', 'boxMargin': 8, 'width': 140}, 'themeVariables': {'fontSize': '18px'}}}%%
sequenceDiagram
    participant App as Application
    participant XSUB as XSUB socket
    participant Up as Upstream peer
    App->>XSUB: zlink_set_subscription(filter)
    XSUB->>Up: Forward subscription message
    Note over Up: XPUB filters by subscription
    Up-->>XSUB: Message selected and sent by upstream
    App->>XSUB: zlink_subscribe()
    XSUB-->>App: Copy topic bytes + return the complete payload record

Query the number of currently subscribed topics with the ZLINK_SUB_OPT_TOPICS_COUNT option and query individual subscription filters with zlink_subscription_at.

3. Sub Options (`zlink_sub_option_t`)¶

Use these options with zlink_set_sub_option() / zlink_get_sub_option().

typedef enum zlink_sub_option_t
{
    ZLINK_SUB_OPT_TOPICS_COUNT = 0x3400  // Number of subscribed topics (read-only, int)
} zlink_sub_option_t;

4. Functions¶

zlink_set_sub_option¶

Sets an option specific to SUB/XSUB sockets.

ZLINK_EXPORT zlink_config_result_t zlink_set_sub_option (void *handle_,
                           zlink_sub_option_t option_,
                           const void *optval_,
                           size_t optvallen_);

Sets a SUB/XSUB socket option. Use zlink_set_option() for common options shared by all socket types.

Returns: ZLINK_CONFIG_OK on success; otherwise a zlink_config_result_t value. zlink_errno() retains the detailed internal errno for diagnostics.

See also: zlink_get_sub_option, zlink_set_option

zlink_get_sub_option¶

Gets an option specific to SUB/XSUB sockets.

ZLINK_EXPORT zlink_config_result_t zlink_get_sub_option (void *handle_,
                           zlink_sub_option_t option_,
                           void *optval_,
                           size_t *optvallen_);

Gets the current value of a SUB/XSUB socket option.

Returns: ZLINK_CONFIG_OK on success; otherwise a zlink_config_result_t value. zlink_errno() retains the detailed internal errno for diagnostics.

See also: zlink_set_sub_option

zlink_set_subscription¶

Subscribes to a topic filter.

ZLINK_EXPORT zlink_config_result_t zlink_set_subscription (void *handle_, const char *filter_);

Registers filter_ in XSUB's subscription list and forwards a subscription message upstream. filter_ is a NUL-terminated string and cannot contain an embedded NUL. The bytes before the terminating NUL are the basis for the upstream XPUB's byte-prefix filtering. An empty string requests every message. There is no wildcard syntax, and a trailing * is a literal byte. XSUB itself does not use this filter to filter received messages. Registering the same filter more than once increments its reference count.

Applicable types: raw SUB, raw XSUB.

Returns: ZLINK_CONFIG_OK on success; otherwise a zlink_config_result_t value. zlink_errno() retains the detailed internal errno for diagnostics.

Errors: EFAULT if handle_ is NULL. EINVAL if filter_ is NULL or the handle type does not support subscriptions.

See also: zlink_unset_subscription, zlink_subscribe

zlink_unset_subscription¶

Unsubscribes from a topic filter.

ZLINK_EXPORT zlink_config_result_t zlink_unset_subscription (void *handle_, const char *filter_);

Removes a previously registered subscription. filter_ must be a NUL-terminated string without an embedded NUL. It uses the same byte-prefix interpretation as zlink_set_subscription(): the bytes before the terminating NUL must match the previously registered prefix. If the same filter was registered more than once, this function only decrements the reference count. It sends an upstream unsubscribe message only when removing the last registration. Removing an unregistered filter sends no upstream unsubscribe message.

Applicable types: raw SUB, raw XSUB.

Returns: ZLINK_CONFIG_OK on success; otherwise a zlink_config_result_t value. zlink_errno() retains the detailed internal errno for diagnostics.

Errors: EFAULT if handle_ is NULL. EINVAL if filter_ is NULL or the handle type does not support unsubscription.

See also: zlink_set_subscription

zlink_subscribe¶

Receives the topic and complete payload record from a raw XSUB socket.

ZLINK_EXPORT zlink_recv_result_t zlink_subscribe (
  void *sub_,
  const zlink_routing_id_t **source_rid_out_,
  char *topic_id_buf_, size_t topic_id_capacity_, size_t *topic_id_len_out_,
  zlink_msg_t *parts_out_, size_t parts_capacity_, size_t *part_count_out_,
  zlink_recv_flags_t flags_);

topic_id_len_out_, parts_out_, and part_count_out_ are required. The array slots need not be initialized. source_rid_out_ is optional and receives NULL on success for raw XSUB. On success, the function copies the binary topic bytes into the caller's buffer without a NUL and fills the array with the complete payload record. The caller closes the leading *part_count_out_ slots exactly once with zlink_multipart_close.

If topic_id_capacity_ is smaller than the topic length (a zero-length topic succeeds with capacity 0), the function writes the required topic length to *topic_id_len_out_ and returns ZLINK_RECV_BUFFER_TOO_SMALL with ENOBUFS. Core keeps that message's topic and payload internally, and leaves parts_out_ and every output other than topic_id_len_out_ unchanged. It also does not transfer slot ownership; calling again with a sufficient buffer returns the same retained message. If capacity is greater than zero but topic_id_buf_ is NULL, the function returns ZLINK_RECV_INVALID_HANDLE with EFAULT before inspecting or consuming the queue and leaves every output and parts_out_ unchanged.

If parts_capacity_ is smaller than the payload part count, the record is not consumed, the needed count is written to *part_count_out_, and the call returns ZLINK_RECV_BUFFER_TOO_SMALL with ENOBUFS. Other outputs and array slots are unchanged; retrying with a large enough array receives the same record. This function applies to raw SUB and raw XSUB.

zlink_subscription_at¶

Gets the subscription filter at the specified index.

ZLINK_EXPORT zlink_config_result_t zlink_subscription_at (void *handle_,
                           size_t index_,
                           char *filter_out_,
                           size_t *filter_len_inout_,
                           int *is_pattern_out_);

index_ is a zero-based index into a snapshot of the subscriptions at query time, sorted in lexicographically ascending order by the filter byte sequence; it is not the registration order. On success, filter_out_ contains only the filter bytes and no terminating NUL. This output is therefore not a C string. On entry, *filter_len_inout_ is the buffer size; on return, it is the filter length in bytes. is_pattern_out_ is an optional output that may be NULL. When it is not NULL, the function reports whether the filter is a pattern subscription. All raw subscriptions are byte-prefix filters, so it writes 0.

If the buffer is too small, the function writes the required length to *filter_len_inout_ and returns ZLINK_CONFIG_BUFFER_TOO_SMALL with errno == ENOBUFS. It writes no partial data to filter_out_ and leaves *is_pattern_out_ unchanged. It does not consume or modify the subscription inventory, so the caller can query the same index_ again with a sufficient buffer.

Applicable types: raw SUB, raw XSUB.

Returns: ZLINK_CONFIG_OK on success; otherwise a zlink_config_result_t value. zlink_errno() retains the detailed internal errno for diagnostics.

Errors: ENOENT if index_ is out of range. ENOBUFS if the buffer is too small. ENOTSUP if the handle type does not support subscription queries.

See also: zlink_set_subscription, zlink_get_sub_option

5. Receive Flow State¶

XSUB is not a socket type that supports receive flow. zlink_socket_set_receive_flow_state() returns ZLINK_CONFIG_NOT_SUPPORTED with errno == ENOTSUP for an XSUB socket and changes nothing. The byte HWM defined by Socket Common (the value that applies backpressure by limiting the bytes retained in a queue), low water mark, and transport backpressure remain in effect. An XSUB socket monitor does not set ZLINK_MONITOR_STATUS_DETAIL_FLOW_STATE and does not emit ZLINK_EVENT_SEND_FLOW_PAUSED, ZLINK_EVENT_SEND_FLOW_RESUMED, or ZLINK_EVENT_FLOW_STATE_STALE.

6. Implementation and Contract Test Verification Requirements¶

Verify the following through only the public surface (zlink_set_sub_option/zlink_get_sub_option, the subscription registration, removal, and query functions, zlink_subscribe, zlink_socket_set_receive_flow_state, return values, and errno). Each item maps to one unit test.

Options

Querying ZLINK_SUB_OPT_TOPICS_COUNT with zlink_get_sub_option returns the number of subscribed topics as an int (read-only).
Each function that returns zlink_config_result_t returns ZLINK_CONFIG_OK on success and a zlink_config_result_t value on failure; zlink_errno() retains the detailed internal errno for diagnostics.

Subscription Registration, Removal, and Delivery

zlink_set_subscription requests byte-prefix filtering by the upstream XPUB based on the bytes before the terminating NUL. An empty string requests every message, and a trailing * is a literal byte rather than a wildcard.
XSUB does not apply its own filter matching on receive. It delivers every message that actually arrives from upstream to the application regardless of the registered filters.
Registering a subscription on raw XSUB forwards a subscription message upstream. XPUB owns publisher-side filtering and observation.
Registering the same filter more than once increments its reference count. zlink_unset_subscription decrements the count and sends an upstream unsubscribe message only when removing the last registration. Removing an unregistered filter sends nothing upstream.
Both functions report EFAULT when handle_ is NULL and EINVAL when filter_ is NULL or the handle type does not support subscription or unsubscription, respectively.

Subscription Query

The zero-based index_ of zlink_subscription_at follows snapshot order sorted by the filter byte sequence, not registration order.
On success, filter_out_ contains only the filter bytes and no terminating NUL. *filter_len_inout_ is the byte length, and filter_out_ is not a C string.
is_pattern_out_ is an optional output that may be NULL. When provided, it receives 0 because all raw subscriptions are byte-prefix filters.
If the buffer is too small, the function writes the required length to *filter_len_inout_ and fails with ZLINK_CONFIG_BUFFER_TOO_SMALL and ENOBUFS. It writes no partial data to filter_out_, leaves *is_pattern_out_ unchanged, and does not consume or modify the subscription inventory, so the caller can query the same index_ again with a sufficient buffer.
An out-of-range index_ reports ENOENT; a handle type that does not support subscription queries reports ENOTSUP.

Topic and Payload-Record Receive

When zlink_subscribe succeeds, it copies the binary topic bytes into the caller's buffer without a NUL and fills the array with the complete payload record. The caller closes the leading *part_count_out_ slots exactly once with zlink_multipart_close. On raw XSUB, source_rid_out_ receives NULL on success.
If topic_id_capacity_ is smaller than the topic length (a zero-length topic succeeds with capacity 0), the function writes the required topic length to *topic_id_len_out_ and returns ZLINK_RECV_BUFFER_TOO_SMALL with ENOBUFS. Core retains that message's topic and payload internally, leaves parts_out_ and every output other than topic_id_len_out_ unchanged, and does not transfer slot ownership, so calling again with a sufficient buffer receives the same message.
If parts_capacity_ is smaller than the payload part count, the call writes the needed count to *part_count_out_ and returns ZLINK_RECV_BUFFER_TOO_SMALL with ENOBUFS. The record and other outputs remain unchanged, and retrying with a large enough array receives the same record.
A record whose topic frame is not followed by a payload part (the topic frame lacks MORE) returns ZLINK_RECV_INTERNAL_ERROR with EPROTO.
If capacity is greater than zero but topic_id_buf_ is NULL, the function returns ZLINK_RECV_INVALID_HANDLE with EFAULT before inspecting or consuming the queue and leaves every output and parts_out_ unchanged.
All payload parts of a multipart message are returned in array order in one call, with no partial-record state.

No Receive Flow State

zlink_socket_set_receive_flow_state() returns ZLINK_CONFIG_NOT_SUPPORTED with errno == ENOTSUP for an XSUB socket and changes nothing. The byte HWM, low water mark, and transport backpressure remain in effect.
An XSUB socket monitor does not set ZLINK_MONITOR_STATUS_DETAIL_FLOW_STATE and does not emit ZLINK_EVENT_SEND_FLOW_PAUSED, ZLINK_EVENT_SEND_FLOW_RESUMED, or ZLINK_EVENT_FLOW_STATE_STALE.

Socket Index | Previous: XPUB | Next: DEALER

Socket — XSUB¶

1. XSUB Overview¶

2. Subscription Behavior¶

3. Sub Options (zlink_sub_option_t)¶

4. Functions¶

zlink_set_sub_option¶

zlink_get_sub_option¶

zlink_set_subscription¶

zlink_unset_subscription¶

zlink_subscribe¶

zlink_subscription_at¶

5. Receive Flow State¶

6. Implementation and Contract Test Verification Requirements¶

3. Sub Options (`zlink_sub_option_t`)¶