English | 한국어

소켓 목차 | 이전: SUB | 다음: XSUB

Socket — XPUB

이 장이 정의하는 것 — XPUB socket(구독 이벤트를 message로 노출하는 PUB)의 공개 계약.

1. XPUB 개요

XPUB는 구독 전달과 수동 제어를 지원하는 확장 publisher다. message를 주고받는 endpoint인 socket 중 발행하는 쪽이라는 점은 PUB와 같지만, XPUB는 구독하는 쪽 peer(subscriber)가 보내는 구독·해제 요청을 구독 이벤트 message로 수신하고 수동 구독 관리를 지원한다. 기본 mode에서 event로 공개되는 것은 어떤 topic을 처음 요청한 subscribe와 그 topic을 요청하는 subscriber가 더는 남지 않게 한 마지막 unsubscribe뿐이다 — 이미 구독 중인 topic의 중복 subscribe나 다른 subscriber가 아직 남은 unsubscribe는 ZLINK_PUB_OPT_VERBOSE / ZLINK_PUB_OPT_VERBOSER를 켜야 event로 보인다(§4).

이 문서는 XPUB 고유의 공개 계약 — PUB/XPUB 전용 옵션, whole-message 발행, 구독 이벤트 수신 — 을 정의한다.

관련 계약의 소유 문서는 다음과 같다.

관련 계약	정의하는 문서
모든 socket 타입 공통 옵션·API 형태·수신 모델	Socket 공통
PUB socket의 계약	PUB
SUB·XSUB socket의 계약	SUB · XSUB
공개 result와 errno 대응	Errors

2. 구독 이벤트 흐름

subscriber가 보낸 구독·해제 요청은 XPUB에 구독 이벤트 message로 도착한다. application은 zlink_xpub_recv로 이 이벤트를 하나씩 꺼내며, 각 이벤트에서 어느 peer가 보냈는지(routing ID), 구독인지 해제인지, 어느 topic인지를 관찰한다. 구독 이벤트는 단일 frame이다 — 첫 byte가 subscribe면 0x01, unsubscribe면 0x00이고 나머지 byte가 topic(구독 대상을 구분하는 byte 열)이다. 빈 topic의 이벤트도 첫 byte 하나를 가진다. zlink_xpub_recv가 첫 byte를 *subscribed_out_로, 나머지를 topic output으로 나눠 돌려주므로 application이 frame을 직접 해석할 일은 없다.

수동 구독 관리는 ZLINK_PUB_OPT_MANUAL로 켠다. manual 모드의 구독 승인은 ZLINK_PUB_OPT_APPROVE_SUBSCRIBE, 거부는 ZLINK_PUB_OPT_REJECT_SUBSCRIBE 옵션으로 설정한다. 중복 구독·해제까지 이벤트로 공개하는 verbose 계열을 포함한 전체 옵션은 §4에 있다.

%%{init: {'sequence': {'actorFontSize': '18px', 'messageFontSize': '18px', 'noteFontSize': '18px', 'boxMargin': 8, 'width': 140}, 'themeVariables': {'fontSize': '18px'}}}%%
sequenceDiagram
    participant Sub as Subscriber
    participant XPub as XPUB socket
    participant App as Application
    Sub->>XPub: topic 구독 요청
    App->>XPub: zlink_xpub_recv()
    XPub-->>App: 구독 이벤트 (peer routing ID, subscribe=1, topic byte)
    App->>XPub: zlink_publish(topic, parts, part_count, flags)
    XPub-->>Sub: 발행 message 전달

3. HWM 도달 시 drop과 backpressure

ZLINK_PUB_OPT_NODROP의 기본값은 0이다. fanout 전달은 손실을 허용한다. 송신 queue가 유지할 byte 상한인 HWM(High-Water Mark)에 도달했을 때 zlink_publish()는 해당 subscriber에 대한 message를 버리고 성공을 보고한다. 송신 queue가 찼을 때 drop 대신 publisher의 추가 제출을 제한하는 backpressure를 주려면 명시적으로 1로 설정해야 하며, 이때 zlink_publish()는 ZLINK_SUBMIT_BACKPRESSURED를 반환한다. HWM 검사 대상은 현재 topic의 filter에 맞는 pipe뿐이다 — 그중 하나라도 차 있으면 그 record를 filter가 맞는 어떤 subscriber에도 전달하지 않고, filter가 맞지 않는 subscriber의 pipe 상태는 이 publish에 영향을 주지 않는다.

1로 설정하면 publisher가 가장 느린 subscriber에 묶인다. 한 pipe가 차면 같은 socket의 모든 subscriber에 대한 전달이 멈추기 때문이다. subscriber 속도에 의존하면 안 되는 신뢰 전달은 XPUB/XSUB가 아니라 request-reply socket이 담당한다.

4. Pub 옵션 (zlink_pub_option_t)

zlink_set_pub_option() / zlink_get_pub_option()과 함께 사용한다.

typedef enum zlink_pub_option_t
{
    ZLINK_PUB_OPT_VERBOSE = 0x3301,            // 이미 구독 중인 topic의 subscribe도 event로 공개 (int; 0=off, 양수=on (getter는 0/1 반환))
    ZLINK_PUB_OPT_VERBOSER = 0x3302,           // 중복 subscribe와 모든 unsubscribe를 event로 공개 (int; 0=off, 양수=on (getter는 0/1 반환))
    ZLINK_PUB_OPT_MANUAL = 0x3303,             // XPUB 수동 구독 관리 (int; 0=off, 양수=on (getter는 0/1 반환))
    ZLINK_PUB_OPT_MANUAL_LAST_VALUE = 0x3304,  // manual 모드 활성 + 다음 발행을 마지막 구독 event pipe에만 전달 (int; 0=off, 양수=on (getter는 0/1 반환))
    ZLINK_PUB_OPT_NODROP = 0x3305,             // HWM 시 drop 대신 EAGAIN 반환 (int; 0=off, 양수=on (getter는 0/1 반환), 기본값 0)
    ZLINK_PUB_OPT_WELCOME_MSG = 0x3306,        // 새 subscriber 연결 시 전송 message (binary)
    ZLINK_PUB_OPT_TOPICS_COUNT = 0x3307,       // 구독된 topic 수 (int, 읽기 전용)
    ZLINK_PUB_OPT_APPROVE_SUBSCRIBE = 0x3308,  // manual 모드 구독 승인 (binary)
    ZLINK_PUB_OPT_REJECT_SUBSCRIBE = 0x3309    // manual 모드 구독 거부 (binary)
} zlink_pub_option_t;

ZLINK_PUB_OPT_NODROP이 바꾸는 HWM 도달 시 동작은 §3이 설명한다.

기본 옵션 상태는 다음과 같다.

• VERBOSE, VERBOSER, MANUAL, MANUAL_LAST_VALUE는 모두 0이다.
• WELCOME_MSG는 빈 byte 열이며, 새 subscriber pipe에 welcome message를 전송하지 않는다.
• TOPICS_COUNT는 초기에 0이다.

ZLINK_PUB_OPT_MANUAL_LAST_VALUE를 활성화하면 manual 모드도 함께 활성화되고, 다음 발행은 마지막 구독 event를 낸 pipe에만 전달된다. 발행 message를 저장하는 latest-value cache를 활성화하는 옵션이 아니다.

5. 함수

zlink_set_pub_option

PUB/XPUB socket 전용 옵션을 설정한다.

ZLINK_EXPORT zlink_config_result_t zlink_set_pub_option (void *handle_,
                           zlink_pub_option_t option_,
                           const void *optval_,
                           size_t optvallen_);

PUB/XPUB socket 옵션을 설정한다. 모든 socket 타입에 공유되는 공통 옵션은 zlink_set_option()을 사용한다.

