Notifications Subscriber Sample

Message Receiver is one of many well-defined extension points supported by the WSO2 Governance Registry. Read more on Supported Extension Points for a complete list of extension points supported by WSO2 Governance Registry.

WSO2 Governance Registry generates events for each significant registry operation. Through the Management Console and through APIs, you can create subscriptions to these events. WSO2 Governance Registry can publish notifications to external or internal endpoints. On the other hand, WSO2 Governance Registry contains an Apache Axis2 based runtime as a part of the WSO2 Carbon Framework. You can deploy web services on this runtime, which can be used as internal endpoints to receive notifications generated on the WSO2 Governance Registry. A Message Receiver is an extension mechanism of Apache Axis2 which allows service authors to define how message processing should happen within a service endpoint. Using a message receiver you can implement an endpoint without physically requiring an implementation class for your service. This makes it much easier for us to very easily develop event subscribers.

This sample explains how to:

Create a message receiver to subscribe to notifications on WSO2 Governance Registry
Forward notifications from WSO2 Governance Registry to WSO2 Business Activity Monitor.

This sample is similar to the Statistics Collector Sample and uses the same BAM Toolbox.

Once successfully deployed this sample will forward notification generated on WSO2 Governance Registry to the WSO2 Business Activity Monitor. We will be reusing the code of the Handler Sample in this example. This sample requires Apache Maven and WSO2 Business Activity Monitor. See Installation Prerequisites for links on how to install it. Also see Installing Business Activity Monitor on Windows from Binary Distribution or Installing Business Activity Monitor on Linux and Solaris from Binary Distribution.

Instructions

1. Open to BAM_HOME/ repository/conf/carbon.xml and set the port offset to 1.

<Offset>1</Offset>

2. Start the WSO2 Business Activity Monitor. See Installing Business Activity Monitor on Windows from Binary Distribution or Installing Business Activity Monitor on Linux and Solaris from Binary Distribution.

3. Deploy the KPI_Registry_Activity.tbox BAM Toolbox to WSO2 Business Activity Monitor.

A BAM Toolbox is an installable archive which contains stream definitions, dashboard components and analytics for WSO2 Business Activity Monitor. The KPI_Registry_Activity.tbox is a pre-built BAM Toolbox based on the KPI Monitoring Sample example of WSO2 Business Activity Monitor, which is specifically designed for this sample. Read the KPI Monitoring Sample example of WSO2 Business Activity Monitor to learn how to make changes to this BAM Toolbox or create your own.

Select the ToolBox From File System option and fill in the URL of the KPI_Registry_Activity.tbox Toolbox. Then click on the Installbutton.

If the BAM Toolbox fails to install from the URL download the KPI_Registry_Activity.tbox Toolbox to your local file system and upload it.

4. Navigate to GREG_HOME/ samples/handler/src to find the source code of the Handler Sample.

5. Add the following dependencies to your POM file:

Project Object Model (POM) is an XML representation of a Maven project held in a file named pom.xml. You should find POM file by the name pom.xml inside GREG_HOME/ samples/handler/src.

<dependency>
    <groupId>org.apache.axis2.wso2</groupId>
    <artifactId>axis2</artifactId>
</dependency>
<dependency>
    <groupId>org.wso2.carbon</groupId>
    <artifactId>org.wso2.carbon.databridge.agent.thrift</artifactId>
    <version>4.0.1</version>
</dependency>
<dependency>
    <groupId>org.wso2.carbon</groupId>
    <artifactId>org.wso2.carbon.databridge.commons</artifactId>
    <version>4.0.0</version>
</dependency>

6. Comment-out the following in your POM file:

<!--<exclusion>
    <groupId>org.wso2.carbon</groupId>
    <artifactId>org.wso2.carbon.context</artifactId>
</exclusion>-->

<!--Fragment-Host>org.wso2.carbon.registry.core</Fragment-Host-->

The Fragment-Host Bundle Manifest Header declares this sample bundle to be a Fragment of the Registry Kernel. This would mean that the sample handler bundle will become an extension to the Registry Kernel. However, bundles containing services.xml files would not be deployed until the Apache Axis2 runtime has been initialized. Due to OSGi bundle start-up order, it is required that the Registry Kernel starts up before the Apache Axis2 runtime.

Since the Fragment now also being an extension to the Registry Kernel, the combination of this bundle and the Registry Kernel (plus any other fragments) should start up before the Apache Axis2 runtime. But, due to the bundle not getting deployed until the Apache Axis2 runtime has been initialized, it will create a deadlock situation. Due to this reason, we have to get rid of the Fragment-Host header.

7. Add the following content to a file named services.xml in GREG_HOME/samples/handler/src/resources/META-INF:

<serviceGroup>
    <service name="Subscriber" scope="transportsession">
        <transports>
            <transport>http</transport>
        </transports>
        <operation name="receive">
            <actionMapping>http://ws.apache.org/ws/2007/05/eventing-extended/Publish</actionMapping>
            <messageReceiver
                    mep="http://www.w3.org/2004/08/wsdl/in-only"
                    class="org.wso2.carbon.registry.samples.receiver.SampleMessageReceiver" />
        </operation>
    </service>
