catalyst-datastreams

工作原理

1. 频道和访问控制只能在控制台中配置。 在 Console → Cloud Scale → Data Streams 中创建频道并选择其访问控制；SDK、REST 或 MCP 工具都无法创建频道。请向用户询问频道 ID 及其访问控制类型，或使用 getAllChannels() 列出频道（control_type 为 0 表示 User Roles，1 表示 Custom Logic）。加载 references/datastreams-basics.md，了解相关概念、令牌规则、订阅类型、确认机制和 REST API。

2. 根据访问控制选择令牌标识符，两者不能互换。

   • User Roles → userId = Catalyst Authentication 中某个应用用户的 ID。控制台管理员的用户 ID 会被拒绝；connectionName 会报错："Mandatory parameter app_user_id is missing."
   • Custom Logic → connectionName，且必须在被选为频道访问控制的 Advanced I/O 函数内部生成令牌。该函数就是令牌端点：它会运行检查，然后调用 getTokenPair。Catalyst 不会自行调用该函数。从其他函数或 REST 调用会报错："Invalid access control source!"。此类频道会签发用户 ID 令牌，但 WebSocket 会拒绝它们（3000）。

3. 服务端代码（发布、令牌对、读取频道）——加载运行时参考文档。

   • Node.js / JavaScript → references/datastreams-javascript-basics.md。使用应用初始化：new DataStreams(await zcAuth.init(req))。Advanced I/O 处理程序接收 Node 的原始 req/res：不能使用 req.body、res.status() 或 res.json()。
   • Python → references/datastreams-python-basics.md。需要 zcatalyst-sdk 1.5.0+（Python 3.10+）。使用 app = zcatalyst_sdk.initialize(req=request) 初始化，然后调用 app.datastreams()（复数）。向 get_token_pair 传入字符串形式的频道 ID，以及关键字参数 user_id= / connection_name=；结果的键名带连字符（token["wss-id"]）。订阅者在调用 connect() 之前不会执行任何操作。
   • Java → references/datastreams-java-basics.md。依赖项是来自 https://maven.zohodl.com 的 com.zoho.catalyst:java-sdk（2.5.0+）；不存在 datastreams 工件。
   • publishData、getAllChannels、getChannelDetails 和 getLiveCount 是管理员方法：仅限服务端使用，绝不可用于浏览器或移动端代码。浏览器 SDK 中也有 getTokenPair，但 Custom Logic 令牌必须来自已配置的函数。

4. 订阅者代码——正确读取令牌并始终确认消息。 令牌对的键名带连字符：将 token['wss-id'] 作为 zuid，并传入 token.url 和 token.key。仅在 open 事件之后订阅。每收到一条消息后都调用 sendAck()，否则不会传递下一条消息。消息包含 data（始终为字符串）和 streamingId（API 操作消息还包含 url/method）。重新连接时，先释放旧会话（旧令牌 → unsubscribe()），再获取新令牌，调用 subscribe('0')，并从自己的存储中重新加载状态；使用 '-2' 恢复已断开的会话会让遗漏的消息重复投递两次，并重放 48 小时的保留消息。每次主动关闭前都要取消订阅；随后出现代码为 1000 的 error 事件属于预期情况，因为这是在你调用 unsubscribe() 后发生的。在 Node 中，getLiveCount() 返回 { status, data: { connections } }。Java：subscribe() 会抛出受检异常 ZCDataStreamException；在 Catalyst 函数之外，应使用 new ZCDataStreamsWebSocket(config) 创建客户端，而不是使用 ZCDatastream.getInstance()。

5. 报告成功前进行端到端验证。 打开一个订阅者，检查 getLiveCount() 是否显示该订阅者，发布一条测试消息，并确认订阅者已收到消息。对于参考文档未涵盖的内容，请以 Markdown 格式获取 https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/ 下的页面（<page-url>/index.md）；如参考文档与之不一致，优先采用参考文档。

安全检查清单

• publishData 和读取通道的方法属于管理员操作：只能在 Catalyst 函数中运行，绝不能在浏览器或移动应用中运行。
• 保护令牌端点。任何能调用该端点的人都可以订阅通道，因此在签发令牌前，请先对调用方进行身份验证。
• 对于 Custom Logic 通道，配置的函数是唯一的把关方：在调用 getTokenPair 前，请在该函数中进行实际验证。
• 除了返回给请求令牌的客户端外，切勿在任何地方记录或返回令牌 key。

触发条件

适用于以下查询："Data Streams"、"Catalyst Data Streams"、"datastreams"、"实时通道"、"实时更新"、"向浏览器推送更新"、"发布/订阅"、"发布数据"、publishData、getTokenPair、"令牌对"、"wss-id"、DataStreamsWebSocket、ZCDataStreamsWebSocket、ZCDatastream、"WebSocket 订阅者"、sendAck、"订阅类型"、"从上一条消息恢复"、"实时计数"、getLiveCount、"通道访问控制"、"User Roles 通道"、"Custom Logic 通道"、"缺少 app_user_id"、"无效的访问控制来源"、"身份验证失败 3000"、"重新连接"、"重新连接后出现重复消息"、"DUPLICATE_VALUE"、@zcatalyst/datastreams、zcatalyst_sdk、get_token_pair、publish_data、send_ack 或 "Data Streams Python"。

参考资料

参考资料	在查询涉及以下内容时加载……
references/datastreams-basics.md	概念、通道、访问控制（User Roles 与 Custom Logic）、应使用哪种令牌标识符、令牌对结构、订阅类型、确认、保留期、消息字段、REST API 端点和作用域
references/datastreams-javascript-basics.md	Node.js 函数（初始化、发布、令牌端点、Custom Logic 函数、通道读取）、Node 和浏览器订阅者、软件包版本
references/datastreams-python-basics.md	Python 函数和订阅者：zcatalyst-sdk 1.5.0+、app.datastreams()、get_token_pair 参数规则、带有 connect() 和 set_on_* 回调的 DataStreamsWebSocket、部署 Python 函数
references/datastreams-java-basics.md	Java 函数和订阅者：Maven 配置、ZCDatastream、ZCDataStreamsWebSocket、受检异常、在 Catalyst 外部运行