반환값: 성공 시 ZLINK_CONFIG_OK, 실패 시 zlink_config_result_t 값. zlink_errno()는 진단용 내부 errno를 그대로 유지한다.

참고: zlink_get_pub_option, zlink_set_option

---

zlink_get_pub_option

pub 전용 옵션을 조회한다.

ZLINK_EXPORT zlink_config_result_t zlink_get_pub_option (void *handle_,
                           zlink_pub_option_t option_,
                           void *optval_,
                           size_t *optvallen_);

PUB/XPUB socket 옵션의 현재 값을 가져온다.

반환값: 성공 시 ZLINK_CONFIG_OK, 실패 시 zlink_config_result_t 값. zlink_errno()는 진단용 내부 errno를 그대로 유지한다.

참고: zlink_set_pub_option

---

zlink_publish

raw XPUB socket에서 message record 하나를 발행한다.

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

topic_id_ == NULL이면 첫 message frame이 wire prefix 규칙에 따라 topic을 운반한다. NULL이 아니면 topic_id_는 NUL로 끝나며 내부 NUL이 없는 byte 문자열이어야 한다. 종료 NUL 앞의 모든 byte가 topic이며 Core가 이 byte를 message 앞의 topic frame으로 추가한다.

별도의 topic 전용 최대 길이는 없으며 topic byte도 message와 storage 크기 제한에 포함된다. 크기 제한을 넘으면 ZLINK_SUBMIT_INVALID_ARGUMENT와 EMSGSIZE, topic frame용 storage를 확보하지 못하면 ZLINK_SUBMIT_OUT_OF_MEMORY와 ENOMEM을 반환한다.

parts_ 배열과 part_count_가 publish record를 구성한다. part_count_는 양수여야 하며 0은 ZLINK_SUBMIT_INVALID_ARGUMENT+EINVAL이다. Part는 배열 순서대로 전달된다.

이 함수는 성공과 실패 모두에서 모든 입력 슬롯을 소비한다. 같은 record를 다시 사용할 가능성이 있으면 호출 전에 전체를 복사한다. 소비된 각 zlink_msg_t는 초기화된 빈 message로 남으므로 그대로 close하거나 다시 쓸 수 있다.

non-blocking 발행은 flags_에 ZLINK_DONTWAIT를 전달하며, 즉시 진행할 수 없으면 ZLINK_SUBMIT_BACKPRESSURED를 반환한다.

Core는 record 전체를 원자적으로 admission한다. 실패하면 subscriber에는 어느 part도 보이지 않으며, caller는 보관한 record 전체를 다시 제출한다. 이 계약은 PUB §3이 소유한다.

적용 타입은 raw PUB, raw XPUB다. 다른 타입은 ZLINK_SUBMIT_NOT_SUPPORTED, errno == ENOTSUP이다. 전체 결과 대응은 errno map을 따른다.

참고: zlink_xpub_recv

---

zlink_xpub_recv

XPUB socket에서 구독 이벤트를 수신한다.

ZLINK_EXPORT zlink_recv_result_t zlink_xpub_recv (
  void *xpub_,
  const zlink_routing_id_t **source_rid_out_,
  int *subscribed_out_,
  char *topic_id_buf_, size_t topic_id_capacity_, size_t *topic_id_len_out_,
  zlink_recv_flags_t flags_);

recv 모드에서 다음 구독 이벤트를 수신한다. subscribed_out_과 topic_id_len_out_은 필수 output이고, topic_id_capacity_가 0보다 크면 topic_id_buf_도 필수다 — 이 pointer 중 하나라도 NULL이면 event를 읽거나 output을 바꾸기 전에 ZLINK_RECV_INVALID_HANDLE과 EFAULT로 실패한다. source_rid_out_는 NULL을 허용하는 선택 output이다. NULL이 아니면, 아직 연결된 peer가 보낸 이벤트에서는 *source_rid_out_가 그 peer의 Core 소유 routing ID view로 설정되며 수명과 무효화 조건은 Socket 공통의 borrowed RID 규칙을 따른다. peer 연결이 끊겨 Core가 만든 unsubscribe 이벤트이거나, 이벤트를 꺼내기 전에 그 peer가 끊겼으면 *source_rid_out_는 NULL이다. *subscribed_out_는 subscribe이면 1, unsubscribe이면 0이다. topic_id_buf_ / *topic_id_len_out_에 topic byte가 기록된다(binary-safe).

