データ・ストリームへの接続とデータの消費

次のステップを使用して、データ・ストリームに接続し、データを消費します:

ステップ1: WebSocketのURLの作成

生成されたAsyncAPI YAMLから次の値を読み取ります。

servers.<name>.protocol: wsまたはwss。
servers.<name>.host: ホスト名および存在する場合はポート。
channels.<name>.address: データ・ストリームのWebSocketパスを完了します。

wss://<server-host><channel-address>?begin=<percent-encoded-begin-value>

パスをハードコードしないでください: OCI GoldenGate、オンプレミス・デプロイメントおよびリバース・プロキシでは、異なるルーティング接頭辞を公開できます。生成されたYAMLまたはREST APIによって返された正規リンクからホストおよびチャネル・アドレスをコピーします。

ステップ2: 開始値の選択

次の表に示すガイドラインに基づいて、BEGIN値を指定します。

値	使用	ガイドライン
`now`	新規アクティビティから開始	現在のタイムスタンプに変換され、使用可能な証跡に対して解決されます。変換された点数より前の変更は、送信されません。
`earliest`	使用可能な証跡からの初期バックフィル	タイムスタンプ0に変換されました。配信は、ソース履歴の開始ではなく、構成済の証跡で引き続き使用可能な最も早いレコードから開始されます。
ISO 8601タイムスタンプ	ビジネス時間の境界付近から開始	サービスはタイムスタンプベースの参照を実行します。明示的なタイムゾーン(`2026-07-13T04:00:00Z`など)を使用します。
保存済`pos`	前回の正常な実行後のリカバリ	サービスは、ポジション・ベースの参照を実行します。ポジションが使用できない場合の結果は、サービスの質によって異なります。

開始値のパーセントエンコード

posトークンには、URLで特別な意味を持つ+、/、=などの文字を含めることができます。完全なトークンをクエリー文字列に追加する前に、そのトークンをパーセントエンコードします。アプリケーション・ロジックでトークンを変更またはデコードしないでください。

ステップ3: 認証とTLS

認証方法を選択します。次の表を使用して、最適な方法を決定します。

モード	使用するタイミング	クライアント応答
基本的な認証	直接公開されたサービスまたはリバース・プロキシに共通	`Authorization: Basic ...`ヘッダーを送信します。資格証明は、URLまたはソース・コードではなく、シークレット・ストアまたは環境インジェクションに保持します。
Digestの認証	リバース・プロキシで必要な場合	WebSocketハンドシェイク中にHTTPダイジェスト・チャレンジを実装するクライアントまたは生成されたテンプレートを使用します。
クライアント証明書	相互TLSまたは証明書ベースのサービス認証	クライアント証明書および秘密キーをTLSコンテキストにロードし、構成されたCAに対してサーバーを検証します。
セキュアなリバース・プロキシ	ゲートウェイはTLSを終了または転送します	プロキシ認証モードに従います。セキュア・プロキシには、基本/ダイジェスト資格証明とクライアント証明書の両方が必要な場合があります。

ステップ4: AsyncAPIからクライアントを生成

Oracleには、AsyncAPIジェネレータ用のWebSocketクライアント・テンプレートが用意されています。JavaScript、Python、Java、Go、またはサポートされているすべてのクライアントを生成できます。ジェネレータには、Node.js 18.12以降およびnpm 8.19以降が必要です。

npm install -g @asyncapi/generator

git clone https://github.com/oracle-samples/websocket-client-template.git
cd websocket-client-template

ag /path/to/sales-orders.yaml .
  -o output
  -p server=production
  -p language=python
  -p authorization=basic

cd output
python -m pip install -r requirements.txt
export ASYNCAPI_WS_CLIENT_USERNAME="$OGG_USER"
export ASYNCAPI_WS_CLIENT_PASSWORD="$OGG_PASSWORD"
python client.py

テンプレートでは、証明書変数ASYNCAPI_WS_CLIENT_CERT、ASYNCAPI_WS_CLIENT_KEYおよびASYNCAPI_WS_CA_CERTもサポートされます。組織の通常のセキュアなソフトウェア開発プロセスで、生成されたコードおよび依存関係のバージョンを確認します。

ステップ5: WebSocketメッセージの理解

通常のメッセージ・ペイロードはJSON配列です。フレームには、1つ以上のスキーマ、メタデータ、DMLまたはDDLレコードを含めることができます。クライアントは配列を反復処理し、各オブジェクトを個別に分類する必要があります。

Backpressure:サービスは、レコードごとのクライアント確認応答なしでレコードをプッシュします。receiveループを軽量に保ちます。制限された内部キューまたは永続ステージング・レイヤーの後ろに低速なビジネス処理を置き、キューの深さを監視し、メモリーを使い果たす前にコンシューマを停止またはスケーリングします。

ステップ6: コンシューマ処理ループ

完全なWebSocketメッセージを受信します。
JSON配列を解析します。CloudEventsが有効な場合は、dataメンバーをアンラップします。
スキーマ、ストリーム・メタデータ、オブジェクト・メタデータ、DMLおよびDDLレコードを分類します。
検証が必要な場合は、最新のキャッシュされたスキーマに対してレコードを検証します。
ビジネス操作を適用するか、永続ダウンストリーム・システムに公開します。
操作が成功したら、選択したアトミック・チェックポイント設計または冪等チェックポイント設計を使用してレコードposを保存します。
接続が失われた場合は、最後の永続posおよび指数関数的なバックオフに再接続します。