アダプタ

4 アダプタ

アダプタは、EPNの入出力データを管理します。Oracle Stream Analyticsには、CSV、RMIおよびHTTPなどの各種タイプのデータを処理するために様々な種類のインバウンド・アダプタとアウトバウンド・アダプタが用意されています。すべてのアダプタには、OSGi登録アダプタ・ファクトリ・サービスに対する参照であり、アダプタが処理するデータのタイプを定義するproviderプロパティがあります。

インバウンド・アダプタは、EPNに入るデータ・ストリームからイベント・データを受信し、データをイベント・タイプに応じてイベントに割り当て、データをEPN内の次のステージに送信します。アウトバウンド・アダプタは、EPNによって処理されたイベントを受信し、イベントを出力フォームに変換し、変換したデータを別のEPN、非EPNアプリケーション、CSVファイルまたはWebページなどの出力データ・ソースに送信します。

この章の内容は次のとおりです。

高可用性アダプタの詳細は、高可用性アプリケーションを参照してください。

csvgenアダプタとともにロード・ジェネレータを使用してデータ・フィードをシミュレートしてアプリケーションをテストする方法の詳細は、テスト1-2-3を参照してください。

4.1 アダプタの作成

ほとんどのアダプタを作成する上で最適な方法は、Oracle JDeveloperを使用する方法です。Oracle JDeveloperコンポーネント・ウィンドウには、インバウンド・アダプタおよびアウトバウンド・アダプタとしてCSV、EDN、RMI、HTTPおよびJMSが用意されています。

他のアダプタの場合、構成ファイルを直接編集します。この章の各アダプタの項では、設定を確認できるようにアセンブリと構成ファイルの構成の例を示します。

この章では、様々なタイプのアダプタに関するアセンブリおよび構成ファイルの設定について説明します。アダプタ・スキーマは、Oracle Stream AnalyticsのインストールのOracle/Middleware/my_oep/oep/wlevs_application_config.xsdディレクトリも参照してください。

4.2 クラスタ配分サービス

クラスタ配分機能には、クラスタ内のすべてのサーバーに着信イベントを配分するために様々なOracle Stream Analyticsアダプタ・タイプ用のメカニズムが採用されています。

イベントを配分するために個別アダプタまたはイベントBeanインスタンスを構成でき、この場合、このアダプタによって処理されるすべての入力イベントはクラスタ内のすべてのサーバーに送信(配分)されます。

配分アダプタにより、すべての入力イベントがクラスタ内のすべてのサーバーに確実に送信(配分)されます。入力アダプタを配分アダプタに変換するには、distributeInput要素を追加し、次のように、これをtrueに設定します。distributionThreadsCountプロパティはオプションで、1にデフォルト設定されます。

<wlevs:adapter id="myLoadgenAdapter" provider="loadgen">
  <wlevs:instance-property name="distributeToClusterGroup" value="true"/>
  <wlevs:instance-property name="distributionThreadsCount" value="1"/>
</wlevs:adapter>

Oracle Stream Analyticsは、loadgen、CSVインバウンド・アダプタおよびJMSインバウンド(キュー)アダプタ用のクラスタ配分サービスをサポートしています。Oracle Stream Analyticsは、CSVアウトバウンド・アダプタ、JMSインバウンド(トピック)アダプタ、JMSアウトバウンド・アダプタおよびHTTPパブリッシュ/サブスクライブ・アダプタ用のクラスタ配分サービスはサポートしていません。入力配分に対してトピック宛先を構成するとエラーになります。トピック構成によって警告ログ・メッセージが生成され、トピック構成は無視されます。

前述のアダプタ・タイプ以外にも、次の例に示すように、BeanのプロバイダをclusterGroupDistributorとして指定することにより、受信したすべてのイベントをすべてのクラスタ・メンバーに配分するためのイベントBeanを構成することもできます。

<wlevs:event-bean id="distributor-bean" provider="clusterGroupDistributor">
   <wlevs:instance-property name="distributionThreadsCount" value="1"/>
   ... other event bean properties ...
</wlevs:event-bean>

現在、Oracle JDeveloperには、クラスタ配分サービス用のコンポーネントは用意されていません。ただし、Oracle Stream Analyticsアプリケーション用のアセンブリ・ファイルおよび構成ファイルにエントリを追加することにより、クラスタ配分サービスを作成できます。

4.3 パスワードの暗号化

一部のアダプタには、ユーザー名およびパスワードの子要素があります。Oracle Stream Analyticsには、パスワードが含まれるファイルを暗号化できるようencryptMSAConfigコマンドが用意されています。

4.4 JAXBサポート

Oracle Stream Analyticsには、アダプタおよびイベントBeanでJava Architecture for XML Binding (JAXB)マッピング機能を使用してXMLオブジェクトとJavaオブジェクト間でイベント・データをマーシャリングおよびアンマーシャリングするための簡易インタフェースが用意されています。

JAXBインタフェースは、JAXB 2.2仕様およびEclipseLink Moxyプロバイダ拡張機能をサポートしています。

マッピング操作は次の方法で構成できます。

一連の注釈付きJavaクラスを出力するために、XMLスキーマからJavaオブジェクトにマップします。
JAXB注釈を使用して、1つのJavaオブジェクト・セットから別のJavaオブジェクト・セットまたはXMLにマップします。
既存のXMLスキーマから既存の事前定義済のJavaオブジェクト表現にマップします。このアプローチでは、EclipseLink Moxy拡張機能を使用し、マッピング詳細が含まれる外部のmetadataファイルを必要とします。metadataファイルは、アプリケーション構成によって参照されます。

4.4.1 EclipseLink Moxy

EclipseLink Moxyには、注釈を指定せずにXMLスキーマまたはJavaクラスを変更しないで既存のXMLスキーマと事前定義済のJavaクラス間でマップできる拡張機能が用意されています。マッピング情報は、XPath構文を使用して外部メタデータ・ファイルで指定します。

柔軟性の高いEclipseLink Moxy拡張機能を使用すると、複雑な操作を実行できます。たとえば、複雑なXMLデータのサブセットをより簡単なイベント表現にマップできます。また、深くネストしたXMLドキュメントをフラットなJava Beanイベント形式にフラット化し、Oracle CQLによる処理も可能にします。

