XSUB 발행-구독¶

이 장의 계약 소유 문서 — PUB · SUB · XPUB · XSUB socket 스펙이 다룬다. 이 챕터는 그 계약을 언어별 예제로 보여준다.

1. 개요¶

발행-구독(Publish-Subscribe) 패턴은 메시지를 토픽 기반으로 분배한다. zlink는 기본 PUB/SUB과 고급 XPUB/XSUB 두 가지 레벨을 제공한다.

소켓	역할	특성
PUB	발행자	모든 구독자에게 브로드캐스트. 수신 불가.
SUB	구독자	토픽 prefix match 필터링. 송신 불가.
XPUB	고급 발행자	PUB + 구독 이벤트 수신 가능
XSUB	고급 구독자	로컬 필터링 없이 모든 메시지 수신. 프록시/중계용

유효한 소켓 조합: - PUB → SUB, PUB → XSUB - XPUB → SUB, XPUB → XSUB

SUB vs XSUB — 핵심 차이¶

SUB와 XSUB은 모두 zlink_set_subscription()으로 구독 정보를 upstream PUB에 전송한다. 공개 API 사용법은 같다. 차이는 로컬 필터 엔진의 on/off다.

	SUB (필터 적용)	XSUB (필터 미적용)
구독 있을 때	매칭되는 메시지만 수신	모든 메시지 수신 (필터 체크 안 함)
구독 없을 때	아무것도 수신하지 않음	모든 메시지 수신
`""` 빈 구독	모든 메시지 수신 (모든 토픽 매칭)	구독 없이도 이미 전부 수신
용도	일반 구독자	프록시/중계 (전체 스트림 통과)

흔한 혼동: "SUB에 "" 빈 구독을 넣으면 XSUB과 같지 않나?" → 모든 메시지를 받는다는 결과는 같지만, SUB은 매 메시지마다 로컬 byte-prefix 필터를 평가하고 XSUB은 필터 자체를 적용하지 않는다. 또 SUB은 구독이 없으면 아무것도 받지 못하지만, XSUB은 구독 없이도 전부 받는다.

프록시 패턴에서 XSUB/XPUB을 쓰는 이유:

%%{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은 자신의 필터를 적용하지 않고 PUB의 모든 메시지를 통과시킨다.
XPUB은 SUB의 구독 이벤트를 zlink_xpub_recv()로 노출해 프록시가 구독 관리 로직(필터링, 로깅, 인가 등)을 끼워 넣을 수 있다.
일반 SUB/PUB으로는 이 중계 구조를 만들 수 없다.

%%{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 기본 사용법¶

발행자 (PUB)¶

zlink_publish()는 PUB 또는 XPUB 소켓에서 payload part 배열 전체를 message record 하나로 발행한다.

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

/* Publish message -- HWM에 도달한 구독자에게는 drop된다 */
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);

구독자 (SUB)¶

zlink_subscribe()는 토픽과 payload record 전체를 한 번에 받는다. 콜백 표면은 없다. poller에서 ZLINK_POLLIN을 관찰한 뒤 이 함수로 꺼내 쓰는 것이 기본 패턴이다.

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);
}
/* 그 밖의 rc 값: ZLINK_RECV_NO_DATA (EAGAIN), TERMINATED, BUFFER_TOO_SMALL */

참고: core/tests/integration/test_pubsub.cpp — 빈 구독("") → 모든 메시지 수신

송수신 요약¶

소켓	방향	송수신 API	비고
PUB	송신 전용	`zlink_publish()`	수신 불가
SUB	수신 전용	`zlink_subscribe()`	토픽 + payload 배열 분리 반환
XPUB	양방향	`zlink_publish()` / `zlink_xpub_recv()`	구독 이벤트 수신
XSUB	수신 전용	`zlink_subscribe()`	필터 없이 전체 수신

참고: PUB/SUB 계열 4소켓에서 zlink_send()/zlink_recv()는 모두 ZLINK_SUBMIT_NOT_SUPPORTED / ZLINK_RECV_NOT_SUPPORTED 이다. 발행은 zlink_publish(), 수신은 zlink_subscribe()를 쓴다.

