Application Development Guide

Using Oracle CEP Caching

This section contains information on the following subjects:

Overview of Oracle Complex Event Processing Caching

Oracle Complex Event Processing (or Oracle CEP for short ) applications can optionally publish or consume events to and from a cache to increase the availability of the events and increase the performance of their applications. A cache is a temporary storage area for events, created exclusively to improve the overall performance of your application; it is not necessary for the application to function correctly.

Oracle CEP includes its own caching implementation for local single-JVM caches; this implementation uses an in-memory store. If you require a different kind of cache, such as one that stores to disk, then you can plug-in different cache providers, such as Oracle Coherence.

In this document, a caching system refers to a configured instance of a caching implementation, either the local one included in the product or a third-party one. A caching system defines a named set of configured caches as well as the configuration for remote communication if any of the caches are distributed across multiple machines.

A caching system is always configured at the application level, regardless of its implementation (Oracle CEP or third-party). Other Oracle CEP applications can use the caching system, but they must be deployed after the application that owns the caching system to ensure that the caching system is made available to them.

A cache can be considered a stage in the event processing network in which an external element (the cache) consumes or produces events; this is similar to an adapter that uses a JMS destination. You configure a caching system, along with one or more caches, for an application declaratively using the EPN assembly file. A cache, however, does not have to be an actual stage in the network; another component or Spring bean can access a cache programmatically using the caching APIs.

The next sections describe specific use cases the Oracle CEP caching feature.

Use Case: Publishing Events to a Cache

An example of this use case is a financial application that publishes events to a cache while the financial market is open and then processes data in the cache after the market closes.

Publishing events to a cache makes them highly available or available to other Oracle CEP applications running in the server. Publishing events to a cache also allows for asynchonous writes to a secondary storage by the cache implementation. You can configure any stage in an Oracle CEP application that generates events (input adapter, stream, business POJO, or processor) to publish its events to the cache.

Use Case: Consuming Data From a Cache

Oracle CEP applications may sometimes need to access non-streaming data in order to do its work; caching this data can increase the performance of the application.

The standard components of an Oracle CEP application that are allowed direct programming access to a cache are input- and output-adapters and business POJOs.

Additionally, applications can access a cache from EPL, either by a user-defined EPL function or directly from an EPL statement. In the case of a user-defined EPL function, programmers use the caching API to inject the cache resource into the implementation of the function. Applications can also query a cache directly from an EPL statement that both runs in a processor or is submitted programmatically to the cache. In this case, the cache essentialy functions as another type of stream data source to a processor so that querying a cache is very similar to querying a stream.

An example of using EPL to query a cache is from a financial application that publishes orders and the trades used to execute the orders to a cache. At the end of the day when the markets are closed, the application queries the cache in order to find all the trades related to a particular order.

Use Case: Updating and Deleting Data in a Cache

An Oracle CEP application can both update and delete data in a cache when required.

For example, a financial application may need to update an order in the cache each time individual trades that fulfill the order are executed, or an order may need to be deleted if it has been cancelled. The components of an application that are allowed to consume data from a cache are also allowed to update it.

Additional Caching Features

In addition to the major caching features described in the preceding sections, Oracle CEP caching includes the following features:

Caching APIs

Oracle CEP provides a number of caching APIs that you can use in your application to perform certain tasks. The APIs are in two packages:

The creation, configuration, and wiring of the caching systems and caches is all done using the EPN assembly file and component configuration files. This means that you typically never explicitly use the Cache and CachingSystem interfaces in your application; the only reason to use them is if you have additional requirements than the standard configuration. For example, if you want to provide integration with a third-party cache provider, then you must use the CachingSystem interface. If you want to perform operations on a cache that are not part of the java.util.Map interface, then you can use the Cache interface.

If you create cache listeners, loaders, or stores, then the Spring bean must implement the CacheListener, CacheLoader, or CacheStore interfaces. The following sections describe additional details.

Typical Steps to Use Oracle CEP Caching

Depending on how you are going to use the cache, there are different tasks that you must perform, as described in the following procedure that in turn point to sections with additional details.

Configuring the Oracle CEP Caching System and Caches

You configure a caching system and its caches in the caching configuration file, similar to how you configure other components in the event processing network such as processors and adapters. For general information about these configuration files, see Component Configuration Files.

If you use a third-party caching implementations such as GemFire, you can configure only one caching system per Oracle CEP instance. If, however, you use the caching implementation provided by Oracle CEP, then you can configure multiple caching systems per application.

The following procedure describes the main steps to configure the caching system provided by Oracle CEP for your application. For simplicity, it is assumed in the procedure that you are going to configure all components of an application, including the caching system, in a single configuration XML file and that you have already created this file for your application.

See XSD Schema Reference for Component Configuration Files for the complete XSD Schema that describes the caching system configuration file elements.

Declaring the Caching System in the EPN Assembly File

To declare a caching system that uses the Oracle CEP implementation declaratively in the EPN assembly file, use the <wlevs:caching-system> tag without any additional attributes, as shown in the following example:

  <wlevs:caching-system id="caching-system-id"/>

The value of the id attribute must match the name specified for the caching system in the external configuration metadata. If the application allows other applications to use the caching system, it can specify that the caching system be advertised using the advertise attribute (by default set to false):

  <wlevs:caching-system id="caching-system-id" advertise="true"/>

You also use the <wlevs:caching-system> tag to declare a third-party implementation; use the class or provider attributes to specify additional information.

For simplicity, you can include the third-party implementation code inside the Oracle CEP application bundle itself to avoid having to import or export packages and managing the lifecycle of a separate bundle that contains the third-party implementation. In this case the <wlevs:caching-system> tag appears in the EPN assembly file as shown in the following example:

  <wlevs:caching-system id="caching-system-id" 
                         class="third-party-implementation-class"/>

The class attribute specifies a Java class that must implement the com.bea.wlevs.cache.api.CachingSystem interface. For details about this interface, see the Oracle CEP Javadocs.

Sometimes, however, you might not be able, or want, to include the third-party caching implemenation in the same bundle as the Oracle CEP application that is using it. In this case, you must create a separate bundle whose Spring application context includes the <wlevs:caching-system> tag, with the advertise attribute mandatory:

  <wlevs:caching-system id ="caching-system-id" 
                         class="third-party-implementation-class"
                          advertise="true"/>

Alternatively, if you want to decouple the implementation bundle from the bundle that references it, or you are plugging in a caching implementation that supports multiple caching systems per Java process, you can specify a factory as a provider:

  <wlevs:caching-system id ="caching-system-id" 
                         provider="caching-provider"/>

  <factory id="factory-id" provider-name="caching-provider">
     <class>the.factory.class.name</class>
  </factory>

You must deploy this bundle alongside the application bundle so that the latter can start using it.

Declaring a Cache in the EPN Assembly File

After you have declared a caching system for an application, you can configure one or more caches using the <wlevs:cache> tag:

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
    </wlevs:cache>

The name attribute is optional; specify it only if the name of the cache in the caching system is different from its ID. The <wlevs:caching-system> child element references the already-declared caching system that contains the cache. You must specify this child element only if the caching system is ambiguous: there is more than one caching system declared (either implicitly or explicitly) or if the caching system is in a different application or bundle.

You may not export the cache as an OSGI service using the advertise attribute; only caching systems can be advertised.

The caching system is responsible for creating the cache associated with a particular name and returning a reference to the cache. The resulting cache bean implements the java.util.Map interface.

Configuring and Programming a Business POJO to Access a Cache

A business POJO, configured as a standard Spring bean in the EPN assembly file, can be injected with a cache using the standard Spring mechanism for referencing another bean. In this way the POJO can view and manipulate the cache. A cache bean implements the java.util.Map interface which is what the business POJO uses to access the injected cache.

First, the configuration of the business POJO in the EPN assembly file must be updated with a <property> child element, as shown in the following example based on the Output bean of the FX example (see Helloworld Example):

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
    </wlevs:cache>
    ...
    <bean class="com.bea.wlevs.example.helloworld.HelloWorldBean">
        <property name="map" ref="cache-id"/>
    </bean>

In the example, the ref attribute of <property> references the id value of the <wlevs:cache> tag. Oracle CEP automatically injects the cache, implemented as a java.util.Map, into the business POJO bean.

In the business POJO bean Java source, add a setMap (Map) method with the code that implements whatever you want the POJO to do with the cache:

  package com.bea.wlevs.example.helloworld;

  …

  import java.util.Map;

  public class HelloWorldBean implements EventSink {

  ...

    public void setMap (Map map) {...}

  }

Configuring and Programming an Adapter to Access a Cache