EclipseLink Moxyの外部のmetadataはXMLで指定します。http://www.eclipse.org/eclipselink/xsds/eclipselink_oxm_2_2.xsdにあるスキーマにアクセスしてください。

4.4.2 API

JAXB機能を必要とするアダプタまたはイベントBeanは、com.oracle.cep.mappers.api.Mapperインタフェースを実装するBeanのインジェクションによってこの機能を取得します。Mapperインタフェースは、次のようになります。

public interface Mapper {
  Marshaller createMarshaller() throws MapperException;
  Unmarshaller createUnmarshaller() throws MapperException;
}

アダプタまたは他のEPNコンポーネント・コードでは、インジェクトされたBeanを使用してマーシャリング・オブジェクトおよびアンマーシャリング・オブジェクトを作成します。次に示すcom.oracle.cep.mappers.api.Marshallerおよびcom.oracle.cep.mappers.api.Unmarshaller インタフェースは、ほとんどのアプリケーションに対して機能します。

public interface Marshaller { 
  void marshal(Object object, javax.xml.transform.Result result) 
               throws MapperException;
 }

public interface Unmarshaller {
  Object unmarshal(javax.xml.transform.Source source) 
                   throws MapperException;
}

一部のアプリケーションでは、ターゲット・クラスを引数として使用するアンマーシャリング・メソッドなど、マーシャリングおよびアンマーシャリング用として特別なメソッド・シグネチャが必要な場合があります。このような場合は、かわりにcom.oracle.cep.mappers.jaxb.JAXBMarshallerImpl およびcom.oracle.cep.mappers.jaxb.JAXBUnmarshallerImplインタフェースを使用します。これらのインタフェースには、javax.xml.bind.Marshallerおよびjavax.xml.bind.Unmarshallerインタフェースによってサポートされる完全セットのマーシャリング・メソッドおよびアンマーシャリング・メソッドに対応するメソッドが用意されています。

アセンブリ・ファイル

次のアセンブリ・ファイル・エントリでは、プロパティを使用してマッパーBeanをコールし、イベント・タイプおよびメタデータ・ファイルを指定しています。

<bean id="mapperBean" class="com.oracle.cep.mappers.jaxb.JAXBMapperImpl" >
  <property name="eventTypeName" value="CallCenterActivity" />
  <property name="metadata" value="external_metadata_case1.xml" />
</bean>

ファクトリをコールしてマッパーBeanを作成するには、Bean要素に対して次を指定します。

<bean id="mapperBean" class="com.oracle.cep.mappers.jaxb.JAXBMapperFactory"
      factorymethod="create"/>

構成ファイル

次の構成ファイル・エントリは、マッパーBeanに対してプロパティを指定します。

<jaxb-mapper>
  <name>mapperBean</name>
  <event-type-name>CallCenterActivity</event-type-name>
  <metadata>external_metadata_case1.xml</metadata>
</jaxb-mapper>

プロパティ

マッパーBeanは、プロパティをサポートしています。metadataMapを除くすべてのプロパティは、アセンブリ・ファイルのプロパティまたは構成ファイルの要素として構成できます。

表4-1 マッパーBeanのプロパティおよび要素

アセンブリ・ファイルのプロパティ名	構成ファイルの要素名	説明
eventTypeName	event-type-name	アプリケーションによってイベント・タイプ・リポジトリに登録されているイベント・タイプの名前。イベント・タイプはJavaクラスに対応しています。このクラスのパッケージ名は、マッパーBeanによって表されるJAXBContextを初期化する際にコンテキスト・パスとして使用されます。この構成では、マッパーによって表される`JAXBContext`のコンテキスト・パスの作成で使用される`eventTypeName`または`contextPath`を指定する必要があります。これらのパッケージは、アプリケーションの`classpath`上に存在し、スキーマによって生成されたクラス、JAXBによって注釈が付けられたクラス、またはMoxy外部メタデータによって参照されたクラスを含む必要があります。これらのクラスは、マーシャリング操作およびアンマーシャリング操作用のJavaオブジェクト・グラフとして使用されます。
`contextPath`	`context-path`	Javaパッケージ名のコロン区切りリスト。マッパーBeanによって表される`JAXBContext`を初期化するために指定したコンテキスト・パス。この構成では、マッパーによって表される`JAXBContext`のコンテキスト・パスの作成で使用される`eventTypeName`または`contextPath`を指定する必要があります。これらのパッケージは、アプリケーションの`classpath`上に存在し、スキーマによって生成されたクラス、JAXBによって注釈が付けられたクラス、またはMoxy外部メタデータによって参照されたクラスを含む必要があります。これらのクラスは、マーシャリング操作およびアンマーシャリング操作用のJavaオブジェクト・グラフとして使用されます。
`validate`	`validate`	`false`にデフォルト設定される`Boolean`値。`true`の場合、`schema`を指定する必要があります。マーシャリングおよびアンマーシャリング中にスキーマ検証が行われます。
`schema`	`schema`	検証に使用されるXMLスキーマ・ファイルのファイル名。`META-INF/wlevs/mappers/jaxb`ディレクトリ内でアプリケーションを使用してスキーマ・ファイルをパッケージ化します。
`metadata`	`metadata`	マッピングのカスタマイズ用のEclipseLink Moxy外部メタデータが含まれるファイルの名前。`META-INF/wlevs/mappers/jaxb`ディレクトリ内でアプリケーションを使用してファイルをパッケージ化します。
`metadataMap`	N/A	`contextPath`の各コンポーネントに対応する1つのエントリが含まれるSpringの`<map>`要素。各エントリについて、キーは`contextPath`のパッケージ名であり、対応する値は、このパッケージに対するEclipseLink Moxyの外部`metadata`が含まれるファイルの名前です。

アプリケーションでEclipseLink Moxy固有の外部metadataが使用される場合、metadataの場所はmetadataプロパティまたはmetadataMapプロパティによって指定されます。contextPath上に複数のパッケージがある場合、metadataMapプロパティが必要です。コンポーネント構成ファイルでのmetadataMapプロパティの指定はサポートされていません。

