Application Development Guide

Using and Creating HTTP Publish-Subscribe Adapters

This section contains information on the following subjects:

Overview of HTTP Publish-Subscribe Functionality in Oracle CEP

An HTTP Publish-Subscribe Server (for simplicity, also called pub-sub server in this document) is a mechanism whereby Web clients, such as browser-based clients, subscribe to channels, receive messages as they become available, and publish messages to these channels, all using asynchronous messages over HTTP. A channel is similar to a JMS topic. For additional general information about HTTP Publish-Subscribe Servers, see Using the HTTP Publish-Subscribe Server.

Every instance of Oracle CEP includes a pub-sub server that programmers can use to implement HTTP publish-subscribe functionality in their applications. The pub-sub server is configured in the config.xml file along with other server services such as Jetty and JDBC datasources. The pub-sub server is based on the Bayeux protocol proposed by the cometd project. The Bayeux protocol defines a contract between the client and the server for communicating with asynchronous messages over HTTP.

In Oracle CEP, programmers access HTTP publish-subscribe functionality by using three built-in HTTP publish-subscribe adapters (called pub-sub adapters for short): two for local and remote publishing and one for subscribing to a channel. Oracle CEP also provides a pub-sub API for programmers to create their own custom pub-sub adapters for publishing and subscribing to a channel, if the built-in pub-sub adapters are not adequate. For example, programmers might want to filter incoming messages from a subscribed channel, dynamically create or destroy local channels, and so on. The built-in pub-sub adapters do not provide this functionality, which is why programmers must implement their own custom pub-sub adapters in this case.

The three built-in pub-sub adapters work like any other adapter: they are stages in the event processing network, they are defined in the EPN assembly file, and they are configured with the standard component configuration files. Typical configuration options include specifying channels, specifying the local or remote pub-sub server, and user authentication.

The following sections provide additional conceptual information about these built-in pub-sub adapters:

The pub-sub server can communicate with any client that can understand the Bayeux protocol. Programmers developer their Web clients using one of the following frameworks:

Converting Between JSON Messages and Event Types

Oracle CEP can automatically convert incoming JSON messages to event types, and vice versa in the outbound case. However, if you want to customize the way a JSON message (either inbound via a HTTP pub-sub adapter for subscribing or outbound via an HTTP pub-sub adapter for publishing) is converted to an event type, or vice versa, you must create your own converter bean. See Creating a Custom Converter Between the HTTP Pub-Sub Messages and Event Types for details.

If you do not provide your own converter class, and instead let Oracle CEP take care of the conversion between messages and event types, the following is true:

Overview of the Built-In Pub-Sub Adapter for Local Publishing

The following graphic shows how the built-in pub-sub adapter for local publishing fits into a simple event processing network. The arbitrary adapter and processor are not required, they are just an example of possible components in your application in addition to the pub-sub adapter.

In the preceding graphic:

Overview of the Built-In Pub-Sub Adapter for Remote Publishing

The following graphic shows how the built-in pub-sub adapter for remote publishing fits into a simple event processing network.

In the preceding graphic:

Overview of the Built-In Pub-Sub Adapter for Subscribing

The following graphic shows how the built-in pub-sub adapter for subscribing fits into a simple event processing network. The arbitrary processor and business POJO are not required, they are just an example of possible components in your application in addition to the pub-sub adapter.

In the preceding graphic:

Using the Built-In HTTP Pub-Sub Adapters in an Application: Typical Steps

The following procedure describes typical steps for using the built-in pub-sub adapters in your Oracle CEP application.

Creating a Custom Converter Between the HTTP Pub-Sub Messages and Event Types

If you want to customize the way a message (either inbound via a HTTP pub-sub adapter for subscribing or outbound via an HTTP pub-sub adapter for publishing) is converted to an event type, or vice versa, you must create your own converter bean.

The custom converter bean for an inbound HTTP pub-sub message must implement the com.bea.wlevs.adapters.httppubsub.api.InboundMessageConverter interface. This interface has a single method:

public List convert(JSONObject message) throws Exception;

The message parameter corresponds to the incoming HTTP pub-sub message and the return value is a List of events that will be passed on to the next stage of the event processing network. The incoming message is assumed to be the JSON format.

The custom converter bean for an outbound HTTP pub-sub message must implement the com.bea.wlevs.adapters.httppubsub.api.OutboundMessageConverter interface. This interface has a single method:

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

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

See the Javadoc for a full description of these APIs.

The following example shows the Java source of a custom converter bean that implements both InboundMessageConverter and OutboundMessageConvert; this bean can be used for both inbound and outbound HTTP pub-sub adapters:

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

Configuring the HTTP Pub-Sub Adapters

You configure the built-in pub-sub adapters in their respective configuration files, similar to how you configure other components in the event processing network, such as processors or streams. For general information about these configuration files, see Component Configuration Files.