</serviceGroup>

8. Add a new Java Class named SampleMessageReceiver at

GREG_HOME/samples/handler/src/src/main/java/org/wso2/carbon/registry/samples/receiver/SampleMessageReceiver.java with the following source:

package org.wso2.carbon.registry.samples.receiver;
 
import org.apache.axiom.om.OMElement;
import org.apache.axiom.om.xpath.AXIOMXPath;
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.receivers.AbstractMessageReceiver;
import org.wso2.carbon.databridge.agent.thrift.Agent;
import org.wso2.carbon.databridge.agent.thrift.DataPublisher;
import org.wso2.carbon.databridge.agent.thrift.conf.AgentConfiguration;
import org.wso2.carbon.databridge.commons.Event;
import org.wso2.carbon.databridge.commons.exception.NoStreamDefinitionExistException;
import org.wso2.carbon.registry.core.utils.RegistryUtils;
import org.wso2.carbon.utils.NetworkUtils;
 
import java.util.ArrayList;
 
public class SampleMessageReceiver extends AbstractMessageReceiver {
 
    public static final String NAMESPACE = "http://wso2.org/ns/2011/01/eventing/registry/event";
    public static final String REGISTRY_ACTIVITY_STREAM = "org.wso2.bam.registry.activity.kpi";
    public static final String VERSION = "1.0.0";
 
    protected void invokeBusinessLogic(MessageContext messageContext) throws AxisFault {
        SOAPEnvelope envelope = messageContext.getEnvelope();
        try {
            // Find Username and Operation
            AXIOMXPath xPath = new AXIOMXPath("//ns:RegistryOperation");
            xPath.addNamespace("ns", NAMESPACE);
            String operation = ((OMElement)((ArrayList)xPath.evaluate(envelope)).get(0)).getText();
            xPath = new AXIOMXPath("//ns:Username");
            xPath.addNamespace("ns", NAMESPACE);
            String username = ((OMElement)((ArrayList)xPath.evaluate(envelope)).get(0)).getText();
 
            // Create Data Publisher
            RegistryUtils.setTrustStoreSystemProperties();
            DataPublisher dataPublisher = new DataPublisher(
                    "tcp://" + NetworkUtils.getLocalHostname() + ":7612", "admin", "admin",
                    new Agent(new AgentConfiguration()));
 
            // Find Data Stream
            String streamId;
            try {
                streamId = dataPublisher.findStream(REGISTRY_ACTIVITY_STREAM, VERSION);
            } catch (NoStreamDefinitionExistException ignored) {
                streamId = dataPublisher.defineStream("{" +
                        "  'name':'" + REGISTRY_ACTIVITY_STREAM + "'," +
                        "  'version':'" + VERSION + "'," +
                        "  'nickName': 'Registry_Activity'," +
                        "  'description': 'Registry Activities'," +
                        "  'metaData':[" +
                        "          {'name':'clientType','type':'STRING'}" +
                        "  ]," +
                        "  'payloadData':[" +
                        "          {'name':'operation','type':'STRING'}," +
                        "          {'name':'user','type':'STRING'}" +
                        "  ]" +
                        "}");
            }
 
            if (!streamId.isEmpty()) {
                // Publish Event to Stream
                dataPublisher.publish(new Event(
                        streamId, System.currentTimeMillis(),
                        new Object[]{"external"}, null, new Object[]{
                        operation, username}));
                dataPublisher.stop();
                System.out.println("Successfully Published Event");
            }
 
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

9. Compile the source code by running the following command inside GREG_HOME/ samples/handler/src:

mvn clean install

The command mvn clean install will trigger an Apache Maven Build in your command line. This requires you having installed Apache Maven. See Installation Prerequisites for links on how to install it.

A successful run of Apache Maven will generate a report similar to the following:

10. Copy the GREG_HOME/ samples/handler/src/target/ org.wso2.carbon.registry.samples.handler-4.5.0.jar into GREG_HOME/repository/components/dropins.

11. Start the server and observe the command prompt. See Running the Product for more information.

You should also observe a log similar to the following explaining that your Message Receiver was successfully deployed.

[2012-09-12 01:57:38,059]  INFO {org.wso2.carbon.core.deployment.DeploymentInterceptor} -  Deploying Axis2 service: Subscriber {super-tenant}

12. After the server has started, log into the Management Console and add a notification to the root collection with the following settings:

Event - Update
Notification - SOAP
Endpoint - http://localhost:9763/services/Subscriber
Hierarchical Subscription Method - Collection, Children and Grand Children

13. Now perform various operations on the registry such such Adding/Updating Resources, Setting Properties etc. This should generate notifications which will then be forwarded for BAM. You should see lines similar to the following printed on your command prompt, which indicates that you events were successfully generated. For best results, create multiple user accounts and log in using different credentials while you perform operations.

Successfully Published Event

14. Navigate to the Gadget Portal of WSO2 Business Activity Monitor to see the statistics corresponding to your operations.

A Message Receiver must implement the org.apache.axis2.engine.MessageReceiver interface. Read more about Message Receivers to get a better understanding of their uses.