4.5 CSVアダプタ

CSVアダプタは、カンマで区切られたインバウンド・データおよびアウトバウンド・データを処理します。EPNに入力するためにカンマ区切りの値形式でデータを受信するにはCSVInboundアダプタを使用し、EPNから出力するカンマ区切りの値形式でデータを送信するにはCSVOutboundアダプタを使用します。

Oracle Stream Analyticsインストールで提供されるロード・ジェネレータで、CSVインバウンド・アダプタを使用するOracle Stream Analyticsアプリケーションをテストできます。ロード・ジェネレータにより、サンプル・データが含まれるASCIIファイルが読み取られます。CSVインバウンド・アダプタを使用する必要があるのは、CSVインバウンド・アダプタはロード・ジェネレータによって生成されたデータ・パケットを解読するようコーディングされているためです。ロード・ジェネレータとcsvgenアダプタを参照してください。

注意:

java.sql.TimeStampタイプの場合、CSVアダプタは、yyyy-mm-dd'T'hh:mm:ss[.fffffffff]の形式でデータの読取りおよび書込みを行います。たとえば、"2012-12-12T12:12:12.120"です。

CSVアダプタを作成する上で最適な方法は、Oracle JDeveloperコンポーネント・ウィンドウを使用する方法です。Oracle JDeveloperによって生成された次のアセンブリおよび構成ファイルは、CSVのインバウンド・アダプタおよびアウトバウンド・アダプタの構成を示しています。

アセンブリ・ファイル

CSVインバウンド・アダプタは、StockData.csvファイルから読み取られたデータをTradeEventイベント・タイプのイベントに変換します。

wlevs:listener要素は、TradeEventアクティビティについてCSVインバウンド・アダプタをリスニングするコンポーネントを指定します。この例では、リスナーはAdapterOutputChannelです。AdapterOutputChannelコンポーネントは、StockTradeCSVInboundAdapterをリスニングしてここから取引イベントを受信し、次のステージに送信します。

<wlevs:adapter id="StockTradeCSVInboundAdapter" provider="csv-inbound">
    <wlevs:listener ref="AdapterOutputChannel"/>
    <wlevs:instance-property name="eventType" value="TradeEvent"/>
    <wlevs:instance-property name="sourceUrl"
    value="file:/scratch/mpawlan/oep9-19/oep/utils/load-generator/StockData.csv"/>
  </wlevs:adapter>

アウトバウンド・アダプタのアセンブリ・ファイル構成は、インバウンド・アダプタと類似していますが、append属性が含まれます。trueに設定すると、Oracle Stream Analyticsにより、既存の出力ファイルにデータが追加されます。falseに設定すると、Oracle Stream Analyticsにより、新規ファイルが作成されるか、または同じ名前の既存ファイルが上書きされます。

 <wlevs:adapter id="StockTradeCSVOutboundAdapter" provider="csv-outbound">
    <wlevs:instance-property name="eventType" value="TradeEvent"/>
    <wlevs:instance-property name="outputFile" value="/scratch/mpawlan/oep9-19/oep/utils/load-generator/StockData.csv"/>
    <wlevs:instance-property name="append" value="false"/>
  </wlevs:adapter>

outputFile値には絶対パスまたは相対パスを指定できます。相対パスの場合、../filename.csv、../result.csvまたはupload/result.csvを指定できます。相対パスを指定する場合、抽象パスに親ディレクトリが含まれることを確認してください。たとえば、UNIXの場合、カレント・ディレクトリ内のファイルを単純にresult.csvとして指定するのではなく./result.csvとして指定します。

構成ファイル

構成ファイル内のアダプタ要素には、アダプタのname属性とその値が示されます。アダプタ名は、アセンブリ・ファイル内のアダプタのid属性と一致する必要があります。

<csv-adapter>
  <name>StockTradeCSVInboundAdapter</name>
  <event-interval units="nanoseconds">5</event-interval>
</csv-adapter>

<csv-adapter>
  <name>StockTradeCSVOutboundAdapter</name>
</csv-adapter>

4.6 EDNアダプタ

イベント配信ネットワーク(EDN)のインバウンド・アダプタおよびアウトバウンド・アダプタは、JAXBを使用してEPNからOracle SOA Suiteイベント・ネットワークへのインタフェースを実現します。

EDNアダプタには、EDN XMLデータ転送をRAW XML (trueの場合)として表すかJAXBを使用してJavaオブジェクトとして表すかを指定するraw-xml-content構成要素があります。JAXBの場合、アダプタは、一連の適切なスキーマ(xjc)によって生成されたクラスがOracle Stream Analyticsアプリケーション・バンドルのクラス・パスに含まれることを想定します。

EDNアダプタは、イベント・タイプおよびEDLファイルへの参照を使用して構成します。初期化時には、構成されたイベント・タイプと一致するイベント定義QNameが含まれるEDLファイルが検索されます。構成されたイベント・タイプがEDL内で見つかった場合は、対応するQNameについてEDNにサブスクリプションが登録されます。

4.6.1 使用方法

EDNInboundアダプタを使用して、Oracle SOA Suiteイベント・ネットワークからの着信データを受信します。EDN入力アダプタは、指定されたEDNイベント・タイプをサブスクライブし、着信EDNイベントをOracle Stream Analyticsイベント・タイプに変換し、Oracle Stream Analyticsアプリケーションが処理できるようにします。

EDNOutboundアダプタを使用して、Oracle SOA Suiteイベント・ネットワークにアウトバウンド・データを送信します。EDN出力アダプタは、Oracle Stream Analyticsイベントを対応するEDNイベントに変換し、これらをEDNにパブリッシュします。パブリッシュされたイベントは、Oracle Stream Analyticsアプリケーションで生成された新規イベントである場合や、EDN入力アダプタによって受信され、Oracle Stream Analyticsアプリケーションによって処理され、EDN出力アダプタに送信されたEDNイベントである場合があります。

4.6.2 EDNアダプタの作成