The following configuration file shows a complete example of configuring each of the three built-in pub-sub adapters; the procedure following the example uses the example to show how to create your own file:

<?xml version="1.0" encoding="UTF-8"?>

<n1:config 
  xsi:schemaLocation="http://www.bea.com/ns/wlevs/config/application wlevs_application_config.xsd"
  xmlns:n1="http://www.bea.com/ns/wlevs/config/application" 
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

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

    <http-pub-sub-adapter>
        <name>localPublisher</name>
        <server-context-path>/pubsub</server-context-path>
        <channel>/channel2</channel>
    </http-pub-sub-adapter>

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

</n1:config>

The following procedure describes the main steps to configure the built-in pub-sub adapters for your application. For simplicity, it is assumed in the procedure that you are going to configure all components of an application 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 configuration of the built-in pub-sub adapters.

Updating the EPN Assembly File

For each pub-sub adapter in your event processing network, you must add a corresponding <wlevs:adapter> tag to the EPN assembly file that describes the network; use the provider attribute to specify whether the adapter is for publishing or subscribing. Follow these guidelines:

As with any other stage in the EPN, add listeners to the <wlevs:adapter> tag to integrate the pub-sub adapter into the event processing network. Typically, a pub-sub adapter for subscribing is the first stage in an EPN (because it receives messages) and a pub-sub adapter for publishing would be in a later stage (because it sends messages). However, the requirements of your own Oracle CEP application define where in the network the pub-sub adapters fit in.

Also be sure that the event types used by the pub-sub adapters have been registered in the event type repository using the <wlevs:event-type-repository> tag.

The following sample EPN file shows an event processing network with two built-in pub-sub adapters for publishing both local and remote publishing); see the text after the example for an explanation:

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:osgi="http://www.springframework.org/schema/osgi"
       xmlns:wlevs="http://www.bea.com/ns/wlevs/spring"
       xsi:schemaLocation="
  http://www.springframework.org/schema/beans
  http://www.springframework.org/schema/beans/spring-beans.xsd
  http://www.springframework.org/schema/osgi
  http://www.springframework.org/schema/osgi/spring-osgi.xsd
  http://www.bea.com/ns/wlevs/spring
  http://www.bea.com/ns/wlevs/spring/spring-wlevs.xsd">

    <wlevs:event-type-repository>
        <wlevs:event-type type-name="com.mycompany.httppubsub.PubsubEvent">
            <wlevs:class>com.mycompany.httppubsub.PubsubEvent</wlevs:class>
        </wlevs:event-type>
    </wlevs:event-type-repository>

    <wlevs:adapter id="receiveFromFeed"
                   class="com.mycompany.httppubsub.ReceiveFromFeed">
    </wlevs:adapter>

    <wlevs:processor id="pubsubProcessor" />

    <wlevs:adapter id="remotePublisher" provider="httppub"/>

    <wlevs:adapter id="localPublisher" provider="httppub"/>

    <wlevs:stream id="feed2processor">
        <wlevs:source ref="receiveFromFeed"/>
        <wlevs:listener ref="pubsubProcessor"/>
    </wlevs:stream>

    <wlevs:stream id="pubsubStream">
        <wlevs:listener ref="remotePublisher"/>
        <wlevs:listener ref="localPublisher"/>
        <wlevs:source ref="pubsubProcessor"/>
    </wlevs:stream>

</beans>

In the preceding example:

The following sample EPN file shows an event processing network with one built-in pub-sub adapter for subscribing; see the text after the example for an explanation:

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:osgi="http://www.springframework.org/schema/osgi"
       xmlns:wlevs="http://www.bea.com/ns/wlevs/spring"
       xsi:schemaLocation="
  http://www.springframework.org/schema/beans
  http://www.springframework.org/schema/beans/spring-beans.xsd
  http://www.springframework.org/schema/osgi
  http://www.springframework.org/schema/osgi/spring-osgi.xsd
  http://www.bea.com/ns/wlevs/spring
  http://www.bea.com/ns/wlevs/spring/spring-wlevs.xsd">

    <wlevs:event-type-repository>
        <wlevs:event-type type-name="com.mycompany.httppubsub.PubsubEvent">
            <wlevs:class>com.mycompany.httppubsub.PubsubEvent</wlevs:class>
        </wlevs:event-type>
    </wlevs:event-type-repository>

    <wlevs:adapter id="remoteSubscriber" provider="httpsub">
        <wlevs:listener ref="myEventBean"/>
    </wlevs:adapter>

    <bean id="myEventBean"
          class="com.mycompany.httppubsub.MyEventBean">
    </bean>

    <wlevs:stream id="pubsubStream" manageable="true">
        <wlevs:listener>
            <bean id="mySink"
                  class="com.mycompany.httppubsub.MySink"/>
        </wlevs:listener>
        <wlevs:source ref="myEventBean"/>
    </wlevs:stream>

</beans>

In the preceding example: