WS-Eventing with SAVAN

WS-Eventing defines a protocol for one Web service (called a "subscriber") to register interest (called a "subscription") with another Web service (called an "event source") in receiving messages about events (called "notifications" or "event messages").

The subscriber may manage the subscription by interacting with a Web service (called the "subscription manager") designated by the event source.

Savan is the WS-Eventing implementation for Apache Axis2.

This post guides you through WS-Eventing with an example in Savan/Axis2.

-First download Axis2 1.4.1 and extract it to a local folder [AXIS2_HOME].

-Start Axis2 simple server with [AXIS2_HOME]\bin\axis2server.bat and make sure http://localhost:8080/axis2/services/ lists all avaialble services.

The sample application includes three components.

1. Event source[axis2-savan-event-source] - a stock quote service publishes stock prices iteratively

2. Event sink[axis2-savan-event-sink] - a stock quote consumer subsribes to the stock quote service to receive events

3. [axis2-savan-client] - initiates subscription for the stock quote consumer

You can download Eclipse projects for above three components from here and import those into your Eclipse workspace, say [SAVAN].

Copy all the jar files from [AXIS2_HOME]\lib to [SAVAN]\lib.

Copy [SAVAN]\lib\savan-SNAPSHOT.mar to [AXIS2_HOME]\repository\modules.

Copy [SAVAN]\lib\savan-core-SNAPSHOT.jar to [AXIS2_HOME]\lib.

Share the folder [SAVAN]\lib and map it to the network drive "W".

All set now, refresh your workspace and it should build with no compile errors.

Lets start with, Event source[axis2-savan-event-source].
// org.apache.ws.axis2/StockQuoteService
public OMElement getNextPublicationData() {

OMFactory factory = null;
OMNamespace namespace = null;
OMElement stockElement = null;
OMElement data = null;

// <ns1:publish xmlns:ns1="http://stocktrader.consumer">
//       <ns1:Stock xmlns:ns1="http://stocktrader.consumer">12345
// </ns1:publish>

factory = OMAbstractFactory.getOMFactory();
namespace = factory.createOMNamespace(NAMESPACE, "ns1");
stockElement = factory.createOMElement(STOCK, namespace);

int value = r.nextInt();
stockElement.setText(Integer.toString(value));

data = factory.createOMElement("publish", namespace);
data.addChild(stockElement);

return data;
}
The above code will generate the data to publish - where the subscribers can recieve. At the same time pay your attention to line:20.

Here the subscribed services should have a method called 'publish' [that is, they should match the name used here].
store = CommonUtil.getSubscriberStore(serviceContext.getAxisService());

if (store != null)  {
data = getNextPublicationData();
publicationClient = new PublicationClient(serviceContext.getConfigurationContext());
publicationClient.sendPublication(data, serviceContext.getAxisService(),null);
}
Line:1 will get the SubcriberStore for the corresponding service, in this case, for the StockQuoteService. Savan, module will create the SubcriberStore for the given service upon the first subscription request.

Following, shows the services.xml for StockQuoteService, where you need to engage "addressing" and "savan" modules.
<!-- src/META-INF/services.xml -->
<service name="StockQuoteService" scope="application">

<module ref="savan" />
<module ref="addressing" />

<parameter name="ServiceClass" locked="false">
org.apache.ws.axis2.StockQuoteService
</parameter>

</service>
Now lets deploy the StockQuoteService.

Right click the project axis2-savan-event-source, select 'Export' --> Select 'General/Archive File'. --> Select settings as in the image below.

Lets, move to the Event sink[axis2-savan-event-sink] or the StockQuoteConsumer.

This service has a single method, which simply prints the notifications received from the StockQuoteService.
public void publish(OMElement param) throws Exception {
System.out.println(param);
}
services.xml for the StockQuoteConsumer service will look like, following.
<service name="StockQuoteConsumer">

<parameter name="ServiceClass" locked="false">org.apache.ws.axis2.StockQuoteConsumerImpl</parameter>

<operation name="publish" mep="http://www.w3.org/2004/08/wsdl/in-only">
<messageReceiver class="org.apache.axis2.receivers.RawXMLINOnlyMessageReceiver" />
</operation>

</service>
Note down here that we do not engage 'savan' module with the StockQuoteConsumer.

Now deploy the StockQuoteConsumer service in the same way as we did in StockQuoteService.

Let's focus on the Client code.
final static String SUBSCRIBER_ID = "wso2-svan-subscriber";
final static String EVENT_SINK_EPR = "http://localhost:8080/axis2/services/StockQuoteConsumer";
final static String EVENT_SOURCE_EPR = "http://localhost:8080/axis2/services/StockQuoteService";

ServiceClient client = null;
Options options = null;
ConfigurationContext confContext = null;
EventingClientBean bean = null;
EventingClient eventingClient = null;


confContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem("repo","repo/conf/client.axis2.xml");
client = new ServiceClient(confContext, null);
options = new Options();
options.setTo(new EndpointReference(EVENT_SOURCE_EPR));
client.setOptions(options);
client.engageModule("addressing");  

bean = new EventingClientBean();

// indicates where to send notifications
bean.setDeliveryEPR(new EndpointReference(EVENT_SINK_EPR));

// where to send a SubscriptionEnd message 
// bean.setEndToEPR(new EndpointReference(""));

// indicates the event sink would prefer to have the subscription expire
// bean.setExpirationTime(new Date(""));
// bean.setExpirationDuration(new Duration("10"));

// bean.setFilter(filter)'
// bean.setFilterDialect(filterDialect);

eventingClient = new EventingClient(client);
eventingClient.subscribe(bean, SUBSCRIBER_ID);
All set, start the Axis2 simple server first and then simply run the Client.

You can see the out put on Axis2 simple server console.

Labels: ,