EDNアダプタを作成する上で最適な方法は、Oracle JDeveloperコンポーネント・ウィンドウを使用する方法です。Oracle JDeveloperによって生成された次のアセンブリおよび構成ファイルは、EDNのインバウンド・アダプタおよびアウトバウンド・アダプタの構成を示しています。

アセンブリ・ファイル

EDN入力アダプタは、EDN入力チャネルをリスニングしてFraudCheckRequestタイプのすべてのイベントを検索します。
EDN出力アダプタは、FraudCheckRequestタイプのイベントをEPN内の次のステージに送信します。

<wlevs:adapter id="edn-inbound-adapter" provider="edn-inbound">
  <wlevs:listener ref="ednInputChannel"/>
  <wlevs:instance-property name="eventType" value="FraudCheckRequest"/>
</wlevs:adapter>

<wlevs:adapter id="edn-outbound-adapter" provider="edn-outbound">
  <wlevs:instance-property name="eventType" value="FraudCheckResponse"/>
</wlevs:adapter>

構成ファイル

注意:

バンドルされたJARファイルの固定パスにEDLファイルおよびスキーマ(xsd)ファイルを配置する必要があります。

<edn-adapter>
  <name>edn-outbound-adapter</name>
  <edl-file>FraudCheckEvent.edl</edl-file>
  <validate>false</validate>
  <raw-xml-content>false</raw-xml-content>
  <jndi-provider-url>t3://localhost:7101</jndi-provider-url>
  <jndi-factory>weblogic.jndi.WLInitialContextFactory</jndi-factory>
  <user>weblogic</user>
  <password>welcome1</password>
</edn-adapter>

<edn-adapter>
  <name>edn-inbound-adapter</name>
  <edl-file>FraudCheckEvent.edl</edl-file>
  <schema-file>FraudCheckType.xsd</schema-file>
  <validate>false</validate>
  <raw-xml-content>false</raw-xml-content>
  <jndi-provider-url>t3://localhost:7101</jndi-provider-url>
  <jndi-factory>weblogic.jndi.WLInitialContextFactory</jndi-factory>
  <user>weblogic</user>
  <password>welcome1</password>
</edn-adapter>

4.7 ファイル・アダプタ

ファイル・アダプタは、ファイルからデータを読み取ってEPNに書き込み、このデータをイベントに変換します。

現在、Oracle JDeveloperには、ファイル・アダプタ用のコンポーネントは用意されていません。ただし、Oracle Stream Analyticsアプリケーション用のアセンブリ・ファイルおよび構成ファイルにエントリを追加することにより、ファイル・アダプタを作成できます。

アセンブリ・ファイル

pathプロパティは、入力ファイルの場所を示します。アダプタは、入力ファイルからデータを読み取る際に、受信データをOrderArrivalEventタイプのイベントに変換します。ファイル・アダプタがファイルの読取りを開始するまでに5000ナノ秒のinitialDelayがあります。ダウンストリームのOrderArrivalチャネルは、OrderArrivalEventタイプのイベントをリスニングします。

<wlevs:adapter id="inputAdapter" provider="file" >
        <wlevs:instance-property name="path"
            value="@wlevs.domain.home@/inpOrderArrival.txt"/>
        <wlevs:instance-property name="eventType" value="OrderArrivalEvent"/>
        <wlevs:instance-property name="initialDelay" value="5000"/>
        <wlevs:listener ref="OrderArrival"/>
</wlevs:adapter>

構成ファイル

 <adapter>
    <name>inputAdapter</name>
 </adapter>

4.8 HTTPパブリッシュ/サブスクライブ・アダプタ

HTTPパブリッシャ・アダプタを使用して、JavaScript Object Notation (JSON)イベント・データをEPNからWebベースのユーザー・インタフェースに送信します。HTTPサブスクライバ・アダプタを使用して、EPNに入るJavaScript Object Notation (JSON)イベント・データを受け入れます。JSONイベント・データは、ユーザー・アクションによってイベントを生成するHTTPサーバーから送信されます。

Oracle Stream Analytics内のHTTPパブリッシュ/サブスクライブ・サーバーは、cometdプロジェクトで提案されているBayeuxプロトコルに基づいています。Bayeuxプロトコルには、クライアントとサーバーがHTTPを介して非同期メッセージで通信するための規約が定義されています。

リモートまたはローカルHTTPパブリッシャ・アダプタおよびリモートHTTPサブスクライバ・アダプタを作成できます。HTTPアダプタがローカルとリモートのどちらであるかは、必要な<server-url>子要素に指定するローカルまたはリモートURLによって決定されます。

HTTPパブリッシュ/サブスクライブ・アダプタを作成する上で最適な方法は、Oracle JDeveloperを使用する方法です。

注意:

HTTPパブリッシュ/サブスクライブ・アダプタで使用されるイベント・タイプのプロパティ・タイプとして、バイト配列はサポートされていません。

アセンブリ・ファイル

<wlevs:adapter id="http-pub-adapter" provider="httppub"/>

<wlevs:adapter id="http-sub-adapter" provider="httpsub" />

構成ファイル

すべてのローカル・アダプタおよびリモート・アダプタについて、サーバーのURLをserver-urlプロパティに指定します。サーバーには、Oracle Stream Analyticsサーバー、WebLogic Serverインスタンス、または任意のサードパーティのHTTPパブリッシュ/サブスクライブ・サーバーを指定できます。

パブリッシュ対象のすべてのローカル・アダプタについて、server-context-path要素を追加し、現在のOracle Stream AnalyticsアプリケーションをホストするOracle Stream Analyticsインスタンスに関連付けられているローカルHTTPパブリッシュ/サブスクライブ・サーバーへのパスを指定します。

デフォルトでは、各Oracle Stream Analyticsサーバーは、パス/pubsubを持つHTTPパブリッシュ/サブスクライブ・サーバーを使用して構成されます。新規ローカルHTTPパブリッシュ/サブスクライブ・サーバーを作成したか、デフォルトの構成を変更した場合は、サーバー・ファイル内のサーバーの場所を指定します。ファイル内で、サーバーの場所とともにhttp-pubsub要素のpath値を指定します。Oracle Stream Analyticsインストール内のサーバー・ファイルは、/Oracle/Middleware/my_oep/examples/domains/<my_domain>/defaultserverにあります。

