データ・ストリームの追加

Oracle GoldenGateデータ・ストリームでは、ユーザー指定形式のデータへの直接アクセスは、単純なストリーミング・プロトコルに従った専用のWebソケット・チャネルを介して有効になります。

データ・ストリーム・プロトコルは、プッシュ・モードを使用してクライアントにデータを送信します。クライアントは、最初にHTTP RESTfulリクエストを介してサーバー上にストリーミング・リソースを作成します。ストリーミング・リソースが作成されると、クライアントはストリーミング・リソース・エンドポイントを介してWebSocket接続を確立します。WebSocketチャネルが確立された後、データ・ストリームは、クライアントからのレスポンスまたは確認を待たずに、ただちに継続的にデータのプッシュを開始します。

次のサンプルpythonクライアントは、クライアントとデータ・ストリーミング・サービス間の相互作用を示しています:

import asyncio
import requests
import websockets
import json

async def client():
    ### create the streaming resource
    payload = {"source":{"trail":"a1"}}
    response = requests.post(
  'http://name:pswd@localhost:9002/services/v2/stream/s1', json=payload)

### establish websocket connection and receive data continuously
uri = "ws://name:pswd@localhost:9002/services/v2/stream/s1?begin=earliest"
async with websockets.connect(uri) as websocket:
        while True:
            resp = await websocket.recv()
            records = json.loads(resp)
            for rec in records:
                print(rec)
asyncio.get_event_loop().run_until_complete(client())

指定されたクライアント・プログラムでは、データ・ストリーム・リソース・エンドポイントs1の作成時に、ソース・データ証跡名を指定する単純なデータ・ストリーム・ペイロードが提供されます。実際のアプリケーションでは、ストリーミング・プロトコルのハンドシェイク・フェーズで非常に複雑なデータ・ストリーム・ペイロードを使用して、データ・ストリーミング動作を構成できます。

たとえば、次のデータ・ストリーム・リクエスト・ペイロードでは、必要なデータ・ソース証跡名とともにフィルタリング・ルール、エンコーディング形式およびbufferSizeを指定します。

{   
    "$schema"    : "ogg:dataStream",
    "source"     : {"trail":"a1"},
    "rules"      : [{
        "action" : "exclude",
        "filter" : {
            "objectNames" : [ "PDB2.AR.*", "PDB1.U1.*" ]
        }
    }],
    "encoding"   : “json",
    "bufferSize" : 2097152
}

Webソケット接続の確立中に、クライアント側は、データのストリーミングを開始するための開始位置を(Webソケット接続URLの問合せパラメータとして)指定します。開始位置には、次のいずれかの値を指定できます。

各非メタデータLCRレコードには、不透明な位置(トランザクション内のCSN、XID、レコード番号を含む)が含まれます。クライアント側は、最後に処理されたLCRレコードの位置を維持します。データ・ストリーム・サービスは、指定された開始位置に基づいて正しい開始/再起動ポイントを特定します。

クライアントがデータ・ストリーム・サービスに初めて接続する場合、クライアントはストリーミング・データの開始場所のタイムスタンプを指定する必要があります。キーワードnowは現在のタイムスタンプに変換され、キーワードearliestはタイムスタンプ0に変換されます。

または、begin位置にはISO 8601タイムスタンプ文字列を使用できます。いずれの場合も、データ・ストリーム・サービスは、開始位置を決定するために、ソース証跡に対してタイムスタンプ・ベースの参照を実行します。

これがリカバリ/再起動の場合、クライアントはハンドシェイク中に、最後に処理された保存済位置をデータ・ストリーム・サービスに提供する必要があります。データ・ストリーム・サービスは、開始位置を決定するために、ソース証跡に対して位置ベースの参照を実行します。データ・ストリーミング・リカバリの動作は、データ・ストリームで指定されたQoSレベルにも依存します。

次のRest APIを使用して、GoldenGateデータ・ストリームを管理できます。

CloudEventsは、サービス、プラットフォームおよびシステム間の相互運用性を提供するために、共通の形式でイベント・データを記述するための仕様です。Oracle GoldenGateデータ・ストリームは現在JSONデータ・エンコーディングのみをサポートしているため、CloudEvents形式のサポートはJSONイベント形式に制限されます。CloudEventsのJSONイベント形式の完全な仕様は、次の場所にあります。

https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md

https://github.com/cloudevents/spec/blob/main/cloudevents/formats/json-format.md

https://github.com/cloudevents/spec/blob/main/cloudevents/formats/cloudevents.json

{
    "specversion" : "1.0",
    "type" : "com.example.someevent",
    "source" : "/mycontext",
    "id" : "A234-1234-1234",
    "datacontenttype" : "application/json",
    "data: {…}
}