catalyst-datastreams

仕組み

1. チャネルの作成とアクセス制御はコンソールでのみ行えます。 Console → Cloud Scale → Data Streams でチャネルを作成し、アクセス制御を選択してください。SDK、REST、MCP ツールからチャネルを作成することはできません。ユーザーにチャネル ID とアクセス制御タイプを確認するか、getAllChannels() でチャネルを一覧表示してください（control_type 0 = User Roles、1 = Custom Logic）。概念、トークンのルール、サブスクライブタイプ、確認応答、REST API については references/datastreams-basics.md を読み込んでください。

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' で再開すると、未受信メッセージが 2 回再配信され、48 時間分の保持データが再生されます。意図的に接続を閉じる前には必ずサブスクライブを解除してください。自身の unsubscribe() の後に発生する、コード 1000 の error イベントは想定どおりです。Node では getLiveCount() は { status, data: { connections } } を返します。Java では subscribe() はチェック例外の ZCDataStreamException をスローします。また、Catalyst 関数の外でクライアントを作成する場合は、ZCDatastream.getInstance() ではなく new ZCDataStreamsWebSocket(config) を使ってください。

5. 成功を報告する前に、エンドツーエンドで検証します。 サブスクライバーを開き、getLiveCount() に表示されることを確認してから、テストメッセージを公開し、サブスクライバーが受信したことを確認してください。リファレンスに記載されていない内容については、https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/ 配下のページを Markdown として取得してください（<page-url>/index.md）。内容が食い違う場合は、リファレンスを優先してください。

セキュリティチェックリスト

• publishData とチャネル読み取りメソッドは管理者向け操作です。Catalyst 関数内でのみ実行し、ブラウザーやモバイルアプリでは決して実行しないでください。
• トークンエンドポイントを保護してください。これを呼び出せる人は誰でもチャネルをサブスクライブできるため、トークンを発行する前に呼び出し元を認証してください。
• Custom Logic チャネルでは、設定された関数だけが唯一のゲートキーパーです。getTokenPair を呼び出す前に、そこで実際の検証を行ってください。
• トークンの key は、要求したクライアント以外にログ出力したり返したりしないでください。

トリガー

次のような内容には、このスキルを使用してください：「Data Streams」、「Catalyst Data Streams」、「datastreams」、「real-time channel」、「live updates」、「push updates to the browser」、「pub/sub」、「publish data」、publishData、getTokenPair、「token pair」、「wss-id」、DataStreamsWebSocket、ZCDataStreamsWebSocket、ZCDatastream、「WebSocket subscriber」、sendAck、「subscribe type」、「resume from last message」、「live count」、getLiveCount、「channel access control」、「User Roles channel」、「Custom Logic channel」、「app_user_id is missing」、「Invalid access control source」、「Authentication failed 3000」、「reconnect」、「duplicate messages after reconnect」、「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 外部での実行