channel子要素は、アダプタがパブリッシュするチャネルまたはサブスクライブするチャネルを指定します。

<http-pub-sub-adapter>
  <name>http-pub-adapter</name>
  <server-url>http://myhost.com:9102/pubsub</server-url>
  <channel>/channel2</channel>
  <event-type>com.mycompany.httppubsub.PubsubEvent</event-type>
  <user>wlevs</user>
  <password>wlevs</password>
</http-pub-sub-adapter>

<http-pub-sub-adapter>
  <name>http-sub-adapter</name>
  <server-url>http://myhost.com:9102/pubsub</server-url>
  <channel>/channel2</channel>
  <event-type>com.mycompany.httppubsub.PubsubEvent</event-type>
</http-pub-sub-adapter>

4.9 HTTPパブリッシュ/サブスクライブ・アダプタのカスタム・コンバータBean

HTTPパブリッシュ/サブスクライブ・アダプタは、受信したJavaScript Object Notation (JSON)メッセージをイベント・タイプに変換し、再度戻します。JSONのインバウンド・メッセージおよびアウトバウンド・メッセージをイベント・タイプに変換し、再度戻す方法をカスタマイズするには、カスタム・コンバータBeanを作成します。

4.9.1 Bayeuxプロトコル

HTTPパブリッシュ/サブスクライブ(pub-sub)サーバーはcometdプロジェクトで提案されているBayeuxプロトコルに基づいています。Bayeuxプロトコルには、クライアントとサーバーがHTTPを介して非同期メッセージで通信するための規約が定義されています。pub-subサーバーは、Bayeuxプロトコルを理解するすべてのクライアントと通信できます。

次のフレームワークを使用してWebクライアントを開発できます。

BayeuxプロトコルをサポートするDojo JavaScriptライブラリ。Oracle Stream Analyticsでは、このライブラリは提供されていません。詳細は、http://dojotoolkit.org/を参照してください。
Bayeuxプロトコルを使用してpub-subサーバーと通信するFlexクライアントの開発を可能にするWebLogic Workshop Flexプラグイン。

4.9.2 カスタム・コンバータBeanの作成

カスタム・コンバータBeanは、次のインタフェースを実装するJavaクラスです。

InboundMessageConverterインタフェース: インバウンドJSONメッセージをイベントに変換します。
OutboundMessageConverterインタフェース: イベントをJSONメッセージに変換します。

これらのAPIの詳細は、Oracle Stream Analytics Java APIリファレンスを参照してください。

インバウンドHTTP Pub-Sub JSONメッセージ

インバウンドHTTP pub-sub JSONメッセージ用のカスタム・コンバータBeanは、com.bea.wlevs.adapters.httppubsub.api.InboundMessageConverterインタフェースを実装します。このインタフェースは、convertメソッドのみが設定されています。

public List convert(JSONObject message) throws Exception;

messageパラメータは、JSON形式のインバウンドHTTP pub-subメッセージです。戻り値は、EPN内の次のノードに渡すイベントのListです。

アウトバウンドHTTP Pub-Sub JSONメッセージ

アウトバウンドHTTP pub-subメッセージ用のカスタム・コンバータBeanは、com.bea.wlevs.adapters.httppubsub.api.OutboundMessageConverterインタフェースを実装します。このインタフェースは、convertメソッドのみが設定されています。

public List<JSONObject> convert(Object event) throws Exception;

eventパラメータは、EPN内のソース・ノードから送信されてアウトバウンドHTTP pub-subアダプタによって受信されたイベントです。戻り値は、JSONメッセージのListです。

例

次の例は、InboundMessageConverterインタフェースとOutboundMessageConvertインタフェースの両方を実装するカスタム・コンバータBeanを示しています。このBeanは、インバウンドとアウトバウンドの両方のHTTP pub-subアダプタに対して使用できます。

注意:

JavaオブジェクトをJSON形式に変換するためにGSON Javaライブラリを使用できます。詳細は、http://www.json.orgおよびhttp://code.google.com/p/google-gsonを参照してください。

package com.sample.httppubsub;
import com.bea.wlevs.adapters.httppubsub.api.InboundMessageConverter;
import com.bea.wlevs.adapters.httppubsub.api.OutboundMessageConverter;
import com.bea.httppubsub.json.JSONObject;
import java.util.List;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
public class TestConverter implements InboundMessageConverter, OutboundMessageConverter {
  public List convert(JSONObject message) throws Exception {
    List eventCollection = new ArrayList();
    PubsubTestEvent event = new PubsubTestEvent();
    event.setMessage("From TestConverter: " + message);
    eventCollection.add(event);
    return eventCollection;
  }
  public List<JSONObject> convert(Object event) throws Exception {
    List<JSONObject> list = new ArrayList<JSONObject>(1);
    Map map = new HashMap();
    map.put("message", ((PubsubTestEvent) event).getMessage());
    list.add(new JSONObject(map));
    return list;
  }
}

4.10 JMSアダプタ

JMSアダプタを使用して、Oracle Stream AnalyticsのEPNにJava Message Service (JMS)を接続し、JMSメッセージを送受信します。

Oracle Stream Analytics JMSアダプタは、Java EEに準拠する、Javaクライアントを提供するJMSサービス・プロバイダをサポートします。

JMSインバウンド・アダプタが着信JMSメッセージをOracle Stream Analyticsイベントに変換し、JMSアウトバウンド・アダプタがOracle Stream AnalyticsイベントをJMSメッセージに変換します。インバウンド変換をカスタマイズするには、独自のJavaクラスを作成します。

JMSアダプタを作成する上で最適な方法は、Oracle JDeveloperコンポーネント・ウィンドウを使用する方法です。Oracle JDeveloperによって生成された次のアセンブリおよび構成ファイルは、JMSのインバウンド・アダプタおよびアウトバウンド・アダプタの構成を示しています。

注意:

アウトバウンドJMSアダプタに関連付けられたMessageConverterオブジェクトで発生する例外によって基礎となるJMSトランザクションがロールバックされることはありません。アウトバウンドJMSアダプタ内のMessageConverterの外部で例外が発生した場合、既存のJMSトランザクションがロールバックされます。

