4 Adapters

Adapters manage data entering and leaving the EPN. Oracle Stream Analytics provides a number of different kinds of inbound and outbound adapters to handle different types of data such as CSV, RMI, and HTTP. All adapters have a provider property that is a reference to the OSGi-registered adapter factory service and defines the type of data that the adapter handles.

Inbound adapters receive event data from a data stream entering the EPN, assign the data to an event according to the event type, and send the data to the next stage in the EPN. Outbound adapters receive events processed by the EPN, convert the events to their output form, and send the converted data to an output data source such as another EPN, a non-EPN application, a CSV file, or a web page.

This chapter includes the following sections:

For information about the high availability adapters, see High Availability Applications.

See Testing 1-2-3 for information about how to use the csvgen adapter with the load generator to simulate a data feed to test your application.

4.1 Create Adapters

The best way to create most adapters is with Oracle JDeveloper. The Oracle JDeveloper components window provides the following inbound and outbound adapters: CSV, EDN, RMI, HTTP, and JMS.

For the other adapters, edit the configuration files directly. Each adapter section in this chapter provides example assembly and configuration file configurations so that you can see the settings.

Before you create an adapter, use Oracle JDeveloper to create an event type to assign to the adapter. See Create and Register an Event Type in Getting Started with Event Processing for Oracle Stream Analytics for information about creating event types.

This chapter describes some of the assembly and configuration file settings for the different types of adapters. For complete information about adapter settings, see adapter in Schema Reference for Oracle Stream Analytics. See also the Oracle/Middleware/my_oep/oep/wlevs_application_config.xsd directory in your Oracle Stream Analytics installation for adapter schema information.

4.2 Cluster Distribution Service

The cluster distribution feature provides a mechanism for various Oracle Stream Analytics adapter types to distribute incoming events to all of the servers in a cluster.

An individual adapter or event bean instance can be configured to distribute events, and in this case, all input events processed by that adapter are sent (distributed) to all servers in the cluster.

The distribution adapter ensures that all input events are sent (distributed) to all servers in the cluster. To convert an input adapter to a distribution adapter, add the distributeInput element and set it to true as follows. The distributionThreadsCount property is optional and defaults to 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 supports the Cluster Distribution service for the loadgen, CSV inbound, and JMS inbound (queue) adapters. Oracle Stream Analytics does not support the Cluster Distribution service for the CSV outbound, JMS inbound (topic), JMS outbound, and HTTP publish-subscribe adapters. It is an error to configure topic destinations for input distribution. A topic configuration generates a warning log message and is ignored.

In addition to the adapter types listed above, you can also configure an event bean to distribute all of the events it receives to all cluster members by specifying the provider for the bean to be clusterGroupDistributor as shown in the following example:

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

Oracle JDeveloper does not currently provide a component for the Cluster Distribution service. However, you can create a Cluster Distribution service by adding entries to the assembly and configuration files for your Oracle Stream Analytics application.

4.3 Password Encryption

Some of the adapters have user name and password child elements. Oracle Stream Analytics provides the encryptMSAConfig command so that you can encrypt the file that contains the password. See encryptMSAConfig Command-Line Utility in Administering Oracle Stream Analytics for more information.

4.4 JAXB Support

Oracle Event Processing provides a simplified interface for using Java Architecture for XML Binding (JAXB) mapping capabilities in adapters and event beans to marshall and unmarshall event data between XML and Java objects.

The JAXB interface supports the JAXB 2.2 specification and EclipseLink Moxy provider extensions.

You can configure the mapping operations in the following ways:

Map from an XML schema to Java objects to output a set of annotated Java classes.
Map from one set of Java objects to another set of Java Objects or to XML using JAXB annotations.
Map from an existing XML schema to an existing, predefined Java object representation. This approach uses the EclipseLink Moxy extensions and requires an external metadata file that contains the mapping details. The metadata file is referenced by the application configuration.

4.4.1 EclipseLink Moxy

EclipseLink Moxy provides extensions that enable you to map between an existing XML schema and a predefined set of Java classes without modifying the XML schema or the Java classes without providing annotations. You provide the mapping information in an external metadata file using a XPath syntax.

The flexible EclipseLink Moxy extensions enable you to perform complex operations. For example, you can map a subset of complex XML data to a much simpler event representation. You can also flatten a deeply nested XML document into a flat Java bean event format for processing by Oracle CQL.

You specify EclipseLink Moxy external metadata in XML. Access the schema at: http://www.eclipse.org/eclipselink/xsds/eclipselink_oxm_2_2.xsd.

4.4.2 APIs

The adapter or event bean that requires JAXB functionality obtains the functionality by injection of a bean that implements the com.oracle.cep.mappers.api.Mapper interface. The Mapper interface follows:

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

The adapter or other EPN component code uses the injected bean to create marshalling and unmarshalling objects. The com.oracle.cep.mappers.api.Marshaller and com.oracle.cep.mappers.api.Unmarshaller interfaces shown below work for most applications.

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;
}

Some applications might need specialized method signatures for marshalling and unmarshalling such as an unmarshall method that takes the target class as an argument. In these cases, use the com.oracle.cep.mappers.jaxb.JAXBMarshallerImpl and com.oracle.cep.mappers.jaxb.JAXBUnmarshallerImplinterfaces instead. These interfaces provide methods that correspond to the full set of marshall and unmarshall methods that are supported by the javax.xml.bind.Marshaller and javax.xml.bind.Unmarshaller interfaces.

Assembly File

The following assembly file entries call a mapper bean with properties to specify the event type and the metadata file.

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

If you want to call a factory to make the mapper bean, specify the following for the bean element:

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

Configuration File

The following configuration file entries specify properties for the mapper bean.

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

Properties

A mapper bean supports properties. All of the properties except metadataMap can be configured as assembly file properties or as elements in the configuration file.

Table 4-1 Mapper Bean Properties and Elements

Assembly File Property Name	Configuration File Element Name	Description
eventTypeName	event-type-name	The name of an event type registered by the application in the event type repository. The event type corresponds to a Java class. The package name of this class is used as the context path when initializing the JAXBContext represented by the mapper bean. The configuration must specify either an `eventTypeName` or a `contextPath` to be used in constructing the context path for the `JAXBContext` represented by the mapper. The packages must exist on the `classpath` of the application and contain either schema generated classes, JAXB annotated classes, or classes referenced by Moxy external metadata. The classes will be used as the Java object graph for marshalling and unmarshalling operations. See context-path inSchema Reference for Oracle Stream Analytics for more information.
`contextPath`	`context-path`	A colon-separated list of Java package names. The specified context path to initialize the `JAXBContext` represented by the mapper bean. The configuration must specify either an `eventTypeName` or a `contextPath` to be used in constructing the context path for the `JAXBContext` represented by the mapper. The packages must exist on the `classpath` of the application and contain either schema generated classes, JAXB annotated classes, or classes referenced by Moxy external metadata. The classes will be used as the Java object graph for marshalling and unmarshalling operations.
`validate`	`validate`	`Boolean` value that defaults to `false`. When `true`, you must provide the `schema`. Schema validation occurs during marshalling and unmarshalling.
`schema`	`schema`	The file name of the XML schema file used for validation. Package the schema file with the application in the `META-INF/wlevs/mappers/jaxb` directory.
`metadata`	`metadata`	The name of the file that contains the EclipseLink Moxy external metadate for mapping customization. Package the file with the application in the `Meta-INF/wlevs/mappers/jaxb directory`.
`metadataMap`	N/A	The Spring `<map>` element that contains one entry that corresponds to each component of the `contextPath`. For each entry, the key is the package name from the `contextPath` and the corresponding value is the name of a file that contains the EclipseLink Moxy external `metadata` for that package.

If the application uses EclipseLink Moxy-specific external metadata, the location of the metadata is specified by either the metadata property or the metadataMap property. The metadataMap property is required when there is more than one package on the contextPath. There is no support for specifying the metadataMap property in a component configuration file.

4.5 CSV Adapters

CSV adapters handle inbound and outbound data that is separated by commas. Use a CSVInbound adapter to accept data in the form of comma-separated values entering the EPN, and use a CSVOutbound adapter to send data in comma-separated values out of the EPN.

You can test an Oracle Stream Analytics Application that uses CSV inbound adapters with the load generator provided in your Oracle Stream Analytics installation. The load generator reads an ASCII file that contains sample data. You must use the CSV Inbound adapter because it is coded to decipher the data packets generated by the load generator. See Load Generator and the csvgen Adapter.

Note:

With the java.sql.TimeStamp type, the CSV adapter reads and writes data in the format yyyy-mm-dd‘T‘hh:mm:ss[.fffffffff]. For example, "2012-12-12T12:12:12.120".

The best way to create CSV adapters is through the Oracle JDeveloper components window. The following assembly and configuration files generated by Oracle JDeveloper show the CSV inbound and outbound adapter configurations.

Assembly File

The inbound CSV adapter translates data read from the StockData.csv file into an event with the TradeEvent event type.

The wlevs:listener element specifies the component that listens to the inbound CSV adapter for TradeEvent activity. In this example, the listener is AdapterOutputChannel. The AdapterOutputChannel component listens for and receives trade events from StockTradeCSVInboundAdapter to send to the next stage.

<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>

The outbound adapter assembly file configuration is similar to the inbound adapter, but includes an append attribute. When set to true, Oracle Stream Analytics appends data to an existing output file. When set to false, Oracle Stream Analytics creates a new file or overwrites an existing file of the same name.

 <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>

You can provide an absolute or relative path for the outputFile value. For the relative path, you can specify ../filename.csv. ./result.csv, or upload/result.csv. When you specify a relative path, make sure that the abstract path includes the parent directory. For example, in UNIX, specify a file in the current directory as ./result.csv instead of simply result.csv.

Configuration File

The adapter elements in the configuration file show the adapter name attribute and its value. The adapter name must match the adapter id attribute in the assembly file.

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

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

4.6 EDN Adapters

Event Delivery Network (EDN) inbound and outbound adapters use JAXB to enable an EPN to interface with an Oracle SOA Suite event network.

The EDN adapters have a raw-xml-content configuration element that specifies whether to represent the EDN XML data transmission as raw XML (if true) or as a Java object using JAXB. In the JAXB case, the adapter expects the Oracle Stream Analytics application bundle to include the appropriate set of schema (xjc) generated classes on its class path.

You configure an EDN adapter with an event type and a reference to an EDL file. During initialization, the adapter searches the EDL file that contains an event definition QName that matches the configured event type. If the configured event type is found in the EDL, the adapter registers a subscription with EDN for the corresponding QName.

To learn how to use EDN adapters and an EDL file, see Create an Application with EDN Adapter in Getting Started with Event Processing for Oracle Stream Analytics.

4.6.1 Usage

Use an EDNInbound adapter to receive incoming data from the Oracle SOA Suite event network. The EDN input adapter subscribes to a specified EDN event type and converts the incoming EDN events to an Oracle Stream Analytics event type for processing by an Oracle Stream Analytics application.

Use an EDNOutbound adapter to send outbound data to the Oracle SOA Suite event network. The EDN output adapter converts the Oracle Stream Analytics events into corresponding EDN events and publishes them to the EDN. The published events can be new events that originate in the Oracle Stream Analytics application or EDN events that were received by the EDN input adapter, processed by the Oracle Stream Analytics application, and sent to the EDN output adapter.

4.6.2 Create EDN Adapters

The best way to create EDN adapters is through the Oracle JDeveloper components window. The following assembly and configuration files generated by Oracle JDeveloper show the EDN inbound and outbound adapter configurations.

See edn-adapter in Schema Reference for Oracle Stream Analytics for information about the EDN adapter.