다른 socket의 receive와 poller·completion·monitor 호출은 이 view의 수명에 영향을 주지 않는다. 값을 후속 호출 이후에도 보관하려면 반환 즉시 복사한다.

호출자는 topic_id_capacity_로 buffer 크기를 전달하며, topic buffer 부족의 결과는 Socket 공통의 typed receive buffer 계약을 따른다: 용량이 topic 길이보다 작으면 *topic_id_len_out_에 필요한 길이를 기록하고 ZLINK_RECV_BUFFER_TOO_SMALL과 ENOBUFS를 반환하며, event는 보존되어 충분한 buffer의 다음 수신이 같은 event를 한 번 반환한다.

적용 대상: raw XPUB만.

반환값: 성공 시 ZLINK_RECV_OK, 실패 시 zlink_recv_result_t 값. zlink_errno()는 진단용 내부 errno를 그대로 유지한다.

에러: xpub_, subscribed_out_, topic_id_len_out_가 NULL이거나 topic_id_capacity_ > 0인데 topic_id_buf_가 NULL이면 ZLINK_RECV_INVALID_HANDLE과 EFAULT. flags_에 ZLINK_RECV_FLAGS_DONTWAIT가 설정되고 이벤트가 없으면 ZLINK_RECV_NO_DATA와 EAGAIN. topic이 topic_id_capacity_보다 길면 ZLINK_RECV_BUFFER_TOO_SMALL과 ENOBUFS. subject가 XPUB가 아니면 ZLINK_RECV_NOT_SUPPORTED와 ENOTSUP.

참고: zlink_publish

---

6. Receive flow state

XPUB은 receive-flow 대상 socket type이 아니다. zlink_socket_set_receive_flow_state()는 XPUB socket에 대해 errno == ENOTSUP과 함께 ZLINK_CONFIG_NOT_SUPPORTED를 반환하고 아무것도 바꾸지 않는다. 기존 byte HWM, low water mark와 transport backpressure는 그대로 유지된다. XPUB socket의 monitor는 ZLINK_MONITOR_STATUS_DETAIL_FLOW_STATE를 설정하지 않고 ZLINK_EVENT_SEND_FLOW_PAUSED, ZLINK_EVENT_SEND_FLOW_RESUMED, ZLINK_EVENT_FLOW_STATE_STALE를 발생시키지 않는다.

7. 구현 및 contract test 검증 요구

공개 표면(zlink_set_pub_option·zlink_get_pub_option, zlink_publish, zlink_xpub_recv, zlink_socket_set_receive_flow_state, 반환값·errno)만으로 다음을 확인한다. 각 항목은 unit test 하나로 이어진다.

옵션 - zlink_set_pub_option·zlink_get_pub_option은 성공 시 ZLINK_CONFIG_OK, 실패 시 zlink_config_result_t 값을 반환하며 zlink_errno()는 진단용 내부 errno를 그대로 유지한다. - ZLINK_PUB_OPT_TOPICS_COUNT는 읽기 전용 int이며 조회 시 구독된 topic 수를 돌려준다. - 초기에 VERBOSE, VERBOSER, MANUAL, MANUAL_LAST_VALUE, TOPICS_COUNT는 0이고, 빈 WELCOME_MSG는 새 subscriber pipe에 message를 전송하지 않는다. - ZLINK_PUB_OPT_MANUAL_LAST_VALUE를 활성화하면 manual 모드가 활성화되고, 다음 발행은 마지막 구독 event pipe에만 전달된다.

구독 이벤트 수신 (zlink_xpub_recv) - raw XPUB에 구독 이벤트가 있으면 ZLINK_RECV_OK와 함께 *subscribed_out_(subscribe=1, unsubscribe=0), 구독 peer의 routing ID pointer, topic byte(binary-safe)가 관찰된다. - source_rid_out_은 NULL을 허용하는 선택 output이다. - 아직 연결된 peer의 이벤트에서 *source_rid_out_의 routing ID view의 수명과 무효화 조건은 Socket 공통의 borrowed RID 규칙을 따른다. - peer 연결이 끊겨 Core가 만든 unsubscribe 이벤트, 또는 꺼내기 전에 peer가 끊긴 이벤트는 ZLINK_RECV_OK이지만 *source_rid_out_ == NULL이다. - flags_에 ZLINK_RECV_FLAGS_DONTWAIT가 설정되고 이벤트가 없으면 ZLINK_RECV_NO_DATA와 EAGAIN이다. - topic이 topic_id_capacity_보다 길면 *topic_id_len_out_에 필요 길이를 기록하고 ZLINK_RECV_BUFFER_TOO_SMALL·ENOBUFS를 반환하며, event는 내부에 보관되어 충분한 buffer의 다음 수신이 같은 event를 한 번 반환한다. - subject가 raw XPUB가 아니면 ZLINK_RECV_NOT_SUPPORTED와 ENOTSUP이다. - xpub_가 NULL이면 EFAULT다.

발행과 topic (zlink_publish) - topic_id_가 NULL이 아니면 종료 NUL 앞의 byte가 topic frame으로 message 앞에 추가되고, NULL이면 첫 message frame이 wire prefix 규칙에 따라 topic을 운반한다. - topic byte를 포함한 크기 제한을 넘으면 ZLINK_SUBMIT_INVALID_ARGUMENT와 EMSGSIZE, topic frame용 storage를 확보하지 못하면 ZLINK_SUBMIT_OUT_OF_MEMORY와 ENOMEM이다. - raw PUB·raw XPUB가 아닌 타입에 호출하면 ZLINK_SUBMIT_NOT_SUPPORTED, errno == ENOTSUP이다. - 모든 parts_ 슬롯은 성공·실패 모두 소비되며 초기화된 빈 message로 남는다 — 그대로 close하거나 다시 쓸 수 있다. - ZLINK_DONTWAIT로 즉시 진행할 수 없으면 ZLINK_SUBMIT_BACKPRESSURED다.

Whole-message publish record - parts_ 배열의 part는 배열 순서대로 하나의 publish record로 전달된다. - Backpressure를 포함해 호출이 실패하면 subscriber에는 그 record의 어떤 part도 보이지 않고 모든 입력 슬롯은 소비되며, caller는 보관한 record 전체를 다시 제출한다(PUB §3).

HWM 도달 시 drop과 NODROP - ZLINK_PUB_OPT_NODROP이 기본값 0이면 HWM에 도달한 subscriber에 대한 message를 버리고 zlink_publish()는 성공을 보고한다. - 1로 설정하면 송신 queue가 찼을 때 zlink_publish()가 ZLINK_SUBMIT_BACKPRESSURED를 반환한다.

Receive flow state - zlink_socket_set_receive_flow_state()는 XPUB socket에 대해 ZLINK_CONFIG_NOT_SUPPORTED와 errno == ENOTSUP을 반환하고 아무것도 바꾸지 않는다. - XPUB socket의 monitor는 ZLINK_MONITOR_STATUS_DETAIL_FLOW_STATE를 설정하지 않고 ZLINK_EVENT_SEND_FLOW_PAUSED, ZLINK_EVENT_SEND_FLOW_RESUMED, ZLINK_EVENT_FLOW_STATE_STALE를 발생시키지 않는다.

소켓 목차 | 이전: SUB | 다음: XSUB