Oracle® Fusion Middleware Oracle Stream Analyticsによるイベント処理用アプリケーションの開発 12c (12.2.1.1.0) E77666-02 |
|
![]() 前へ |
![]() 次へ |
Oracle Stream Analyticsには、対象と方法に応じて、アプリケーションをテストする様々な方法があります。
この章の内容は次のとおりです。
ロード・ジェネレータ・ユーティリティはデータ・フィードをシミュレートし、実際のデータ・フィードに接続せずにアプリケーションをテストできます。ロード・ジェネレータでアプリケーションをテストするには、アプリケーションでcsvgenアダプタを使用する必要があります。csvgenアダプタは、ロード・ジェネレータで生成されたデータ・パケットを復号化するよう特別にコーディングされているためです。
csvgenアダプタでアプリケーションのテストが終了したら、アダプタで説明しているように、アプリケーションに適した入力アダプタに置き換えることができます。
ロード・ジェネレータは、サンプル・データ・フィード情報を含むASCIIファイルを読み取り、ポート順に各行のデータを送信します。csvgenアダプタは同じポートでデータをリスニングします。
ロード・ジェネレータのプロパティ・ファイルに、データ・ファイルの名前、サーバーがリスニングするポート、サーバー・ホストおよびパケット・タイプを指定します。データ速度と、ロード・ジェネレータが指定した最終的な速度に達するまでの時間も設定できます。
次の手順では、ロード・ジェネレータ・ユーティリティの構成および実行方法の概要を示します。その後、各ステップの詳細を説明します。
ロード・ジェネレータの特定の実行に関する構成プロパティを含むプロパティ・ファイルを作成します。Oracle Stream Analyticsにはデフォルトのプロパティ値が適切である場合に使用できる、デフォルトのプロパティ・ファイルが用意されています。プロパティ・ファイルの作成を参照してください。
実際のデータ・フィード値が含まれたファイルを作成します。データ・フィード・ファイルの作成を参照してください。
ロード・ジェネレータで生成されたデータ・フィードが正常に読み込まれるようにcsvgenアダプタを構成します。アプリケーションのcsvgenアダプタの構成を参照してください。
ロード・ジェネレータを実行してステップ1で作成したプロパティ・ファイルを指定し、シミュレートされたデータ・フィードを開始します。たとえば、プロパティ・ファイル名がc:\loadgen\myDataFeed.prop
の場合は、次のコマンドを実行します。
prompt> runloadgen.cmd c:\loadgen\myDataFeed.prop
アプリケーションを再デプロイした場合は、ロード・ジェネレータも再起動する必要があります。
ジェネレータを停止およびロードするには、ロード・ジェネレータが実行されているディレクトリに移動し、ctrl-c
と入力します。
Oracle Stream Analyticsには、csvgen.prop
という名前のデフォルトのプロパティ・ファイルが、/Oracle/Middleware/my_oep/
utils/load-generator
ディレクトリに用意されています。ファイルの形式は単純で、プロパティと値の各ペアが各行に置かれます。
次の例は、デフォルトのcsvgen.prop
ファイルです。このファイルを独自のプロパティ・ファイルのテンプレートとして使用することをお薦めします。
# name of file containing your test data test.csvDataFile=test.csv # port the server will listen on for client connection test.port=9001 # server host (localhost if not specified) # test.host= # do not change the packetType
test.packetType=CSV
表14-1 ロード・ジェネレータのプロパティ
|
ロード・ジェネレータ・データ・フィード・ファイルは、Oracle Stream Analyticsアプリケーションに登録されたイベント・タイプに対応するサンプルのデータ・フィード値を含みます。
次の例は、EmployeeEvent
およびこのイベント・タイプに対応するロード・ジェネレータのデータ・フィード・ファイルを示します。
<wlevs:event-type-repository> <wlevs:event-type type-name="EmployeeEvent"> <wlevs:properties> <wlevs:property name="name" type="char" /> <wlevs:property name="age" type="int" /> <wlevs:property name="birthplace" type="char" length="512" /> </wlevs:properties> </wlevs:event-type> ... </wlevs:event-type-repository>
Lucy,23,Madagascar Nick,44,Canada Amanda,12,Malaysia Juliet,43,Spain Horatio,80,Argentina
ロード・ジェネレータのデータ・フィード・ファイルの形式は簡単です。
データ・フィードの各項目を各行に配置します。
データ・フィード項目の各フィールドをカンマで区切ります。
文字列フィールドにカンマは含めないでください。
フィールド値の一部としてスペースが含まれているのでない限り、カンマの前後に余分なスペースを含めないでください。
整数、long、doubleおよびfloatなどの、データ・フィード・ファイルの文字列データおよび数値データのみを含みます。
文字列の長さを最大256文字までに収めるか、長い文字列長を指定します。長い文字列を指定するには、event-type内のbirthplace
プロパティについてchar
プロパティのlength
属性を設定します。
CSVアダプタの制約の詳細は、設計制約を参照してください。
ロード・ジェネレータ・ユーティリティを使用する場合、Oracle Stream Analyticsで提供されるcsvgen
アダプタは、ロード・ジェネレータで生成されたデータ・パケットを読み込むように特別にコーディングされているため、アプリケーションではこのアダプタを使用する必要があります。
すべてのアダプタと同じように、アプリケーションのEPNアセンブリ・ファイルのwlevs:adapter
要素を使用して、csvgen
アダプタを登録します。独自のアダプタのかわりにcsvgen
アダプタをプロバイダとして指定するために、provide
属性をcsvgen
に設定します。また、次の子タグを指定する必要があります。
name
属性port
とvalue
属性の
configured_portがある
wlevs:instance-property要素。configured_port
はロード・ジェネレータ・プロパティ・ファイルのtest.port
プロパティに対応します。プロパティ・ファイルの作成を参照してください。
name
属性のeventTypeName
およびvalue
属性のevent_type_name
のある
wlevs:instance-property要素。event_type_name
はload-generatedフィードの項目を表すイベント・タイプ名に対応します。
name
属性のeventPropertyNames
およびvalue
属性のordered_list_of_properties
のある
wlevs:instance-property要素。ordered_list_of_properties
は、ロード・ジェネレータで送信されてcsvgen
アダプタで受信される順番でプロパティ名をリストします。
アダプタの構成する方法を説明する前に、まず次に示すwlevs:metada
要素を使用して、アプリケーションがEPNアセンブリ・ファイルに、PersonType
と呼ばれるイベント・タイプを登録していると仮定します。
<wlevs:event-type-repository>
<wlevs:event-type type-name="PersonType">
<wlevs:properties>
<wlevs:property name="name
" type="char"/>
<<wlevs:property name="age" type="int"/>
<<wlevs:property name="birthplace" type="char"/>
</wlevs:properties>
</wlevs:event-type>
</wlevs:event-type-repository>
このイベント・タイプはデータ・フィード・ファイルの作成で説明しているデータ・フィード・ファイルに対応しています。
このデータを受信するcsvgen
アダプタを構成するには、次のwlevs:adapter
要素を使用します。
<wlevs:adapter id="csvgenAdapter" provider="csvgen"> <wlevs:instance-property name="port" value="9001"/> <wlevs:instance-property name="eventTypeName" value="PersonType"/> <wlevs:instance-property name="eventPropertyNames" value="name,age,birthplace"/> </wlevs:adapter>
アダプタ構成の例に示す太字部分の値がPersonType
イベント・タイプの登録にどのように対応しているかに注意してください。
イベント・タイプの登録でwlevs:class
要素を使用して独自のJavaBeanを指定した場合、eventPropertyNames
値はJavaBeanのプロパティに対応します。たとえば、JavaBeanにgetName
メソッドが含まれている場合、JavaBeanのプロパティの中の1つはname
になります。
イベント・タイプの詳細は、イベントおよびイベント・タイプを参照してください。
開発中にOracle CQL問合せのテストやデバッグを行うためにイベント・インスペクタ・サービスを使用します。イベント・インスペクタ・サービスを使用して、EPNステージから離れるイベントを表示(トレース)し、EPNステージにイベントをインジェクトできます。
イベント・インスペクタ・サービスは、共通HTTPパブリッシュ/サブスクライブ・チャネルとサーバーを使用してイベントをトレースおよびインジェクトします。
注:
本番Oracle Stream Analyticsサーバーでは、イベント・インスペクタ・サービスを使用しないでください。このサービスは開発中にのみ使用します。
トレース・イベントのbinding
属性はoutbound
に設定する必要があり、インジェクトされるイベントのbinding
属性はinbound
に設定する必要があります。イベント・インスペクタ・クライアントを使用して次のようなイベントがインジェクトできます。
StockTick
イベントなど、タイプによる単一のシンプルなイベント。使用可能な特定のイベント・プロパティ・タイプはクライアントに依存します。
JSON形式の文字列として単一イベントのHTTPパブリッシュ・サブスクライブ・チャネルへの直接インジェクトJSONが表現できる任意のイベント・プロパティを使用できます。
1つ以上のJSON形式の文字列を含むファイルを使用した複数のイベントJSONが表現できる任意のイベント・プロパティを使用できます。イベント・インスペクタのサービス・クライアントがファイルを解析し、すべてのJSON文字列をHTTPパブリッシュ/サブスクライブ・チャネルにインジェクトします。
入力ファイルを作成するとき、JavaオブジェクトをJSON形式に変換するためにGSON Javaライブラリを使用できます。詳細は、次を参照してください。
イベント・インスペクタ・サービスは、Oracle Stream Analytics Visualizerをサポートします。『Oracle Stream Analyticsビジュアライザの使用』のイベント・インスペクタ・サービス管理に関する項を参照してください。
イベント・インスペクタ・サービスは、Oracle Stream Analyticsのすべてのイベント・タイプ、すなわちJavaBeanクラス、Map
およびタプルをサポートします。イベント・インスペクタ・サービスは、トレース・チャネルに公開する前にイベントをJavaScript Object Notation (JSON)形式に変換します。イベントはJSON形式でインジェクトする必要があります。
注:
イベント・インスペクタで使用されるイベント・タイプのプロパティ・タイプとして、バイト配列はサポートされていません。
JSON形式のイベントは構造に準拠している必要があります。表14-1は必要な属性を示します。
{ "event-type": "myEventType", "operation": "insert", "binding": "outbound", "value":{ "firstname": "Jane", "lastname": "Doe", "phone": { "code": 12345, "number": "office" }, } }
表14-2 イベント・インスペクタJSONイベントの必要な属性
|
イベント・インスペクタ・サービスは、サーバーconfig.xml
ファイルで定義されている次の名前で、動的HTTPパブリッシュ/サブスクライブ(HTTP pub-sub)チャネルを使用します。
/
SERVERNAME
/
APPLICATIONNAME
/
STAGENAME
/DIRECTION
SERVERNAME
: EPNステージが実行中のOracle Stream Analyticsサーバーの名前。
APPLICATIONNAME
: Oracle Stream Analyticsアプリケーションの名前。
STAGENAME
: EPNステージの名前。
DIRECTION
: 次のいずれか:
input
: イベント・インジェクション
output
: イベント・トレース
次に例を示します。
/server-1/myapp/MyInputAdapter/input
イベント・インスペクタのサービスは、HTTPパブリッシュ/サブスクライブ・サーバーを使用します。次のいずれかになります。
ローカル: event-inspector
要素でサーバー・ファイルを構成し、このマシンで実行中のローカル・パブリッシュ/サブスクライブ・サーバーの名前でpubsub-server-name
子要素を構成します。
リモート: event-inspector
要素でサーバー・ファイルを構成し、リモート・マシンで実行中のHTTPパブリッシュ/サブスクライブ・サーバーへのURLでpubsub-server-url
子要素を構成します。
デフォルト: サーバー・ファイルで1つのみのHTTPパブリッシュ/サブスクライブ・サーバーが定義されていてローカルまたはリモートのHTTPパブリッシュ/サブスクライブ・サーバーが指定されていない場合、イベント・インスペクタ・サービスはデフォルトでローカルのHTTPパブリッシュ/サブスクライブ・サーバーを使用します。
イベント・インスペクタ・サービスは同じHTTPパブリッシュ/サブスクライブ・チャネルとサーバーを使用してイベントをトレースしインジェクトします。
ローカルまたはリモートのHTTPパブリッシュ/サブスクライブ・サーバーでイベント・インスペクタ・サービスを構成できます。コンポーネント構成ファイルでイベント・インスペクタHTTPパブリッシュ/サブスクライブ・サーバーを構成します。サーバーにHTTPパブリッシュ/サブスクライブ・サーバーが1つのみ定義されていて、ローカルまたはリモートのHTTPパブリッシュ/サブスクライブ・サーバーが指定されていない場合、イベント・インスペクタ・サービスはデフォルトでローカルのHTTPパブリッシュ/サブスクライブ・サーバーを使用します。
ローカルHTTPパブリッシュ/サブスクライブ・サーバーのための構成
テストするコンポーネントを含むコンポーネント構成ファイルに対して、次のようにevent-inspector
のname
要素を追加します。
<event-inspector> <name>myEventInspectorConfig</name> <pubsub-server-name>myPubSub</pubsub-server-name> </event-inspector>
次の例に示すように、pubsub-server-name
の値myPubSub
は、ローカルのOracle Stream Analyticsサーバー・ファイルで定義されたhttp-pubsub
要素のname子要素の値です。
<http-pubsub> <name>myPubSub</name> <path>/pubsub</path> <pub-sub-bean> <server-config> <supported-transport> <types> <element>long-polling</element> </types> </supported-transport> <publish-without-connect-allowed>true</publish-without-connect-allowed> </server-config> <channels> ... </channels> </pub-sub-bean> </http-pubsub>
リモートHTTPパブリッシュ/サブスクライブ・サーバーのための構成
テストするコンポーネントを含むコンポーネント構成ファイルに対して、次のようにevent-inspector
のname
要素を追加します。
<event-inspector> <name>myEventInspectorTraceConfig</name> <pubsub-server-url>http://HOST:PORT/PATH</pubsub-server-url> </event-inspector>
HOST
: リモートのOracle Stream Analyticsサーバーのホスト名またはIPアドレス。
PORT
: リモートOracle Stream Analyticsサーバー・ファイルで定義されている、リモートOracle Stream Analyticsサーバーのnetio
ポート。デフォルトは9002です。
PATH
: リモートOracle Stream Analyticsサーバー・ファイルで定義されている、http-pubsub
要素のpath
子要素の値。
次の例に示すhttp-pubsub
構成の場合、有効なpubsub-server-url
は、次のようになります。
http://remotehost:9002/pubsub
次の例に示すように、pubsub-server-name
の値myPubSub
は、ローカルのOracle Stream Analyticsサーバー・ファイルで定義されたhttp-pubsub
要素のname子要素の値です。
<http-pubsub> <name>myPubSub</name> <path>/pubsub</path> <pub-sub-bean> <server-config> <supported-transport> <types> <element>long-polling</element> </types> </supported-transport> <publish-without-connect-allowed>true</publish-without-connect-allowed> </server-config> <channels> ... </channels> </pub-sub-bean> </http-pubsub>
イベント・インスペクタ・サービスHTTPパブリッシュ/サブスクライブ・サーバーを構成した後、イベントをインジェクトするためにイベント・インスペクタ・クライアントを使用できます。イベント・インジェクションを構成するには、Oracle Stream Analytics Visualizerを使用するか、アプリケーションでコンポーネント構成ファイルを編集して、アプリケーションのデプロイ時または再デプロイ時に設定されるインジェクション設定を指定します。
Oracle Stream Analytics Visualizerで、アプリケーションの再デプロイ時に破棄できる設定でイベント・インジェクションを構成します。『Oracle Stream Analyticsビジュアライザの使用』のイベントのインジェクトに関する項を参照してください。
アプリケーションのデプロイ時または再デプロイ時に設定されるイベント・インジェクション構成設定の場合、インジェクトするステージのコンポーネント構成設定を編集して、インジェクションを構成します。
たとえば、例に示したコンポーネント構成の一部は、イベント・インジェクション用にプロセッサを構成する方法を示しています。inject-parameters
要素のactive
子要素はインジェクションをオンにすることを指定し、channel-name
要素はインジェクトされる要素の送信元のHTTPパブリッシュ・サブスクライブ(HTTP pub-sub)チャネルを指定します。
<processor> <name>FindCrossRates</name> <inject-parameters> <active>true</active> <channel-name>/NonClusteredServer/fx/FindCrossRates/output</channel-name> </inject-parameters> <rules> <!-- Query rules omitted. --> </rules> </processor>
要素のリファレンス情報は、Oracle Stream Analyticsスキーマ・リファレンスを参照してください。
イベント・インスペクタ・サービスHTTPパブリッシュ・サブスクライブ(HTTP pub-sub)サーバーを構成した後、イベント・インスペクタ・クライアントを使用して、EPNのいずれかのステージから退出するイベントをトレースできます。イベントをトレースするには、Oracle Stream Analytics Visualizerを使用してトレースを構成するか、アプリケーションでコンポーネント構成ファイルを編集して、アプリケーションのデプロイ時または再デプロイ時に設定されるトレース設定を指定します。
Oracle Stream Analytics Visualizerで、アプリケーションの再デプロイ時に破棄できる設定でイベント・トレースを構成します。『Oracle Stream Analyticsビジュアライザの使用』のイベントのトレースに関する項を参照してください。
アプリケーションのデプロイ時または再デプロイ時に設定されるイベント・トレース構成設定について、トレース元のステージのコンポーネント構成設定を編集して、トレースを構成します。
たとえば、例に示したコンポーネント構成の一部は、イベント・トレース用にプロセッサを構成する方法を示しています。trace-parameters
要素のactive子要素はトレースをオンにすることを指定し、channel-name要素はトレースされる要素の送信先のHTTPパブリッシュ・サブスクライブ(HTTP pub-sub)チャネルを指定します。
<processor> <name>FindCrossRates</name> <trace-parameters> <active>true</active> <channel-name>/NonClusteredServer/fx/FindCrossRates/output</channel-name> </trace-parameters> <rules> <!-- Query rules omitted. --> </rules> </processor>
要素のリファレンス情報は、Oracle Stream Analyticsスキーマ・リファレンスのプロセッサに関する項を参照してください。
イベント・インスペクタAPIを使用して、イベントをインジェクトおよびトレースします。Oracle Stream Analytics APIに用意されているEventInspectorMBean
インタフェースを実装すると、イベントのトレースおよびインジェクション動作を制御できます。startInject
、stopInject
、startTrace
およびstopTrace
メソッドを呼び出すことができるのは、管理者およびモニター管理者のロールのみです。EventInspectorMBean
インタフェースとそのメソッドの詳細は、Oracle Stream Analytics Java APIリファレンスを参照してください。
EventInspectorMBean
インタフェースを実装すると、そのメソッドをアプリケーションから呼び出して、次のようにイベントのインジェクトおよびトレースが可能です。
イベントのインジェクト
JMXを介して、サーバーからcom.bea.wlevs.eventinspector.management.EventInspectorMBean
のインスタンスを取得します。
EventInspectorMBean.startInject
メソッドを呼び出し、HTTPパブリッシュ/サブスクライブ・チャネルのサーバーconfig.xml
ファイル構成で指定されているステージでイベント・インジェクションを開始します。HTTPパブリッシュ/サブスクライブ・チャネルおよびサーバーを参照してください。
指定したHTTPパブリッシュ/サブスクライブ・チャネルにイベントをパブリッシュします。
EventInspectorMBean.isInjecting
メソッドを使用し、ステージでイベント・インジェクションが開始されたことを確認します。
イベント・インジェクションを停止するには、EventInspectorMBean.stopInject
メソッドを呼び出します。
イベントのトレース
JMXを介して、サーバーからcom.bea.wlevs.eventinspector.management.EventInspectorMBean
のインスタンスを取得します。
EventInspectorMBean.startTraceメソッドを呼び出し、HTTPパブリッシュ/サブスクライブ・チャネルのサーバーconfig.xmlファイル構成で指定されているステージでイベント・トレースを開始します。HTTPパブリッシュ/サブスクライブ・チャネルおよびサーバーを参照してください。
EventInspectorMBean
.isTracingメソッドを使用し、HTTPパブリッシュ/サブスクライブ・チャネルでイベント・トレースが開始されたことを確認します。
イベント・インジェクションを停止するには、EventInspectorMBean.stopTrace
メソッドを呼び出します。
EPNシェルには、テスト用のシェル・コマンドが用意されています。
Oracle Stream Analyticsアプリケーション。EPNの操作をプログラムで実行する方法の詳細は、EPNコマンド・インタフェースを参照してください。
EPNシェルはApache Felix Gogoシェルを拡張し、OSGiフレームワークに標準のシェル・コマンド・プロンプトを提供します。Apache Felix Gogoのドキュメント(http://felix.apache.org/documentation/subprojects/apache-felix-gogo.html
)を参照してください。
EPNシェルでサーバーを起動するには、/Oracle/Middleware/my_oep/ user_projects/domains/<my_domain>/<server>
に移動し、次のようにEPNシェルでOracle Stream Analyticsサーバーを起動します。
UNIX:
./startwlevs.sh -shell ... g!
Windows:
startwlevs.cmd -shell ... g!
EPNシェルが起動すると、g!
というプロンプトが表示されます。すべてのコマンドのリストを表示するには、help
と入力します。EPNに固有のコマンドは、epn:channel
のようにepn:
という接頭辞が付きます。
g! help
特定のコマンドのヘルプを表示するには、help <command>
と入力します。たとえば、begin
コマンドのヘルプを表示するには、次のように入力してください。
g! help begin begin - Begins new session for invoking EPN commands scope: epn parameters: CommandSession
注:
Apache Felix Gogoコマンドには、CommandSession
引数を取るものがあります。これは、Apache Felix Gogoコマンドでは実行されない内部シェル引数です。EPNシェル・コマンドのヘルプ一覧を実行すると、ヘルプによってCommandSession
パラメータが一覧表示されますが、これは機能しません。
デフォルトでは、EPNセッションには暗黙的なOracle CQLプロセッサがあり、それがすべての出力をシェル・コンソールに表示するイベント・シンクに接続されます。Oracle CQL問合せをテストするには、次のように入力チャネルを作成し、問合せを定義してイベントを送信します。
g! begin g! channel MyChannel [msg=String] MyChannel g! query "select * from MyChannel" q0 g! send MyChannel [msg='Hi'] 11:14:26 618 -> insert event: {msg=Hi} q! end
EPNセッションを開始するにはbegin
コマンドを使用し、EPNセッションを終了するにはend
コマンドを使用します。セッションを終了すると、そのセッション中に作成されたEPNコンポーネントはすべて破棄されます。セッションはマルチスレッドに対応していません。
channel
コマンドは、String
型のmsg
プロパティを持つevent-type
でMyChannel
を作成します。event-type
にJavaクラス名も指定できます。
query
コマンドは、このセッションの暗黙的なOracle CQLプロセッサで、"select * query from MyChannel"
を登録します。プロセッサから問合せを削除するには、remove
コマンドを使用します。
send
コマンドは、insert
イベントをMyChannel
にディスパッチします。セッションに1つのチャネルしかない場合、チャネルの名前はオプションです。たとえば、"send MyChannel [msg='Hi']"
と"send [msg='Hi']"
の2つのコマンドは同等です。
構文"[msg='Hi']"
は、"msg"
をキー、"Hi"
を値としてキー/値のペアを1つ含むマップを作成します。これは、channel MyChannel [msg=String]
の最初の行で作成されたevent-type
と一致します。
update
コマンドとdelete
コマンドを使用すると、リレーションベースのチャネルに更新イベントおよび削除イベントを送信できます。insert
コマンドは、send
コマンドと同等です。また、statement
コマンドを使用するとセッションで現在登録されている文をチェックでき、channels
コマンドを使用するとセッションで作成したチャネルを確認できます。eventtypes
コマンドでは、サーバーにあるすべてのevent-types
の構造を表示できます。
Oracle CQLプロセッサの問合せの詳細は、Oracle CQLプロセッサを参照してください。
管理コマンドを使用すると、デプロイされているOracle Stream Analyticsアプリケーションのリスト、ライブラリのリスト、アプリケーションのインストール、既存のアプリケーションに対するイベントの送信、チャネル・イベントへのサブスクライブ、すべてのパブリック・ステージのリスト、OSGiサービスの取出し、標準Javaメソッドの呼出し、JMX操作の実行、サーバーの停止を行うことができます。
実行中のサーバーでデプロイされているOracle Stream Analyticsアプリケーションをすべてリストします。
g! listapps com.bea.wlevs.dataservices
Oracle Stream Analyticsライブラリをすべてリストします。次のコマンドは、EPNシェルを実行するためにモジュールのディレクトリで展開したバンドルをすべて一覧表示しています。
g! llistlibs org.apache.felix.bundlerepository org.apache.felix.gogo.runtime com.oracle.cep.shell org.apache.felix.gogo.command org.apache.felix.gogo.shell
新しいアプリケーションをインストールします。次の例では、アプリケーションをデプロイしてすぐに起動しています。
g! deployapp file:///Users./oepapps/hwlloworld.jar
アプリケーションをアンデプロイします。
g! deployapp file:///Users./oepapps/hwlloworld.jar g! undeployapp file:///Users./oepapps/hwlloworld.jar
send
、insert
、update
およびdelete
の各コマンドを使用するチャネルの完全名を指定して、既存のOracle Stream Analyticsアプリケーションにイベントを送信します。この例では、helloworld
アプリケーションでhelloworldInputChannel
にイベントを送信します。これが機能するためには、EPNシェルで検索できるようにチャネルを公開する必要があります。チャネルを公開
するには、まずアセンブリ・ファイルでadvertise
属性をtrue
に設定します。
アセンブリ・ファイルの設定は、次のとおりです。
<wlevs:channel id="helloworldInputChannel" event-type="HelloWorldEvent" advertise="true" >
EPNシェル・コマンドは次のとおりです。
g! event1=createevent HelloWorldEvent g! $event1 message "Hi Shell" g! send helloworld:helloworldInputChannel $event1
チャネルをサブスクライブします。サブスクライブするイベントは、ファイル名を指定するとシェル・コンソールまたはファイルに送信されます。アセンブリ・ファイルでadvertise
属性をtrue
に設定して、チャネルが公開されていることを確認します。次のコマンドは、helloworldアプリケーションからのすべての出力にサブスクライブし、その出力をシェル・コンソールに送信します。
g! subscribe helloworld:helloworldOutputChannel
アプリケーションのパブリックの(公開されている)ステージをすべてリストします。
g!introspect helloworld Application 'helloworld' provides the following OCEP services: Event Channel 'helloworldOutputChannel' for Event Type 'HelloWorldEvent'
helloworld
Oracle CQLプロセッサのCQLProcessorMBean
を取り出し、getAllQueries
操作を呼び出します。
g! proc = mbean helloworld:helloworldProcessor CQLProcessor g! $proc allQueries helloworldRule select * from helloworldInputChannel
取り出したmbeanをmbean
コマンドで使用すると、JMX操作のテストと操作が可能です。
サーバーを停止します。
g! stop
シェルをヘッドレスで実行してスクリプトを使用すると、アプリケーションのテストを自動化できます。たとえば、startwlevs
コマンドでgosh.args
システム・プロパティを指定すると、send-event.oep
という名前の次のスクリプトを実行できます。
begin channel -a [a=Long] query "select * from ch0" send 0 [a=1] send 1 [a=2] end
注:
スクリプトでは、send
コマンドの文字列タプル値を二重引用符で囲みます。たとえば、send [msg="Enter a value."]
のようにします。一重引用符はスクリプトで機能しません。
サーバーを起動するときにsend-event.oep
スクリプトを実行するには、startwlevs
コマンドの最後の行を次のように編集します。
"$JAVA_HOME/bin/java" -Dgosh.args=send-event.oep $JVM_ARGS $DEBUG_ARGS -Dwlevs.home="$USER_INSTALL_DIR" -Dbea.home="$BEA_HOME" -jar "${USER_INSTALL_DIR}/bin/wlevs.jar" $ARGS
次のように、他のスクリプトを呼び出すテスト・ドライバ・スクリプトを使用し、その出力を指示するのも1つの方法です。
source send-event.oep | tac test-output/log/send-event.out source test-delete.oep | tac test-output/log/test-delete.out
EPNセッションに対してタイムスタンプ形式を設定する、またはタイムスタンプをオフにすることができます。
OUT_TIME: FORMATTED | PLAIN | OFF
出力イベントでタイムスタンプをオフにするには、次のように入力します。
g! OUT_TIME=OFF
書式設定したタイムスタンプを指定するには、次のように入力します。
h! OUT_TIME=FORMATTED
次のリストでは、EPNコマンドを示して説明します。
epn:begin
: EPNコマンドを呼び出す新しいセッションを開始します。
epn:channel
: パラメータに基づいて、次のように異なるアクションを実行します。
マップベースのイベント・タイプでチャネルを作成します。
更新キーとJavaクラス・イベント・タイプを持つ名前付きのリレーション・チャネルを作成します。
マップベースのイベント・タイプで名前付きのチャネルを作成します。
Javaクラスのイベント・タイプで名前付きのチャネルを作成します。
Javaクラスのイベント・タイプでチャネルを作成します。
更新キーとマップベースのイベント・タイプを持つ名前付きのリレーション・チャネルを作成します。
epn:channels
: EPNスコープ内のすべてのチャネルを一覧表示します。
epn:createevent
: 指定したイベント・タイプでイベントを作成します。
epn:end
: EPNセッションを終了します。
epn:eventtypes
: すべてのイベント・タイプを一覧表示します。
epn:heartbeat
: アプリケーションによってタイムスタンプが設定されたチャネルにハートビートを送信します。
epn:query
: Oracle CQL問合せを登録します。
epn:remove
: EPNスコープ内に登録されているすべての文を削除します。
epn:send
: 名前付きのチャネルに挿入イベントを送信します。sendinsert
コマンドと同じです。
epn:senddelete
: 名前付きのチャネルで名前付きのイベントを削除します。
epn:sendinsert
: 名前付きのチャネルに挿入イベントを送信します。send
コマンドと同じです。
epn:sendupdate
: 指定したURLにあるバンドルで、OSGiバンドルを更新します。
epn:statements
: EPNスコープ内に登録されているすべてのOracle CQL文を一覧表示します。
epn:subscribe
: イベント・チャネルにサブスクライブし、イベントをファイルに出力します。
epn:unsubscribe
: イベント・チャネルからサブスクライブ解除します。
epn:view
: Oracle CQLビューを登録します。
次のリストでは、管理コマンドを示して説明します。
mngt:deployapp: オプションのパラメータを指定してアプリケーションをデプロイします。
mngt:deployrepapp: リポジトリからアプリケーションをデプロイします。
mngt:introspect: アプリケーションでそのパブリック・インタフェースをイントロスペクトします。
mngt:listapps: デプロイされているすべてのアプリケーションを一覧表示します。
mngt:listlibs: デプロイされているすべてのライブラリを一覧表示します。
mngt:mbean: ローカルMBeanサーバーからMBeanステージを取得します。
mngt:shutdown: サーバーをシャットダウンします。
mngt:undeployapp: 指定したアプリケーションをアンデプロイします。
EpnCommand
インタフェースは、Oracle Stream Analytics EPN CQLプログラミング・モデルの上層に位置するJava APIです。JavaアプリケーションでEpnCommand
インタフェースを使用すると、EPNコンポーネントを操作できます。
EPNセッションをプログラムで開始し、EPNシェルで説明したコマンド行EPNシェルを使用する場合と似たEPNコマンドを実行できます。
EPNセッションを開始するにはbeginSession()
を使用し、EPNセッションを終了するにはendSession()
を使用します。セッションを終了すると、そのセッション中に作成されたEPNコンポーネントはすべて破棄されます。セッションはマルチスレッドに対応していないため、アプリケーション・コードで同期する必要があります。
Oracle Stream AnalyticsアプリケーションでEPNコマンドを使用するには、まずEpnCommand
OSGiサービスを取得します。これによって、サービスがOSGiサービス・ファクトリとして登録されます。OSGiサービスを取得すると、JUnit
テスト・ケースに似たテスト・ケースを作成し、EPNコンポーネントとOracle CQL文のテストに使用できます。JUnit
と同様のテスト・ケースに使用できるのは、EpnCommand
クラスに公開されたコマンドのみです。
EpnCommand
インタフェースを使用し、プログラミングによりOracle Stream Analyticsアプリケーションを作成することもできます。
void beginSession()
: コマンドを呼び出す新しいセッションを開始します。1回にアクティブにできるのは1つのセッションのみです。次のリストに、EpnCommand
のメソッドとパラメータを示します。詳細は、EpnCommand
のJavadocを参照してください。
endSession()
: 現在のセッションを終了します。1回にアクティブにできるのは1つのセッションのみです。
void getEventChannels()
: 現在のセッションで作成されたすべてのチャネルを返します。
EventChannel[] getEventChannel(String channelName)
: 名前付きのイベント・チャネルを返します。
EventChannel getEventChannel(String channelName)
: 名前付きのイベント・チャネルを取得します。
パラメータに基づいて異なるタイプのチャネルを作成します。
EventChannel createChannel(String channelName, Class<?> clazz)
: Javaクラスのイベント・タイプで、システム・タイムスタンプの付加された新しいチャネルを作成します。
EventChannel createChannel(boolean relation, String applicationTimestampProp, boolean totalOrder, List<String> keys, String channelName, Class<?> class)
: Javaクラスのイベント・タイプで新しいチャネルを作成します。
EventChannel createChannel(boolean relation, String applicationTimestampProp, boolean totalOrder, List<String> keys, String channelName, Class<?> class, EventBuilder.Factory factory):
Javaクラスのイベント・タイプで新しいチャネルを作成します。
EventChannel createChannel(String channelName, Map<String, String> metadata)
: メタデータベースのイベント・タイプで、システム・タイムスタンプの付いた新しいチャネルを作成します。
EventChannel createChannel(boolean relation, String applicationTimestampProp, boolean totalOrder, List<String> keys, String channelName, Map<String, String> metadata)
: メタデータベースのイベント・タイプで新しいチャネルを作成します。
EventChannel createChannel(boolean relation, String applicationTimestampProp, boolean totalOrder, List<String> keys, String channelName, Map<String, String> metadata, EventBuilder.Factory factory)
: メタデータベースのイベント・タイプで新しいチャネルを作成します。
Processor[] getProcessors()
: 現在のセッションで作成されたすべてのOracle CQLプロセッサを返します。
Processor createProcessor()
: 現在のセッションでOracle CQLプロセッサを作成します。
StreamSource link(StreamSource fromStage, StreamSink toStage)
: ステージとステージを接続します。
void unlink(StreamSource source, StreamSink sink)
: シンクからソースを切断します。
Statement createView(Processor processor, String id, String StatementValue)
: Oracle CQLビューを作成します。
Statement createQuery(Processor processor, String id, String statementValue)
: Oracle CQL問合せを作成します。
Object createEvent(String eventTypeName)
: イベント・タイプ名に基づいてイベントを作成します。イベントが作成されるのは、イベント・タイプ・リポジトリで使用できる場合に限られます。
MapEventObject createMapEvent(EventChannel channel, Map<String, Object> event)
: メタデータベースのチャネルに対してマップ・イベントを作成します。
void sendInsert(String channelName, Object event)
: チャネルに挿入イベントを送信します。
void sendUpdate(String channelName, Object event)
: チャネルに更新イベントを送信します。
void sendDelete(String channelName, Object event)
: チャネルに削除イベントを送信します。
void sendHeartbeat(Long timestamp, String channelName)
: チャネルにハートビートを送信します。
次の例は、基本的なJUnit
テスト・ケースです。
//Get the OSGI service ServiceReference ref = ctx.getServiceReference(EpnCommand.class.getName()); commandSession = (EpnCommand) ctx.getService(ref); //Begin a session commandSession.begin(); //Create an Oracle CQL processor Processor p1 = commandSession.createProcessor("p1"); //Create two stream channels from the MyEvent Java class. EventChannel ch1 = commandSession.createChannel(false, "c1", MyEvent.class); EventChannel ch2 = commandSession.createChannel(false, "c2", MyEvent.class); //Create a listener to retrieve events that equal event e1 //The MyEventListener class implements the StreamSink or RelationSink interface MyEventListener listener = new MyEventListener(); //Connect channel ch1 and processor p1 commandSession.link((StreamSource) ch1, (StreamSink) p1); //Connect channel p1 and processor ch2 commandSession.link((StreamSource) p1, (StreamSink) ch2); //Connect channel ch2 and processor listener commandSession.link((StreamSource) ch2, (StreamSink) listener); //Create the query in processor p1 commandSession.createQuery(p1, "q1", "select * from c1"); //Create event e1 and assign two properties, a and 1. Object e1 = new MyEvent("a", 1); //Send insert event e1 to channel ch1 ch1.sendInsertEvent(e1); //Test whether event e1 equals the event retrieved by the listener assertEquals(e1, listener.getEvent()); //End the session commandSession.end();