4.10.1 サービス・プロバイダ

Oracle Stream Analyticsは、次のサービス・プロバイダに対してテストされます。

Oracle T3プロトコルを使用してOracle WebLogic Serverと通信するJava RMIクライアントであるWebLogic T3クライアント
Oracle WebLogic Server JMSのバージョン10.0、10.3および10.3.1
Tibco EMS JMSの現在のバージョン

使用するサービス・プロバイダがリストにない場合は、使用しているサービス・プロバイダに連絡し、jms-adapter構成に必要なjndi-provider-urlおよびjndi-factoryの情報を取得することにより、使用しているサービス・プロバイダとともに使用するようOracle Stream Analytics JMSアダプタを構成できます。

4.10.2 インバウンド・アダプタ構成

アセンブリ・ファイル

<wlevs:adapter id="jms-inbound-adapter" provider="jms-inbound" />

構成ファイル

インバウンド・アダプタは、着信JMSメッセージをTradeEventに変換します。JNDIファクトリおよびサービス・プロバイダは、weblogic.jndi.WLInitialContextFactoryおよびt3://localhost:7101です。着信クライアントは、JNDI名がJNDINameであるアダプタを検索します。JMSアダプタは、JMSメッセージをイベントに変換した後、イベントをQueue1のJNDI宛先に送信します。

オプションのconnection-jndi-name要素は、JMS接続ファクトリのJNDI名を提供します。必須のdestination-jndi-name要素は、JMS宛先のJNDI名を提供します。session-transacted要素がfalseである場合、セッションがトランザクション・セッションではないことを示します。

  <jms-adapter>
    <name>jms-inbound-adapter</name>
    <event-type>TradeEvent</event-type>
    <jndi-provider-url>t3://localhost:7101</jndi-provider-url>
    <jndi-factory>weblogic.jndi.WLInitialContextFactory</jndi-factory>
    <connection-jndi-name>JNDIName</connection-jndi-name>
    <destination-jndi-name>Queue1</destination-jndi-name>
    <session-transacted>false</session-transacted>
  </jms-adapter>

4.10.2.1 シングルおよびマルチスレッドのインバウンドJMSアダプタ

デフォルトでは、インバウンドJMSアダプタはシングルスレッドです。インバウンドJMSアダプタがシングルスレッドであると、イベント順序が保証されます。

スケーラビリティを向上させるために、インバウンドJMSアダプタでマルチスレッドを使用してJMS宛先からメッセージを読み取るように構成できます。インバウンドJMSアダプタがマルチスレッドであると、イベント順序は保証されません。複数のスレッドを使用するには、work-manager子要素を使用してワーク・マネージャとともにアダプタを構成します。アダプタのみが使用する専用のワーク・マネージャを指定するか、他のアダプタやJettyなどのいくつかのコンポーネントでワーク・マネージャを共有できます。

4.10.2.2 恒久サブスクリプション用のJMSアダプタの構成

JMSトピックへの恒久サブスクリプション内のクライアントになるように、インバウンドJMSアダプタを構成できます。恒久サブスクリプションでは、アダプタが非アクティブになった場合も、アダプタはパブリッシュされたメッセージを確実に受信します。インバウンド・アダプタがJMSサーバーに接続すると、恒久サブスクリプションが登録されます。それ以降にトピックに送信されるメッセージは、サブスクライバが切断されている間も保持され(期限切れではない場合)、再接続すると配信されます。

恒久サブスクリプションは、JMSメッセージをトピックにパブリッシュしているパブリッシャが永続配信モードを使用していることを前提とします。パブリッシャはOracle Stream AnalyticsのアウトバウンドJMSアダプタである(つまりdelivery-modeの値は、デフォルト値のpersistentである必要がある)ことに注意してください。

アダプタ内の恒久サブスクリプションの作成

JMSメッセージ・パブリッシャが永続モードでメッセージを配信していることを確認します。
接続ファクトリのクライアントIDを指定します。Oracle WebLogic Serverで、管理上、コンソールを使用して接続ファクトリにクライアントIDを設定できます。恒久サブスクライバを使用している各アダプタ・インスタンスに専用の接続ファクトリが構成されている必要があります。
次の3つのjms-adapterプロパティを設定します。
- destination-typeをTOPICに設定します。
- durable-subscriptionをtrueに設定します。
- durable-subscription-nameを一意のサブスクリプション識別子に設定します。

4.10.3 アウトバウンド・アダプタ構成

アウトバウンドJMSアダプタは、イベントをJMSマップ・メッセージに変換し、JMSメッセージをJMS宛先に送信します。イベント・タイプが発信JMSメッセージに変換される方法を厳密に指定する独自のJavaクラスを作成することで、この変換をカスタマイズすることもできます。

アセンブリ・ファイル

<wlevs:adapter id="jms-outbound-adapter" provider="jms-outbound"/>

構成ファイル

JMSアウトバウンド・アダプタの構成は、JMSインバウンド・アダプタの構成とほぼ同じです。アウトバウンド・アダプタは、JMS宛先を指定し、JMS宛先にアクセスするためのユーザー名およびパスワードを提供します。このセッションはトランザクション・セッションであり、永続的ではありません。

  <jms-adapter>
    <name>jms-outbound-adapter</name>
    <event-type>TradeEvent</event-type>
    <jndi-provider-url>t3://localhost:7101</jndi-provider-url>
    <jndi-factory>weblogic.jndi.WLInitialContextFactory</jndi-factory>
    <connection-jndi-name>Topic</connection-jndi-name>
    <destination-jndi-name>Queue2</destination-jndi-name>
    <user>weblogic</user>
    <password>welcome1</password>
    <session-transacted>true</session-transacted>
    <delivery-mode>nonpersistent</delivery-mode>
  </jms-adapter>

4.11 JMSカスタム・メッセージ・コンバータBean

JMSメッセージとイベント・タイプ間の変換をカスタマイズするには、インバウンド・コンバータBeanおよびアウトバウンド・コンバータBeanを作成し、これらをOracle Stream Analyticsアプリケーションとともにパッケージ化します。

4.11.1 インタフェースの実装

インバウンド・コンバータBeanおよびアウトバウンド・コンバータBeanは、次の2つのインバウンド・インタフェースおよびアウトバウンド・インタフェースでメソッドを実装します。これらのAPIの詳細は、Oracle Stream Analytics Java APIリファレンスを参照してください。

インバウンド: com.bea.wlevs.adapters.jms.api.InboundMessageConverter。convertメソッドを実装する必要があります。return値は、ダウンストリームで渡されるイベントのListです。
```
public List convert(Message message) 
     throws MessageConverterException, JMSException;
```
messageパラメータ: 着信JMSメッセージに対応します。
アウトバウンド: com.bea.wlevs.adapters.jms.api.OutboundMessageConverterインタフェース。convertメソッドを実装する必要があります。return値は、JMSメッセージのListです。
```
public List<Message> convert(Session session, Object event) 
     throws MessageConverterException, JMSException;
```
sessionパラメータ: メッセージの作成に使用されるjavax.jms.Session。

eventパラメータ: EPN内のソース・ステージから送信されてアウトバウンドJMSアダプタによって受信されるイベント。

4.11.2 インバウンドJMSアダプタの実装

次の例では、インバウンドJMSアダプタ用のconvertメソッドを実装する方法を示します。

Oracle JDeveloperで、Javaクラスをアプリケーション・プロジェクトに追加します。

com.bea.wlevs.adapters.jms.api.InboundMessageConverterインタフェースを実装します。

次の例では、使用可能な実装を示します。

package com.customer;
import com.bea.wlevs.adapters.jms.api.InboundMessageConverter;
import com.bea.wlevs.adapters.jms.api.MessageConverterException;
import com.bea.wlevs.adapters.jms.api.OutboundMessageConverter;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.util.ArrayList;
import java.util.List;
public class MessageConverter implements InboundMessageConverter, 
    OutboundMessageConverter {
    public List convert(Message message) throws MessageConverterException, JMSException {
        TestEvent event = new TestEvent();
        TextMessage textMessage = (TextMessage) message;
        event.setString_1(textMessage.getText());
        List events = new ArrayList(1);
        events.add(event);
        return events;
    }
    public List<Message> convert(Session session, Object inputEvent) 
        throws MessageConverterException, JMSException {
        TestEvent event = (TestEvent) inputEvent;
        TextMessage message = session.createTextMessage(
            "Text message: " + event.getString_1()
        );
        List<Message> messages = new ArrayList<Message>();
        messages.add(message);
        return messages;
    }
}

アプリケーションEPNアセンブリ・ファイルでコンバータを指定します。
- bean要素を使用してコンバータ・クラスを登録します。
- nameをconverterBeanに設定し、refをbeanのidに設定したwlevs:instance-propertyを追加することで、コンバータ・クラスとJMSアダプタを関連付けます。
次の例では、コンバータ・クラスを登録して関連付ける方法を示します。
```
...
    <bean id="myConverter" class="com.customer.MessageConverter"/>
    <wlevs:adapter id="jmsInbound" provider="jms-inbound">
        <wlevs:instance-property name="converterBean" ref="myConverter"/>
        <wlevs:listener ref="mySink"/>
    </wlevs:adapter>
```

4.11.3 アウトバウンドJMSアダプタの実装

次の例では、アウトバウンドJMSアダプタ用のconvertメソッドを実装する方法を示します。

Oracle JDeveloper (または希望するIDE)を使用して、Javaクラスをアプリケーション・プロジェクトに追加します。

com.bea.wlevs.adapters.jms.api.OutboundMessageConverterインタフェースを実装します。

次の例では、使用可能な実装を示します。

package com.customer;
import com.bea.wlevs.adapters.jms.api.InboundMessageConverter;
import com.bea.wlevs.adapters.jms.api.MessageConverterException;
import com.bea.wlevs.adapters.jms.api.OutboundMessageConverter;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.util.ArrayList;
import java.util.List;
public class MessageConverter implements InboundMessageConverter, 
    OutboundMessageConverter {
    public List convert(Message message) throws MessageConverterException, JMSException {
        TestEvent event = new TestEvent();
        TextMessage textMessage = (TextMessage) message;
        event.setString_1(textMessage.getText());
        List events = new ArrayList(1);
        events.add(event);
        return events;
    }
    public List<Message> convert(Session session, Object inputEvent) 
        throws MessageConverterException, JMSException {
        TestEvent event = (TestEvent) inputEvent;
        TextMessage message = session.createTextMessage(
            "Text message: " + event.getString_1()
        );
        List<Message> messages = new ArrayList<Message>();
        messages.add(message);
        return messages;
    }
}

アプリケーションEPNアセンブリ・ファイルでコンバータを指定します。
- bean要素を使用してコンバータ・クラスを登録します。
- nameをconverterBeanに設定し、refをbeanのidに設定したwlevs:instance-propertyを追加することで、コンバータ・クラスとJMSアダプタを関連付けます。
次の例では、コンバータ・クラスを登録して関連付ける方法を示します。
```
<bean id="myConverter" class="com.customer.MessageConverter"/>
  <wlevs:adapter id="jmsOutbound" provider="jms-outbound">
    <wlevs:instance-property name="converterBean" ref="myConverter"/>
  </wlevs:adapter>
```

4.12 Oracle Business Rulesアダプタ

Oracle Business Rules (OBR)アダプタは、Oracle Business Rules製品のビジネス・ルール・エンジンをラップするイベントBeanです。OBRアダプタを使用すると、イベントをファクトとしてアサートおよび取消しを行い、ビジネス・ルールをトリガーできます。

イベントを生成し、ビジネス・ロジックをOracle CQLプロセッサのダウンストリームに追加し、イベントを処理するようOBRルールを構成できます。たとえば、ルール・ファイル内でStreamSender.sendInsertEventを起動し、OBRアダプタからデータをイベントとして送信できます。

Oracle JDeveloperでは、アセンブリ・ファイルまたはEPN図に対してドラッグ・アンド・ドロップ・コンポーネントは提供されていませんが、構成ファイルに対してはドラッグ・アンド・ドロップ・コンポーネントが提供されています。