PUB/XPUB 기본값: ZLINK_PUB_OPT_NODROP 의 기본값은 0 이다. HWM 이 찼을 때 그 구독자에게 보내는 메시지를 조용히 drop 하고 zlink_publish() 는 성공을 반환한다. drop 대신 배압을 받아야 하면 ZLINK_PUB_OPT_NODROP 을 명시적으로 1 로 설정한다.

PUB의 송신 큐가 가득 차면(HWM) 기본값(ZLINK_PUB_OPT_NODROP=0)에서는 그 구독자에게 보내는 메시지를 조용히 drop한다. 상세는 성능 가이드를 참고.

3. 토픽 필터링¶

SUB 소켓의 토픽 필터링은 prefix match 방식이다.

구독 토픽	수신 메시지	매칭
`"weather"`	`"weather: sunny"`	O
`"weather"`	`"weathering storm"`	O
`"weather"`	`"sports: baseball"`	X
`""` (빈 문자열)	모든 메시지	O

다중 토픽 구독¶

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

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

빈 구독 (모든 메시지)¶

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

참고: core/tests/integration/test_pubsub.cpp — zlink_set_subscription(subscriber, "")

4. 메시지 형식¶

zlink_publish()는 토픽과 payload part 배열을 별도 파라미터로 받는다.

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_);

멀티파트 발행은 모든 part를 배열 순서대로 두고 zlink_publish()를 한 번 호출한다. Core는 배열 전체를 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);

/* SUB는 zlink_subscribe() 한 번으로 payload 배열 전체를 받는다:
   topic       = "sensor:cpu"
   parts[0]    = "host"
   parts[1]    = "73"
   part_count  = 2 */

토픽은 와이어(프로토콜 전송 레벨)에서 첫 프레임으로 전송되고, zlink_subscribe()가 토픽과 payload part를 분리해 반환한다. 호출자가 토픽 프레임을 직접 조립할 필요는 없다.

참고: zlink_publish(pub, NULL, &part, ...)처럼 topic_id_를 NULL로 전달하면 첫 메시지 프레임이 와이어 prefix 규칙에 따라 토픽으로 쓰이는 호환 경로가 동작하지만, 이 방식은 권장하지 않는다. 항상 topic_id_ 파라미터를 명시적으로 전달한다.

5. PUB/SUB 소켓 옵션¶

SUB 전용 함수¶

함수	설명
`zlink_set_subscription()`	토픽 구독 추가 (prefix match)
`zlink_unset_subscription()`	토픽 구독 해제

공통 옵션¶

옵션	타입	기본값	설명
`ZLINK_OPT_SNDHWM`	`uint64_t` bytes	자동 (auto-HWM이 profile/role 예산으로 계산)	PUB 계열. 연결 수가 늘면 같은 role budget 안에서 자동 조정되며 `0`은 무제한
`ZLINK_OPT_RCVHWM`	`uint64_t` bytes	자동 (auto-HWM이 profile/role 예산으로 계산)	SUB 계열. 연결 수가 늘면 같은 role budget 안에서 자동 조정되며 `0`은 무제한
`ZLINK_OPT_LINGER`	int	-1	close 시 대기 시간 (ms)

6. PUB/SUB 사용 패턴¶

패턴 1: 기본 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);

/* sub의 zlink_subscribe()가 record 도착 시 "test"를 반환한다 */

참고: core/tests/integration/test_pubsub.cpp — test_tcp()

패턴 2: 다중 SUB¶

하나의 PUB에 여러 SUB가 연결된다. 각 SUB는 자신의 토픽만 받는다.

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 */

패턴 3: 다중 PUB → SUB¶

SUB는 여러 PUB에 connect할 수 있다. Fair-queue로 모든 PUB의 메시지를 받는다.

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 주의사항¶

Slow Subscriber (HWM 처리)¶

PUB/XPUB는 기본적으로 손실 허용 모드로 동작한다 — ZLINK_PUB_OPT_NODROP의 기본값이 0이다. 느린 구독자의 송신 queue가 HWM(High-Water Mark, 보관할 수 있는 accounted byte 상한)에 도달하면 그 구독자에게 보내는 메시지를 오류 반환 없이 조용히 버리고 zlink_publish()는 성공을 반환한다. 나머지 구독자에 대한 전달은 영향을 받지 않는다.

/* 기본 동작 — HWM 도달 시 느린 구독자 몫만 drop, publish는 성공 */
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);

/* 버스트 손실을 줄이려면 HWM을 올린다 */
uint64_t hwm_bytes = 64 * 1024 * 1024;  /* HWM은 byte다 */
zlink_set_option(pub, ZLINK_OPT_SNDHWM, &hwm_bytes, sizeof(hwm_bytes));

NODROP 모드 — 버리는 대신 배압¶

ZLINK_PUB_OPT_NODROP을 1로 설정하면 HWM 도달 시 메시지를 버리지 않고 zlink_publish()가 ZLINK_SUBMIT_BACKPRESSURED를 반환해 호출자가 대응할 수 있다.

/* NODROP 모드 활성화 (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 도달 — 이 호출도 quote를 이미 소비했으므로,
       보관해 둔 record 전체를 새 배열로 다시 제출한다. */
}

이 모드는 publisher를 가장 느린 구독자에 묶는다. 한 pipe가 차면 같은 socket의 모든 구독자에 대한 전달이 멈춘다. 구독자 속도에 의존하면 안 되는 신뢰 전달은 PUB/SUB가 아니라 request-reply socket이 담당한다.

모드	HWM 도달 시 동작	사용 시점
기본 (`NODROP=0`, 손실 허용)	조용히 버림 — 오류 반환 없이 메시지 유실	fanout 일반 (관찰, 알림, 센서, 시세 데이터)
`NODROP=1`	`ZLINK_SUBMIT_BACKPRESSURED` 반환 — 호출자가 배압 제어	유실을 허용할 수 없고 느린 구독자에 묶여도 되는 경우

ZLINK_PUB_OPT_NODROP은 PUB·XPUB 양쪽에 적용된다(PUB은 XPUB 위에 구현됨).

Late Joiner (구독 전 메시지 유실)¶

SUB가 connect한 뒤 구독 정보가 PUB에 전파되기 전에 발행된 메시지는 받을 수 없다.

/* 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 */

방향 제약¶

PUB/SUB는 각각 전용 API만 쓸 수 있다:

/* PUB: zlink_publish()로만 송신. 수신 불가 */
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 */