Assembly File

The following assembly file entries show the elements and attribute settings for the inbound and outbound EDN adapters created in the Fraud Detection walkthrough from Fraud Detection Application in Getting Started with Event Processing for Oracle Stream Analytics.

The input EDN adapter listens to the EDN input channel for events of type FraudCheckRequest.
The EDN output adapter sends events of type FraudCheckRequest to the next stage in the 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>

Configuration file

The following configuration file entries show the configuration settings for the EDN inbound and outbound EDN adapters created in the Fraud Detection walkthrough from Fraud Detection Application in Getting Started with Event Processing for Oracle Stream Analytics.

Note:

You must put the EDL and schema (xsd) files in the fixed path of the bundled JAR file.

<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 File Adapter

The File adapter reads data from a file into the EPN and converts the data to an event.

Oracle JDeveloper does not currently provide a component for the File adapter. However, you can create a File adapter by adding entries to the assembly and configuration files for your Oracle Stream Analytics application.

Assembly File

The path property provides the location of the input file. As the adapter reads the data from the input file, it converts the incoming data to an event of type OrderArrivalEvent. There is an initialDelay of 5000 nanoseconds before the File adapters starts to read the file. The downstream OrderArrival channel listens for events of type 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>

Configuration File

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

4.8 HTTP Publish-Subscribe Adapter

Use the HTTP Publisher adapter to send JavaScript Object Notation (JSON) event data out of the EPN to a web-based user interface. Use the HTTP Subscriber adapter to accept JavaScript Object Notation (JSON) event data entering the EPN. JSON event data comes from an HTTP server where user actions generate events.

The HTTP Publish-Subscribe server in Oracle Stream Analytics is based on the Bayeux protocol that is proposed by the cometd project. The Bayeux protocol defines a contract between the client and the server for communicating with asynchronous messages over HTTP.

You can create a remote or a local HTTP Publisher adapter, and a remote HTTP Subscriber adapter. Whether an HTTP adapter is local or remote is determined by the local or remote URL you supply to the required <server-url> child element.

The best way to create HTTP publish and subscribe adapters is to use Oracle JDeveloper.

Note:

Byte arrays are not supported as property types in event types used with the HTTP Publish and Subscribe adapters.

Assembly File

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

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

Configuration File

For every local and remote adapter, provide a URL to the server in the server-url property. The server can be an Oracle Stream Analytics server, a WebLogic Server instance, or any third-party HTTP Publish-Subscribe server.

For every local adapter for publishing, add a server-context-path element to specify the path to the local HTTP Publish-Subscribe server associated with the Oracle Stream Analytics instance hosting the current Oracle Stream Analytics application.

By default, each Oracle Stream Analytics server is configured with an HTTP Publish-Subscribe server with path /pubsub. If you have created a new local HTTP Publish-Subscribe server or changed the default configuration, then specify the location of the server in the server file. In the file specify the http-pubsub element path value with the location of the server. You can locate the server file in your Oracle Stream Analytics installation at: /Oracle/Middleware/my_oep/examples/domains/<my_domain>/defaultserver.

The channel child element specifies the channel that the adapter publishes or subscribes to.

<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 Publish-Subscribe Adapter Custom Converter Bean

The HTTP Publish-Subscribe adapter converts incoming JavaScript Object Notation (JSON) messages to event types and back again. To customize the way inbound and outbound JSON messages are converted to an event type and back, create a custom converter bean.

4.9.1 Bayeux Protocol

The HTTP Publish-Subscribe (pub-sub) server is based on the Bayeux protocol that is proposed by the cometd project. The Bayeux protocol defines a contract between the client and the server for communicating with asynchronous messages over HTTP. The pub-sub server can communicate with any client that understands the Bayeux protocol.

You can develop your web client with the following frameworks:

Dojo JavaScript library that supports the Bayeux protocol. Oracle Stream Analytics does not provide this library. You can find information about it at: http://dojotoolkit.org/.
WebLogic Workshop Flex plug-in that enables development of a Flex client that uses the Bayeux protocol to communicate with a pub-sub server.

For information about securing an HTTP pub-sub server channel, see HTTP Publish-Subscribe Server in Administering Oracle Stream Analytics.

4.9.2 Create a Custom Converter Bean

A custom converter bean is a Java class that implements the following interfaces:

InboundMessageConverter interface to convert inbound JSON messages to events.
OutboundMessageConverter interface to convert events to JSON messages.

See the Java API Reference for Oracle Stream Analytics for a full description of these APIs.

Inbound HTTP Pub-Sub JSON Message

The custom converter bean for an inbound HTTP pub-sub JSON message implements the com.bea.wlevs.adapters.httppubsub.api.InboundMessageConverter interface. This interface has only the convert method:

public List convert(JSONObject message) throws Exception;

The message parameter is the inbound HTTP pub-sub message in JSON format. The return value is a List of events to pass to the next node in the EPN.

Outbound HTTP Pub-Sub JSON Message

The custom converter bean for an outbound HTTP pub-sub message implements the com.bea.wlevs.adapters.httppubsub.api.OutboundMessageConverter interface. This interface has only the convert method:

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

The event parameter is an event received by the outbound HTTP pub-sub adapter from the source node in the EPN. The return value is a List of JSON messages.

Example

The following example shows a custom converter bean that implements both the InboundMessageConverter and OutboundMessageConvert interfaces. You can use this bean for both inbound and outbound HTTP pub-sub adapters.

Note:

You can use the GSON Java library to convert Java objects to JSON format. For more information, see http://www.json.org and 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 Adapters

Use JMS adapters to connect the Java Message Service (JMS) with an Oracle Stream Analytics EPN to receive and send JMS messages.

The Oracle Stream Analytics JMS adapters support any JMS service provider that provides a Java client that is compliant with Java EE.

The JMS Inbound adapter converts the incoming JMS messages to Oracle Stream Analytics events and the JMS outbound adapter converts Oracle Stream Analytics events to JMS messages. You can customize the inbound conversion by writing your own Java class. See Custom Adapters in Customizing Oracle Stream Analytics.

The best way to create JMS adapters is through the Oracle JDeveloper components window. The following assembly and configuration files generated by Oracle JDeveloper show the JMS inbound and outbound adapter configurations.

Note:

An exception that occurs in the MessageConverter object associated with a outbound JMS adapter does not cause the underlying JMS transaction to roll back. If the exception occurs outside of the MessageConverter object within the outbound JMS adapter, then an existing JMS transaction is rolled back.

4.10.1 Service Providers

Oracle Stream Analytics is tested against the following service providers:

WebLogic T3 Client, which is a Java RMI client that uses Oracle T3 protocol to communicate with Oracle WebLogic Server.
Version 10.0, 10.3, and 10.3.1 of Oracle WebLogic Server JMS
The current version of Tibco EMS JMS

If the service provider you want to use is not in the list, you can configure Oracle Stream Analytics JMS adapters for use with your service provider by contacting your service provider and getting the jndi-provider-url and jndi-factory information needed for the jms-adapter configuration.

4.10.2 Inbound Adapter Configuration

Assembly File

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

Configuration File

The inbound adapter converts incoming JMS messages to a TradeEvent. The JNDI factory and service provider are weblogic.jndi.WLInitialContextFactory and t3://localhost:7101. The incoming client finds the adapter with the JNDI name of JNDIName. After the JMS adapter converts the JMS message to an event, the adapter sends the events to the JNDI destination of Queue1.

The optional connection-jndi-name element provides the JNDI name of the JMS connection factory. The required destination-jndi-name element provides the JNDI name of the JMS destination. The session-transacted element when false indicates that the session is not transactional.

  <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 Single and Multithreaded Inbound JMS Adapters

By default, an inbound JMS adapter is single-threaded. When the inbound JMS adapter is single-threaded, event order is guaranteed.

