JSONメッセージングのパブリッシュ/サブスクライブAPIサポート

概要

SAS Event Stream Processingでは、JSONメッセージングをサポートするCライブラリが提供されます。ライブラリは、バイナリ形式のイベントブロックとシリアライズされたJSONメッセージの間の変換メソッドを提供します。

JSONメッセージをイベントストリーム処理サーバーと交換するには、標準パブリッシュ/サブスクライブAPIを使用するパブリッシュ/サブスクライブクライアントは、C_dfESPpubsubInitJson()を呼び出してJSONをサポートするライブラリをロードできます。JSON形式のソースデータを持つパブリッシャークライアントは、C_dfESPjsonToEb()を呼び出して、バイナリ形式のイベントブロックを作成できます。その後、C_dfESPpublisherInject()を呼び出すことができます。

同様に、サブスクライバークライアントは、受け取ったイベントブロックをJSONメッセージハンドラに渡す前にC_dfESPebToJson()を呼び出してJSONメッセージに変換できます。

注: パブリッシュ/サブスクライブ接続を処理するイベントストリームのサーバー側では、JSONメッセージはサポートされません。イベントブロックはバイナリ形式でのみ送受信し続けます。

パブリッシュ/サブスクライブクライアント接続は、そのウィンドウのスキーマを使用して1つのウィンドウとイベントを交換します。これに対応して、JSON対応のクライアント接続は、固定されたJSONスキーマとメッセージを交換します。ただし、このスキーマの静的な定義はありません。JSONメッセージと関連ウィンドウスキーマの間のスキーマの不一致は、実行時にのみ検出されます。

C_dfESPpubsubInitJson()コールは、JSONオブジェクトインスタンスを返します。このインスタンスは、その後のすべてのJSON呼び出しでクライアントによって渡されます。このオブジェクトインスタンスは、特定のウィンドウへのクライアント接続が起動している間のみ有効です。クライアントが停止してRestartすると、新しいJSONオブジェクトインスタンスを取得する必要があります。

基本的に、単一のJSONメッセージは単一のイベントにマッピングされます。ただし、JSONメッセージ内に複数のイベントを含めることができます。

JSONメッセージのネストされたフィールドをイベントブロックに変換する

ウィンドウスキーマは、JSONイベントスキーマのフラット化された表現でなければなりません。ここで、ウィンドウフィールド名は、ネストされたJSONタグ名の連結をアンダースコアで区切ったものです。入力JSONに、ソースウィンドウスキーマに意図的に欠けているフィールドがあり、無視する必要がある場合は、イベントブロックを作成する前にC_dfESPignoreMissingSchemaFields()メソッドを呼び出します。これにより、ライブラリがERRORをログに記録することが防止されます。

たとえば、バイトオーダーのエンディアンが異なる場合でも、 JSONイベントスキーマ内では、配列とオブジェクトの無制限のネストがサポートされています。

JSONイベントに配列フィールドが含まれている場合、対応するESPウィンドウフィールドのタイプは次のとおりです。

ESP_UTF8STR: すべてのJSONタイプをサポートしています。これは、JSON配列を、配列の値の文字列バージョンを含む1つのコンマ区切りの文字列に変換します。
ESP_ARRAY_I32: JSONのBoolean配列のみをサポートしています。
ESP_ARRAY_I64: JSONの整数配列とJSONのブール配列をサポートしています。
ESP_ARRAY_DBL: JSONの実在する配列をサポートします。

JSONイベントから構築されたイベントには常にInsert演算コードがあります。例外は、JSONイベントにopcodeという名前のフィールドが含まれている場合です。その場合、そのフィールドの値はイベント演算コードを設定するために使用されます。他のJSONフィールドがソースウィンドウスキーマのフィールドと一致しない場合、注入操作は失敗します。

JSONイベントの演算コードフィールドの有効な値
値	演算コード
`i` `I`	Insert
`u` `U`	Update
`d` `D`	Delete
`p` `P`	Upsert
`s` `S`	Safe delete

デフォルトでは、イベントブロックはtype= normalです。イベントブロック配列内のイベントが追加の配列内に含まれている場合(つまり、追加のかっこで囲まれている場合)、イベントブロックはtype= transactional。

設定された部分文字列を含まないJSON値を除外するようにJSONライブラリを設定することができます。この場合、対応するイベントは静かにドロップされます。このようなフィルタリングを有効にするには、matchsubstringsパラメーターを設定し、カンマで区切られたkey:substringペアの文字列を指定します。たとえば、 "foo:bar"その値に "bar"を含まないJSONキー "foo"がイベントストリーム処理イベントを生成しないことを意味します。キーごとに複数の部分文字列を設定する場合、入力JSONデータには、イベントを生成するために構成された部分文字列のうちの1つのみが含まれている必要があります。

JSONフィールド(またはフィールドの組み合わせ)をソースウィンドウのキーとして簡単に使用できない場合は、次のキーフィールドのいずれかまたは両方をソースウィンドウスキーマに追加します。

"eventindex*:int64,adapterindex*:文字列"

ソースウィンドウスキーマに存在する場合、ライブラリは次のようにこれらのフィールドに入力します。

eventindex: イベントごとにインクリメントされる64ビットの整数値。プロセス空間内の複数のライブラリにまたがって一意であることが保証されています。
adapterindex: 異なるプロセス空間で実行されているライブラリのインスタンスに対して一意のGUID文字列。

これらの値を組み合わせることにより、異なるプロセスのライブラリの複数のユーザーが、キーを複製することなくイベントを単一のSourceウィンドウに挿入できます。

注入されたすべてのイベントに静的なJSON値を含める必要がある場合は、C_dfESPaddStaticJson() メソッドに渡します。後続のすべてのイベントには、そのJSONから構築されたフィールドが含まれます。

イベントブロックをJSONメッセージに変換する

イベントから作成されたすべてのJSONイベントには、opcodeフィールドが含まれています。有効な値は、JSONイベントの演算コードフィールドの有効な値にリストされています。

入力はイベントブロックなので、結果として得られるJSONメッセージは常にルートオブジェクトとして配列を持ちます。各配列エントリは単一のイベントを表します。

データ変数型STAMP、DATEおよびMONEYは、フィールド値のCSV表現を含むJSON文字列に変換されます。

JSONメッセージの転送のサポート

JSONメッセージングのサポートは、次のメッセージバスに関連付けられたコネクタとアダプターを使用すると利用できます。

IBM WebSphere MQ
RabbitMQ
Solace
Tibco/RV
Kafka

これらのコネクタとアダプターは、バイナリイベントブロックではなく、メッセージバスを介してJSONエンコードされたメッセージの転送をサポートします。これにより、サードパーティのパブリッシャーまたはサブスクライバーは、パブリッシュ/サブスクライブAPIを使用せずに、メッセージバスに接続し、JSONメッセージをエンジンと交換できます。

メッセージ形式構成は必要ありません。JSONスキーマとウィンドウスキーマが互換性がない場合、JSONをイベントブロックに変換するパブリッシャーは、イベントブロックが挿入されると失敗します。コネクタは、JSON変換をサポートするSAS Event Stream Processingライブラリを使用して、JSONをイベントブロックとの間で変換します。

RabbitMQ、Solace Systems、およびKafkaパブリッシュ/サブスクライブクライアントは、rabbitmq.cfg、solace.cfg、またはkafka.cfgクライアント設定ファイルでJSONをサポートするように設定されている場合、JSONをサポートします。JSON対応クライアントパブリッシャーは、イベントブロックをJSONメッセージに変換して、メッセージバスを介して、JSONメッセージのサードパーティコンシューマーに転送します。同様に、JSON対応クライアントのサブスクライバーは、メッセージバスからJSONメッセージを受け取り、それをイベントブロックに変換します。

最終更新: 2026年7月31日