高级发布订阅

高级发布订阅（Advanced Pub/Sub）API 在普通发布 / 订阅之上提供发布端缓存、丢包检测与历史恢复，适用于对数据完整性要求较高的场景（晚加入的订阅者补齐历史、链路抖动自动重传丢失样本等）。

可用性：完整版（Full）专属；精简版（Tiny）头文件中相关声明被剔除，误用会在编译期报错。 完整的概念说明、可靠传输语义与典型用法见专题指南 高级发布订阅。

发布者与订阅者的句柄类型（onepath_publisher_t / onepath_subscriber_t）与普通版本相同，但需用本页的对应高级 API 声明与发布 / 订阅。销毁时仍用普通版本的 onepath_publisher_destroy / onepath_subscriber_destroy（见 发布与订阅）。

选项

onepath_advanced_pub_opts_t

typedef struct {
    int    cache_enabled;          /* 启用发布端缓存 */
    size_t cache_max_samples;      /* 最大缓存条数 (0 = 不限) */
    int    miss_detection_enabled; /* 启用基于心跳的丢包检测 */
    int    publisher_detection;    /* 通过存活机制通告发布者存在 */
} onepath_advanced_pub_opts_t;

高级发布者选项，传给 onepath_declare_advanced_publisher。

字段	类型	含义
cache_enabled	int	是否启用发布端缓存（晚加入的订阅者据此获取历史）
cache_max_samples	size_t	最大缓存样本条数；0 = 不限
miss_detection_enabled	int	是否启用基于心跳的丢包检测
publisher_detection	int	是否通过存活机制向网络通告本发布者存在

默认初始化宏：

#define ONEPATH_ADVANCED_PUB_OPTS_DEFAULT { 1, 10, 1, 1 }

即默认启用缓存、缓存最多 10 条、启用丢包检测、通告发布者存在。

onepath_advanced_sub_opts_t

typedef struct {
    int    history_enabled;        /* 加入时获取发布端缓存的历史消息 */
    int    detect_late_publishers; /* 检测后上线的发布者 */
    int    recovery_enabled;       /* 自动恢复丢失的消息 */
} onepath_advanced_sub_opts_t;

高级订阅者选项，传给 onepath_subscribe_advanced。

字段	类型	含义
history_enabled	int	加入时是否获取发布端缓存的历史消息
detect_late_publishers	int	是否检测后续上线的发布者
recovery_enabled	int	是否自动恢复检测到的丢失消息

默认初始化宏：

#define ONEPATH_ADVANCED_SUB_OPTS_DEFAULT { 1, 1, 1 }

即默认加入时拉取历史、检测后上线发布者、自动恢复丢失消息。

函数

高级发布者

onepath_declare_advanced_publisher

int onepath_declare_advanced_publisher(onepath_session_t s,
                                       onepath_publisher_t *out,
                                       const char *sign,
                                       const onepath_advanced_pub_opts_t *opts);

声明高级发布者，支持发布端缓存、丢包检测和存活通告。返回的句柄可用 onepath_advanced_publisher_put 发布，也可用 onepath_publisher_put_shm 零拷贝发布。

• 参数：
  • s — 会话句柄
  • out — 成功时写入发布者句柄
  • sign — 路标（sign）字符串
  • opts — 高级选项，传 NULL 使用默认值（等价于 ONEPATH_ADVANCED_PUB_OPTS_DEFAULT）
• 返回值：ONEPATH_OK 成功
• 注意：销毁时调用 onepath_publisher_destroy()（见 发布与订阅）

onepath_advanced_publisher_put

int onepath_advanced_publisher_put(onepath_publisher_t pub,
                                   const void *data, size_t len);

通过高级发布者发送二进制数据。

• 参数：pub — 高级发布者句柄；data — 数据指针；len — 数据长度（字节）
• 返回值：ONEPATH_OK 成功

onepath_advanced_publisher_put_str

int onepath_advanced_publisher_put_str(onepath_publisher_t pub,
                                       const char *str);

通过高级发布者发送以 null 结尾的字符串。

• 参数：pub — 高级发布者句柄；str — 以 null 结尾的字符串
• 返回值：ONEPATH_OK 成功

高级订阅者

onepath_subscribe_advanced

int onepath_subscribe_advanced(onepath_session_t s,
                               onepath_subscriber_t *out,
                               const char *sign,
                               onepath_sample_cb cb,
                               void *userdata,
                               const onepath_advanced_sub_opts_t *opts);

声明高级订阅者，支持历史回放、后上线发布者检测和自动丢包恢复。

• 参数：
  • s — 会话句柄
  • out — 成功时写入订阅者句柄
  • sign — 路标（sign）字符串，支持通配符
  • cb — 数据到达时的回调函数（签名见 回调函数 中的 onepath_sample_cb）
  • userdata — 传递给回调的用户上下文，可为 NULL
  • opts — 高级选项，传 NULL 使用默认值（等价于 ONEPATH_ADVANCED_SUB_OPTS_DEFAULT）
• 返回值：ONEPATH_OK 成功
• 注意：回调在内部线程中触发；销毁时调用 onepath_subscriber_destroy()（见 发布与订阅）

丢失检测

onepath_subscriber_on_miss

int onepath_subscriber_on_miss(onepath_subscriber_t sub,
                               onepath_miss_cb cb,
                               void *userdata);

注册丢包通知监听器。仅对高级订阅者（经 onepath_subscribe_advanced 创建）有效，检测到丢包时调用 cb。

• 参数：
  • sub — 高级订阅者句柄
  • cb — 丢包回调函数（签名见 回调 中的 onepath_miss_cb）
  • userdata — 传递给回调的用户上下文，可为 NULL
• 返回值：ONEPATH_OK 成功，ONEPATH_ERR_PARAM（非高级订阅者）

回调

onepath_miss_cb

typedef void (*onepath_miss_cb)(unsigned int missed_count, void *userdata);

丢包通知回调。由 onepath_subscriber_on_miss 注册，当订阅端检测到样本丢失时由 OnePath 在内部线程触发。

• 参数：
  • missed_count — 本次检测到的丢失消息数量
  • userdata — 注册时传入的用户上下文指针

static void on_miss(unsigned int missed_count, void *userdata) {
    (void)userdata;
    printf("miss detected: %u messages lost, recovering ...\n", missed_count);
}

数据样本回调使用普通版本的 onepath_sample_cb（签名 (onepath_sample_t *sample, void *userdata)），见 回调函数。

内存与所有权

创建 / 获取	销毁	所有权说明
onepath_declare_advanced_publisher	onepath_publisher_destroy	发布者句柄归调用者所有
onepath_subscribe_advanced	onepath_subscriber_destroy	订阅者句柄归调用者所有

丢失检测依赖缓存与心跳

若需要可靠的「不丢消息」语义，发布端应保持 cache_enabled=1 并设置足够的 cache_max_samples，订阅端保持 recovery_enabled=1。样本回调收到的数据已深拷贝、归用户所有，用完须调用 onepath_sample_release()，见 内存管理。

相关指南

• 高级发布订阅 — 概念说明、可靠传输语义、典型用法与示例