To improve scalability, you can configure an inbound JMS adapter to use multiple threads to read messages from the JMS destination. When the inbound JMS adapter is multi-threaded, event order is not guaranteed. To use multiple threads, configure the adapter with a work manager with the work-manager child element. You can specify a dedicated work manager to be used only by the adapter, or you can share a work manager among several components such as other adapters and Jetty.

4.10.2.2 Configure a JMS Adapter for Durable Subscriptions

You can configure an inbound JMS adapter to be a client in a durable subscription to a JMS topic. A durable subscription ensures that the adapter receives published messages even when the adapter becomes inactive. When the inbound adapter connects to the JMS server, it registers the durable subscription and subsequent messages sent to the topic are retained during periods when the subscriber is disconnected (unless they expire) and delivered when the subscriber reconnects.

A durable subscription assumes that the publisher that is publishing JMS messages to the topic is using the persistent delivery mode. Note that publisher might be the Oracle Stream Analytics outbound JMS adapter (in other words, its delivery-mode value must be persistent, the default value).

Create a Durable Subscription in the Adapter

Ensure that the JMS message publisher is delivering messages in persistent mode.
Specify a client ID for the connection factory. On Oracle WebLogic Server, the client ID can be set on the connection factory administratively with the console. You should have a dedicated connection factory configured for each adapter instance that is using durable subscribers.
Set the following three jms-adapter properties:
- destination-type to TOPIC.
- durable-subscription to true.
- durable-subscription-name to a unique subscription identifier.

4.10.3 Outbound Adapter Configuration

The outbound JMS adapter converts events into a JMS map message and sends the JMS message to a JMS destination. You can also customize this conversion by writing your own Java class to specify exactly how you want the event types to be converted into outgoing JMS messages. See Custom Adapters in Customizing Oracle Stream Analytics.

Assembly File

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

Configuration File

The JMS Outbound adapter configuration is almost the same as the JMS Inbound adapter configuration. The outbound adapter specifies a JMS destination and provides a user name and password to access the JMS destination. This session is transactional and non-persistent.

  <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 Custom Message Converter Bean

To customize the conversion between JMS messages and event types, create inbound and outbound converter beans and package them with your Oracle Stream Analytics Application.

4.11.1 Implement Interfaces

The inbound and outbound converter beans implement methods in the following two inbound and outbound interfaces. See the Java API Reference for Oracle Stream Analytics for a full description of these APIs.

Inbound: com.bea.wlevs.adapters.jms.api.InboundMessageConverter. You have to implement its convert method. The return value is a List of events to be passed downstream.
```
public List convert(Message message) 
     throws MessageConverterException, JMSException;
```
message parameter: Corresponds to the incoming JMS message.
Outbound: com.bea.wlevs.adapters.jms.api.OutboundMessageConverter interface. You have to implement its convert method. The return value is a List of JMS messages.
```
public List<Message> convert(Session session, Object event) 
     throws MessageConverterException, JMSException;
```
session parameter: The javax.jms.Session to use to create the messages.

event parameter: An event received by the outbound JMS adapter from the source stage in the EPN.

4.11.2 Implement the Inbound JMS Adapter

This example shows you how to implement the convert method for the inbound JMS adapter.

In Oracle JDeveloper, add a Java class to your application project.

Implement the com.bea.wlevs.adapters.jms.api.InboundMessageConverter interface.

The following example shows a possible implementation.

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;
    }
}

Specify the converter in your application EPN assembly file.
- Register the converter class using a bean element.
- Associate the converter class with the JMS adapter by adding a wlevs:instance-property with name set to converterBean and ref set to the id of bean.
The following example shows how to register and associate the converter class.
```
...
    <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 Implement the Outbound JMS Adapter

This example shows you how to implement the convert method for the outbound JMS adapter.

Using the Oracle JDeveloper (or your preferred IDE), add a Java class to your application project.

Implement the com.bea.wlevs.adapters.jms.api.OutboundMessageConverter interface.

The example shows a possible implementation.

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;
    }
}

Specify the converter in your application EPN assembly file.
- Register the converter class using a bean element.
- Associate the converter class with the JMS adapter by adding a wlevs:instance-property with name set to converterBean and ref set to the id of bean.
The following example shows how to register and associate the converter class.
```
<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 Adapter