アセンブリ・ファイルにエントリを追加すること、およびOBRアダプタを構成ファイルにドラッグ・アンド・ドロップすることで、OBRアダプタを作成できます。エントリの追加によるOBRアダプタ作成の詳細は、OBRドキュメント(http://www.oracle.com/technetwork/middleware/business-rules/documentation/index.html)を参照してください。

アセンブリ・ファイル

event-type-repository要素は、アプリケーションのイベント・タイプ・リポジトリを指定します。次の例では、リポジトリにはHelloWorldEventという名前の単一のイベント・タイプがあり、リポジトリはHelloWorldEvent.javaクラスによって実装されます。

次のアダプタは、HelloWorldAdapter Javaクラスの値を使用してhelloworldAdapter IDと等しいidを指定します。アダプタはHelloWorldAdapter Javaクラスから作成されます。OBRアダプタ構成には、指定したメッセージ・テキストとともにmessage要素が含まれます。HelloWorldAdapterクラスは、アプリケーションの実行時にmessageを出力します。この例では、HelloWorldAdapterクラスがイベント・ソースです。

チャネルおよびOracle CQLプロセッサの構成の後に、OBRアダプタの宣言である<wlevs:adapter id="OBRAdapter" provider = "obr">が続き、その後ろにdecisionFunctionおよびdictionaryURLプロパティが続きます。dictionaryURLプロパティはルールが含まれるOBRディクショナリ・ファイルへのパスであり、decisionFunctionプロパティは使用するOBR関数です。handler1プロパティは、この情報にアクセスするための他のコンポーネントのハンドルです。

注意:

OBRアダプタは、ファクトの自動取消しを処理しません。アップストリーム・プロセッサによってストリームが出力される場合、必要に応じて、またはルールの優先度に応じて最後のルールがトリガーされたら、ルール・ファイル内のファクトを取り消します。

最下部にあるのはHelloWorldBean構成です。HelloWorldBeanは、HelloworldEventおよびHelloWorldAdapterクラスをインスタンス化するJavaクラスです。

<wlevs:event-type-repository>
  <wlevs:event-type type-name="HelloWorldEvent">
    <wlevs:class>com.bea.wlevs.event.example.helloworld.HelloWorldEvent
    </wlevs:class>
  </wlevs:event-type>
</wlevs:event-type-repository>
<wlevs:adapter id="helloworldAdapter"
  class="com.bea.wlevs.adapter.example.helloworld.HelloWorldAdapter" >
  <wlevs:instance-property name="message" value="HelloWorld - The time is:"/>
</wlevs:adapter>
 
<wlevs:channel id="helloworldInputChannel" event-type="HelloWorldEvent" >
  <wlevs:listener ref="helloworldProcessor"/>
  <wlevs:source ref="helloworldAdapter"/>
</wlevs:channel>
 
<wlevs:processor id="helloworldProcessor" />
 
<wlevs:channel id="helloworldOutputChannel" event-type="HelloWorldEvent"
  advertise="true" max-threads="0" max-size="0" >
<wlevs:listener ref="OBRAdapter"/>
<wlevs:source ref="helloworldProcessor"/>
</wlevs:channel>
    
<wlevs:adapter id="OBRAdapter" provider = "obr">
<wlevs:instance-property name="decisionFunction" value="handler1" />
<wlevs:instance-property name="dictionaryUrl" value="file:helloworld.rules"/>
<wlevs:listener ref="OutputBean"/>    
</wlevs:adapter>
    
<wlevs:event-bean id="OutputBean"
  class="com.bea.wlevs.example.helloworld.HelloWorldBean">
</wlevs:event-bean>

構成ファイル

この構成ファイルは、OBRアダプタから受信したHelloworldEventを処理するために使用される問合せルールおよびOracle CQLプロセッサを宣言します。また、OBRルールにアクセスするためのOBRアダプタのハンドラ(handler1)も提供します。dictionary-url要素は、使用する決定関数およびルールが含まれるOBRディクショナリ・ファイルへのパスを指定します。decision-function要素は、使用するOBR決定関数の名前を指定します。

<processor>
  <name>helloworldProcessor</name>
  <rules>
    <query id="helloworldRule">
      <![CDATA[ select * from helloworldInputChannel[range 10 slide 5] ]] >
       select * from helloworldInputChannel[now]
    </query>
  </rules>
</processor>
<obr-adapter>
  <name>OBRAdapter</name>
  <dictionary-url>file:helloworld.rules</dictionary-url>
  <decision-function>handler1</decision-function>
</obr-adapter>

4.13 QuickFixアダプタ

QuickFixアダプタは、Financial Information eXchange (FIX)標準に従ってセキュリティ・トランザクションのリアルタイムな電子交換を処理するためのフル機能のメッセージング・エンジンです。

QuickFixアダプタは、FIXメッセージのリスニング、FIXメッセージのOracle Stream Analyticsイベントへの変換、および次のステージで処理を行うためのイベント送信を行います。

QuickFixアダプタは、基礎となるQuickFIXエンジンが提供する構成データをすべてサポートしています。FIX構成データの詳細は、『QuickFIX/J User Manual』(http://www.quickfixj.org/documentation/)を参照してください。

注意:

12cリリースでは、Oracle Stream AnalyticsはQuickFIX動的アクセプタ・セッションをサポートしません。

QuickFixアダプタはSSLをサポートしていません。

4.13.1 サポートされているQuickFixバージョンおよびサポートされていないメッセージ・タイプ

QuickFIXエンジンは、広範囲なメッセージ・タイプをサポートし、これらのメッセージ・タイプ内のデータを検証します。QuickFIXエンジンでサポートされていないメッセージがOracle Stream Analyticsアプリケーションに必要である場合、デフォルトのQuickFIXメッセージ・ハンドラを拡張し、適切なハンドラ・メソッドをオーバーライドすることにより、カスタム・ハンドラを作成する必要があります。

QuickFIXアダプタは、次のQuickFIXバージョンをサポートしています。各QuickFIXバージョンの下にリストされているメッセージはサポートされていません。

QuickFIXバージョン4.0

次のメッセージ・タイプはこのバージョンでサポートされていません。