An adapter can also be injected with a cache using the standard Spring mechanism for referencing another bean. A cache bean implements the java.util.Map interface which is what the adapter uses to access the injected cache.

First, the configuration of the adapter in the EPN assembly file must be updated with a <wlevs:instance-property> child element, as shown in the following example:

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
    </wlevs:cache>
    ...
    <wlevs:adapter id="myAdapter" provider="myProvider">
         <wlevs:instance-property name="map" ref="cache-id"/>
    </wlevs:adapter>

In the example, the ref attribute of <wlevs:instance-property> references the id value of the <wlevs:cache> tag. Oracle CEP automatically injects the cache, implemented as a java.util.Map, into the adapter.

In the adapter Java source, add a setMap (Map) method with the code that implements whatever you want the adapter to do with the cache:

  package com.bea.wlevs.example;

  …

  import java.util.Map;

  public class MyAdapter implements Runnable, Adapter, EventSource, SuspendableBean  {

  ...

    public void setMap (Map map) {...}

  }

Configuring and Programming a User-Defined EPL Function to Access a Cache

In addition to standard event streams, EPL rules can also invoke the member methods of a user-defined function.

These user-defined functions are implemented as standard Java classes and are declared in the EPN assembly file using the standard Spring bean tags, as shown in the following example:

    <bean id="orderFunction" class="orderFunction-impl-class"/>

The processor in which the relevant EPL rule runs must then be injected with the user-defined function using the <wlevs:function> child element, referencing the Spring with the ref attribute:

    <wlevs:processor id= "tradeProcessor">
        <wlevs:function ref="orderFunction"/>
    </wlevs:processor>

The following EPL rule, assumed to be configured for the tradeProcessor processor, shows how to invoke the existsOrder() method of the orderFunction user-defined function:

INSERT INTO InstitutionalOrder
SELECT er.orderKey AS key, er.symbol AS symbol, er.shares as cumulativeShares
FROM ExecutionRequest er RETAIN 8 HOURS WITH UNIQUE KEY
WHERE NOT orderFunction.existsOrder(er.orderKey)

You can also configure the user-defined function to access a cache by by injecting the function with a cache using the standard Spring mechanism for referencing another bean. A cache bean implements the java.util.Map interface which is what the user-defined function uses to access the injected cache.

First, the configuration of the user-defined function in the EPN assembly file must be updated with a <wlevs:property> child element, as shown in the following example:

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
    </wlevs:cache>
    ...
    <bean id="orderFunction" class="orderFunction-impl-class">
        <wlevs:property name="cache" ref="cache-id"/>
    </bean>

In the example, the ref attribute of <wlevs:property> references the id value of the <wlevs:cache> tag. Oracle CEP automatically injects the cache, implemented as a java.util.Map, into the user-defined function.

In the user-defined function’s Java source, add a setMap (Map) method with the code that implements whatever you want the function to do with the cache:

  package com.bea.wlevs.example;

  …

  import java.util.Map;

  public class OrderFunction {

  ...

    public void setMap (Map map) {...}

  }

Configuring a Cache as a Listener

A cache can be configured as an explicit listener in the event processing network in order to receive events.

For example, to specify that a cache listens to a stream, specify the <wlevs:listener> tag with a reference to the cache as a child of the <wlevs:stream> tag as shown below:

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
    </wlevs:cache>
    ...
    <wlevs:stream id="tradeStream">
        <wlevs:listener ref="cache-id"/>
    </wlevs:stream>

As the stream sends new events to the cache, they are inserted into the cache. If a remove event (an old event that exits the output window) is sent by the stream, then the event is removed from the cache.

Specifying the Key Used to Index a Cache

When you configure a cache to be a listener, events are inserted into the cache. This section describes the variety of options available to you to specify the key used to index a cache in this instance.

The first option is to allow you to specify a property name for the key property when a cache is declared, as shown in the following example:

    <wlevs:cache id="cache-id" key-properties="key-property-name">
        <wlevs:caching-system ref="caching-system-id"/>
    </wlevs:cache>

In this case, all events that are inserted into the cache are required to have a property of this name at runtime, otherwise Oracle CEP throws an exception.

The second option is to use the metada annotation com.bea.wlevs.ede.api.Key to annotate the event property in the Java class that implements the event type.

The third option is to not specify a key at all. In this case the event object itself serves as both the key and value when the event is inserted into the cache. The event class must include a valid implementation of the equals and hashcode methods that take into account the values of the key properties.

The final option is to use the key-class attribute of the <wlevs:cache> tag to specify a composite key in which multiple properties form the key. The value of the key-class attribute must be a JavaBean whose public fields match the fields of the event class. The matching is done according to the field name. For example:

    <wlevs:cache id="cache-id" key-class="key-class-name">
        <wlevs:caching-system ref="caching-system-id"/>
    </wlevs:cache>

Configuring a Cache as an Event Source

A cache can be configured as a source of events to which another component in the event processing network listens. The listening component can be an adapter or processor, or a standard Spring bean. Any component that listens to a cache must implement the com.bea.cache.jcache.CacheListener interface. The following example shows how to configure a cache to be an event source for a Spring bean:

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
        <wlevs:listener ref="cache-listener-id" />
    </wlevs:cache>
    ...
    <bean id="cache-listener-id" class="wlevs.example.MyCacheListener"/>

In the example, the cache-listener-id Spring bean listens to events coming from the cache; the class that implements this component, wlevs.example.MyCacheListener, must implement the com.bea.jcache.CacheListener interface. You must program the wlevs.example.MyCacheListener class yourself.

Configuring a Cache Loader

A cache loader is an object that loads objects into a cache. You configure a cache loader by using the <wlevs:loader> child tag of the <wlevs:cache> tag to specify the bean that does the loading work, as shown in the following example:

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
        <wlevs:loader ref="cache-loader-id" />
    </wlevs:cache>
    ...
   <bean id="cache-loader-id" class="wlevs.example.MyCacheLoader"/>

In the example, the cache-loader-id Spring bean does the cache loading work; it is referenced by the cache using the ref attribute of the <wlevs:loader> child element.

You must program the wlevs.example.MyCacheLoader class yourself, and it must implement the com.bea.cache.jcache.CacheLoader interface. This interface includes the load() method to customize the loading of a single object into the cache; Oracle CEP calls this method when the requested object is not in the cache. The interface also includes loadAll() methods that you implement to customize the loading of the entire cache.

Configuring a Cache Store

You can configure a cache with a custom store that is responsible for writing data from the cache to a backing store, such as a table in a database. You configure a cache store by using the <wlevs:store> child tag of the <wlevs:cache> tag to specify the bean that does the actual storing work, as shown in the following example:

    <wlevs:caching-system id="caching-system-id"/>
    ...
    <wlevs:cache id="cache-id" name="alternative-cache-name">
        <wlevs:caching-system ref="caching-system-id"/>
        <wlevs:store ref="cache-store-id" />
    </wlevs:cache>
    ...
   <bean id="cache-store-id" class="wlevs.example.MyCacheStore"/>

In the example, the cache-store-id Spring bean does the work of writing the data from the cache to the backing store; it is referenced by the cache using the ref attribute of the <wlevs:store> child element.

You must program the wlevs.example.MyCacheStore class yourself, and it must implement the com.bea.cache.jcache.CacheStore interface. This interface includes the store() method that stores the data in the backing store using the passed key; Oracle CEP calls this method it inserts data into the cache. The interface also includes the storeAll() method for storing all data in the cache to a backing store.

Referencing a Cache from an EPL Statement

You can reference a cache from an EPL statement in much the same way you reference a stream; this feature enables you to enrich standard streaming data with data from a separate source. For example, the following EPL query joins trade events from a standard stream with company data from a cache:

INSERT INTO EnrichedTradeEvent
SELECT trade.symbol, trade.price, trade.numberOfShares, company.name
FROM TradeEvent trade RETAIN 8 hours, Company company
WHERE trade.symbol = company.id

In the example, both TradeEvent and Company are event types registered in the repository, but they have been configured in such a way that TradeEvents come from a standard stream of events but Company maps to a cache in the event processing network. This configuration happens outside of the EPL query, which means that the source of the data is transparent in the query itself.

When you use data from a cache in an EPL query, Oracle CEP pulls the data rather than it being pushed, as is the case with a stream. This means that, continuing with the preceding sample, the query executes only when a stream pushes a trade event to the query; the company data in the cache never causes a query to execute, it is only pulled by the query when needed.

Restrictions When Using a Cache in an EPL Statement

You must abide by these restrictions when using a cache in an EPL query:

Typical Steps To Reference a Cache in an EPL Statement

The following procedure assumes you have already configured the caching system and caches; see Typical Steps to Use Oracle CEP Caching for general details.