/* PUB 에서 zlink_send() → ZLINK_SUBMIT_NOT_SUPPORTED 반환 */
zlink_send(pub, &parts[1], 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* SUB: zlink_subscribe() 로만 수신. 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 개요¶

XPUB/XSUB는 구독 프레임을 애플리케이션에서 직접 다룰 수 있는 고급 publish-subscribe 소켓이다. 프록시/브로커 구축, 구독 모니터링, Last-Value Caching에 쓴다.

SUB vs XSUB — 핵심 차이¶

항목	SUB	XSUB
토픽 등록	`zlink_set_subscription()`	`zlink_set_subscription()` (동일)
메시지 수신	`zlink_subscribe()` — 토픽 필터링 후 수신	`zlink_subscribe()` — 필터 없이 전체 수신
로컬 필터	켜짐 — 매칭 안 되면 드롭	꺼짐 — 모든 메시지 통과
구독 없는 상태	아무것도 수신 안 함	모든 메시지 수신

XSUB이 프록시에서 필요한 이유는 자신의 필터를 적용하지 않고도 모든 메시지를 통과시키기 때문이다. 토픽 등록은 zlink_set_subscription()으로 양쪽 모두 동일하게 upstream에 전송된다.

PUB vs XPUB — 핵심 차이¶

항목	PUB	XPUB
메시지 발행	`zlink_publish()`	`zlink_publish()` (동일)
구독 이벤트	노출 안 함	`zlink_xpub_recv()`로 수신

XPUB는 어떤 클라이언트가 어떤 토픽을 구독하거나 해지했는지 파악한다.

프록시에서 XSUB/XPUB의 역할¶

프록시는 두 개의 독립된 흐름을 갖는다.

%%{init: {'flowchart': {'nodeSpacing': 32, 'rankSpacing': 40, 'padding': 8, 'wrappingWidth': 180}, 'themeVariables': {'fontSize': '18px'}}}%%
flowchart LR
  subgraph data ["데이터 흐름 (publish)"]
    direction LR
    P1[PUB] -- publish --> X1[XSUB] == proxy forward ==> X2[XPUB] -- deliver --> S1[SUB]
  end
  subgraph sub ["구독 흐름 (역방향)"]
    direction RL
    S2[SUB] -. subscribe .-> X3[XPUB] -. propagate .-> X4[XSUB] -. register .-> P2[PUB]
  end

데이터 흐름¶

단계	주체	동작	비고
1	PUB	`zlink_publish(pub, topic, ...)`	데이터 발행
2	프록시 내부	XSUB 내부 recv → XPUB 내부 send	`zlink_proxy()`가 처리
3	SUB	`zlink_subscribe()`	최종 소비

핵심: zlink_proxy(xsub, xpub, NULL)의 데이터 릴레이는 내부 recv/send 경로를 쓰며, 공개 zlink_send()/zlink_recv() API를 쓰지 않는다. 사용자가 직접 XSUB recv → XPUB send를 호출할 필요는 없다.

구독 전파 흐름¶

단계	주체	동작	API
1	SUB	구독 → 와이어를 통해 XPUB에 도착	`zlink_set_subscription(sub, "weather")`
2	프록시 앱	XPUB에서 구독 이벤트 수신	`zlink_xpub_recv(xpub, ...)`
3	프록시 앱	XSUB에 등록 → 와이어를 통해 PUB에 전파	`zlink_set_subscription(xsub, "weather")`
4	PUB	매칭되는 데이터 발행	`zlink_publish(pub, "weather", ...)`
5	데이터 흐름	XSUB → XPUB → SUB	`zlink_proxy()`가 처리

zlink_set_subscription()은 SUB와 XSUB 모두 동일하게 구독 정보를 와이어로 전송한다. 프록시에서 XSUB에 이 함수를 호출하는 것은 "XSUB이 송신할 수 있어서"가 아니라 XPUB에서 받은 구독 이벤트를 XSUB에 등록해 upstream으로 전파하기 위해서다.

XSUB/XPUB이 필요한 이유¶

질문	SUB/PUB일 때	XSUB/XPUB일 때
데이터 통과	SUB 로컬 필터가 켜져 있어 구독을 등록해야 함	XSUB 로컬 필터가 꺼져 있어 모두 통과
구독 이벤트	PUB은 노출하지 않음	XPUB이 `zlink_xpub_recv()`로 노출
프록시 적합성	프록시가 직접 토픽을 관리해야 함	중계 전용 — 프록시에 최적

PUB/SUB 소켓 공개 API 요약¶

공개 API	PUB	SUB	XPUB	XSUB
`zlink_publish()`	가능	—	가능	—
`zlink_subscribe()`	—	가능	—	가능
`zlink_set_subscription()`	—	가능	—	가능
`zlink_xpub_recv()`	—	—	가능	—
로컬 필터	N/A	켜짐	N/A	꺼짐

zlink_send() / zlink_recv()는 PUB/SUB 계열 4소켓 모두 ZLINK_SUBMIT_NOT_SUPPORTED / ZLINK_RECV_NOT_SUPPORTED 이다. 발행은 zlink_publish(), 수신은 zlink_subscribe() 전용 API를 쓴다.

Proxy 패턴에서 XSUB/XPUB을 쓰는 방법은 Proxy 가이드를 참고.

9. 구독 프레임 형식¶

XPUB/XSUB 간의 구독/해제 프레임은 다음 형식을 따른다:

바이트	의미
`0x01` + topic	구독 요청
`0x00` + topic	구독 해제

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

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

XPUB는 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);

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

10. XPUB 소켓 옵션¶

옵션	타입	기본값	설명
`ZLINK_PUB_OPT_MANUAL`	int	0	수동 구독 관리 모드 활성화
`ZLINK_PUB_OPT_VERBOSE`	int	0	이미 구독 중인 topic의 중복 subscribe도 event로 공개
`ZLINK_PUB_OPT_APPROVE_SUBSCRIBE`	binary	--	(MANUAL 모드) 가장 최근에 꺼낸 구독 event의 pipe에 topic 구독을 승인
`ZLINK_PUB_OPT_REJECT_SUBSCRIBE`	binary	--	(MANUAL 모드) 가장 최근에 꺼낸 구독 event의 pipe에서 topic 구독을 거부

XPUB_MANUAL 모드¶

기본적으로 XPUB는 SUB의 구독을 자동 처리한다. MANUAL 모드에서는 zlink_xpub_recv()로 구독 event를 받은 뒤 애플리케이션이 zlink_set_pub_option()의 ZLINK_PUB_OPT_APPROVE_SUBSCRIBE / ZLINK_PUB_OPT_REJECT_SUBSCRIBE로 실제 구독을 결정한다. zlink_set_subscription() / zlink_unset_subscription()은 SUB·XSUB 전용이라 XPUB handle에서는 ZLINK_CONFIG_INVALID_ARGUMENT로 실패한다.

/* 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 */

참고: core/tests/integration/test_xpub_manual.cpp — test_basic(): A 구독 요청 → B로 변환

11. XPUB/XSUB 사용 패턴¶

패턴 1: 프록시/브로커 구축¶

XSUB(프론트엔드) + XPUB(백엔드)로 PUB/SUB 프록시를 구축한다.

/* 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);

패턴 2: MANUAL 모드 프록시 (구독 변환)¶

구독 요청을 변환하거나 필터링하는 고급 프록시.

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은 binary-safe라 NUL이 없다 — set_subscription은 C 문자열을 받는다 */

    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);
    }
}

참고: core/tests/integration/test_xpub_manual.cpp — test_xpub_proxy_unsubscribe_on_disconnect()

패턴 3: 구독 모니터링¶

XPUB로 어떤 클라이언트가 어떤 토픽을 구독하는지 관찰한다.

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);
}

패턴 4: 구독자 해제 시 자동 unsubscribe¶

SUB가 연결을 끊으면 XPUB에 자동으로 unsubscribe 프레임이 전달된다.

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

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

참고: core/tests/integration/test_xpub_manual.cpp — test_xpub_proxy_unsubscribe_on_disconnect()

12. 주의사항¶

구독 전파 타이밍¶

구독 메시지는 비동기로 전파된다. 구독 직후 발행된 메시지는 받지 못할 수 있다.

zlink_connect(sub, endpoint);
zlink_set_subscription(sub, "topic");
/* 이 시점에 "topic" 메시지를 발행하면 유실될 수 있다 */
msleep(100);  /* wait for subscription propagation */
/* 이 시점 이후에 발행된 메시지만 안정적으로 받는다 */

XPUB MANUAL 모드에서 구독 관리¶

MANUAL 모드에서 구독 event를 받은 뒤 ZLINK_PUB_OPT_APPROVE_SUBSCRIBE를 적용하지 않으면 그 구독은 등록되지 않는다. 반드시 명시적으로 구독을 처리해야 한다.

다중 구독자 → 단일 XPUB¶

여러 SUB가 같은 토픽을 구독하면 모든 SUB가 해제될 때까지 XPUB의 구독이 유지된다.

참고: core/tests/integration/test_xpub_manual.cpp — test_missing_subscriptions(): 두 구독자를 순차 처리하여 누락 방지

← PAIR | DEALER →

언어별 완전한 예제¶

PUB로 토픽을 발행하고 SUB로 구독·수신하는 자립형 예제다(모든 바인딩, 빌드·실행 검증됨).

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 →