The Oracle Business Rules (OBR) adapter is an event bean that wraps the business rules engine from the Oracle Business Rules product. The OBR adapter lets you assert and retract events as facts to trigger business rules.

You can configure OBR rules to generate events and add business logic to an Oracle CQL processor downstream to process the events. For example you can invoke StreamSender.sendInsertEvent within the rules file to send data out of an OBR adapter as an event.

Oracle JDeveloper does not provide a drag and drop component for the assembly file or the EPN diagram, but it does provide a drag-and-drop component for the configuration file.

You can create an OBR adapter by adding entries to the assembly file and by dragging and dropping the OBR adapter into the configuration file. For more information about creating OBR adapters by adding entries, see the OBR documentation at: http://www.oracle.com/technetwork/middleware/business-rules/documentation/index.html.

Assembly File

The event-type-repository element specifies the event type repository for the application. In the following example, the repository has a single event type named HelloWorldEvent and is implemented by the HelloWorldEvent.java class.

The next adapter specifies an id equal to helloworldAdapter ID with a value of the HelloWorldAdapter Java class. An adapter is created from the HelloWorldAdapter Java class. The OBR adapter configuration includes a message element with the specified message text. The HelloWorldAdapter class prints the message during application execution. In this example, the HelloWorldAdapter class is the event source.

The OBR adapter declaration comes after the channel and Oracle CQL processor configurations: <wlevs:adapter id="OBRAdapter" provider = "obr"> followed by the decisionFunction and dictionaryURL properties. The dictionaryURL property is the path to the OBR dictionary file that contains the rules, and decisionFunction property is the OBR function you want to use. The handler1 property is a handle for other components to access this information.

Note:

The OBR adapter does not handle automatic Fact retraction. If the upstream processor outputs a stream, retract the Fact in the rule file when appropriate or when the last rule is triggered according to rule priority.

At the bottom is the HelloWorldBeand configuration. The HelloWorldBean is a Java class that instantiates the HelloworldEvent and HelloWorldAdapter classes.

<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>

Configuration File

The configuration file declares the Oracle CQL processor and query rules to use to process the HelloworldEvent received from the OBR adapter. It also provides the OBR adapter handler (handler1) for accessing the OBR rules. The dictionary-url element specifies the path to the OBR dictionary file that contains the rules and decision function you want to use. The decision-function element specifies the name of the OBR decision function you want to use.

<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 Adapter

The QuickFix Adapter is a full-featured messaging engine for handling the real-time electronic exchange of securities transactions according to the Financial Information eXchange (FIX) standard.

The QuickFix adapter listens for FIX messages, converts the FIX messages to Oracle Stream Analytics events, and sends the events to the next stage for processing. See Sample Application with QuickFix Adapter in Getting Started with Event Processing for Oracle Stream Analytics for an example application.

The QuickFix adapter supports all of the configuration data that the underlying QuickFIX engine provides. For information about FIX configuration data, see the “QuickFIX/J User Manual" at http://www.quickfixj.org/documentation/.

Note:

Oracle Stream Analytics does not support QuickFIX dynamic acceptor sessions in the 12c release.

The QuickFix adapter does not support SSL.

4.13.1 Supported QuickFIX Versions and Unsupported Message Types

The QuickFIX engine supports a wide-range of message types and validates the data in those message types. If your Oracle Stream Analytics application requires a message that is not supported by the QuickFIX engine, you must create a custom handler by extending the default QuickFIX message handler and overriding the appropriate handler method(s).

The QuickFIX adapter supports the following QuickFIX versions. The messages listed under each QuickFIX version are not supported.

QuickFIX Version 4.0

The following message types are not supported in this version: