Without restart: Enabling WSO2 ESB as a JMS Consumer of WSO2 MB

WSO2 ESB 4.8.1 & WSO2 MB 2.2.0 documentations have information on how to configure WSO2 ESB as a JMS Consumer of WSO2 MB queues and topics. But they do not point out a way to do this without restarting ESB server.

In this blog post we'll solve this issue.

With this method we will be able to create new queues in WSO2 MB and consume them from WSO2 ESB without restarting it.

Configure the WSO2 Message Broker

• Offset the port of WSO2 MB to '1'  
• Copy andes-client-*.jar and geronimo-jms_1.1_spec-*.jar from $MB_HOME/client-lib to $ESB_HOME/repository/components/lib 
• Start the WSO2 MB

Configure the WSO2 Enterprise Service Bus

• Edit the $ESB_HOME/repository/conf/jndi.properties file (comment or delete any existing configuration)

connectionfactory.QueueConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'
connectionfactory.TopicConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'

• Edit the $ESB_HOME/repository/conf/axis2.xml file and uncomment the JMS Sender and JMS Listener configuration for WSO2 Message Broker 
• Start the WSO2 ESB 

Create Proxy Service

The Proxy Service name will become the queue name in WSO2 MB. If you already have a queue in MB and if you want to listen to that queue, then set that queue name as the proxy service name. Here I'm using 'JMSConsumerProxy' as the queue name and the proxy service name.

<?xml version="1.0" encoding="UTF-8"?> 
<proxy xmlns="http://ws.apache.org/ns/synapse" 
       name="JMSConsumerProxy" 
       transports="jms" 
       statistics="disable" 
       trace="disable" 
       startOnLoad="true"> 
   <target> 
      <inSequence> 
         <property name="Action" 
                   value="urn:placeOrder" 
                   scope="default" 
                   type="STRING"/> 
         <log level="full"/> 
         <send> 
            <endpoint> 
               <address uri="http://localhost:9000/services/SimpleStockQuoteService"/> 
            </endpoint> 
         </send> 
      </inSequence> 
      <outSequence> 
         <drop/> 
      </outSequence> 
   </target> 
   <description/> 
</proxy> 

Testing the scenario

• Inside $ESB_HOME/samples/axis2Server/src/SimpleStockQuoteService run ant 
• Now start the Axis2 Server inside $ESB_HOME/samples/axis2Server (run the relevant command line script) 
• Log into the WSO2 Message Broker Management Console and navigate to Browse Queues 
• Find a Queue by the name JMSConsumerProxy 
• Publish 1 message to the JMSConsumerProxy with payload (this has to be done in the Message Broker Management Console) 

<ser:placeOrder xmlns:ser="http://services.samples" xmlns:xsd="http://services.samples/xsd"> 
    <ser:order> 
        <xsd:quantity>4</xsd:quantity> 
    </ser:order> 
</ser:placeOrder>

• Observe the output on the Axis2 Server and WSO2 ESB console.

Hope this helped you :) 

Monitoring ActiveMQ Queues via JMX MBeans with WSO2 Complex Event Processor (WSO2 CEP)

WSO2 CEP and WSO2 Business Activity Monitor have support for JMX MBeans monitoring. You can find the docs for JMX MBeans monitoring for WSO2 BAM from here.

I'm writing this blog to illustrate how WSO2 CEP can be used to monitor the JMX MBeans, and I'm using ActiveMQ JMS broker as an example to demonstrate its capabilities.
Here I have used apache-activemq-5.4.3

First we need to enable ActiveMQ JMS broker to support JMX remotely.

To do so, you have to enable JMX in the broker, in the <activemq_home>/conf/activemq.xml:

<broker xmlns="http://activemq.org/config/1.0" brokerName="localhost" useJmx="true">

 and 

<managementContext>
    <managementContext connectorPort="1099" />
</managementContext> 

Now you have to enable JMX in the JVM parameters in <activemq_home>/bin/activemq: 

Locate the line 

ACTIVEMQ_OPTS="$ACTIVEMQ_OPTS $SUNJMX $SSL_OPTS" 

and change it to: 

ACTIVEMQ_OPTS="$ACTIVEMQ_OPTS $SUNJMX $SSL_OPTS -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false" 

To test whether its correctly working run "jconsole"

and connect to the broker via: service:jmx:rmi:///jndi/rmi://your_host:1099/jmxrmi

Now we need to Enable to CEP to receive JMX Events

To do so first we need to install JMX Agent to CEP 

Add 4.0.7 P2 repository form the URL  http://dist.wso2.org/p2/carbon/releases/4.0.7

as described at http://docs.wso2.org/wiki/display/CEP210/Managing+the+Feature+Repository

Next install the "BAM JMX Agent Aggregate" 

Use http://docs.wso2.org/wiki/display/CEP210/Installing+Features for guidence

Restart CEP.

Now we need to configure JMX.

To do so, go to Configure -> JMX Agent 

Click on "Add Profile"  and configure he profile as below

Here use the Data Receiver user name & password as "admin" & "admin" 

JMX Server URL as  service:jmx:rmi:///jndi/rmi://your_host:1099/jmxrmi  

and JMX Server user name & password as "admin" & "activemq" 

Then  click on the "Load Means" button to load the available MBeans, 

Note : ActiveMQ need to be up an running in-order to load its MBeans

Next when you click on the appropriate MBean a dropdown will appear showing its attributes. 

You have to select your relevant attributes one by one from the drop down. 

You also need to change the Alias to a much readable strings, 

For this Example I have used:

BrokerName, TotalEnqueueCount, TotalDequeueCount, TotalMessageCount, MemoryPercentUsage

Now click "Save" to save & deploy the JMX Agent.

When we add the above configuration JMX agent will automatically create the following WSO2 Event Stream Definition

{
    "streamId": "org.wso2.bam.jmx.agent.activemq:1.0.0",
    "name": "org.wso2.bam.jmx.agent.activemq",
    "version": "1.0.0",
    "nickName": "JMX Dump",
    "description": "JMX monitoring data",
    "metaData": [{
        "name": "clientType",
        "type": "STRING"
    }, {
        "name": "host",
        "type": "STRING"
    }],
    "payloadData": [{
        "name": "BrokerName",
        "type": "STRING"
    }, {
        "name": "TotalEnqueueCount",
        "type": "LONG"
    }, {
        "name": "TotalDequeueCount",
        "type": "LONG"
    }, {
        "name": "TotalMessageCount",
        "type": "LONG"
    }, {
        "name": "MemoryPercentUsage",
        "type": "INT"
    }]
}

Next step is creating a CEP Bucket to use JMX event. 

To do this:

• Shutdown CEP 
• Run "ant deploy-broker-manager" from  <cep_home>/samples/cep-samples
• Copy paste activemq-all-xxx.jar from the <activemq_home> directory to <cep_home>/samples/lib directory. 
• Copy paste activemq-core-xxx.jar and geronimo-j2ee-management_1.1_spec-1.0.1.jar from <activemq_home>/lib to <cep_home>/repository/components/lib directory.
• Copy the queue-monitor.xml to <cep_home>/repository/deployment/server/cepbuckets
• In a new terminal run "ant jmsSubscriber -Dtopic=QueueInfo" from <cep_home>/samples/cep-samples to receive the outputs of the queue-monitor bucket.
• Start CEP 

Now try running the Queue example which is in the <amq_home>/example.

In a new terminal run "ant consumer" to run the sample consumer 

In a new terminal run "ant producer" to send some sample messages to the ActiveMQ Queue. 

you will be able to view some appropriate outputs in CEP QueueInfo jmsSubscriber

Using Apache Qpid with WSO2 CEP

Though WSO2 CEP 2.x.x can work with Qpid, it does not have samples that demonstrates this, where  CEP only have samples for ActiveMQ.

Here I'm providing steps to configure, and run some samples using Apache Qpid.

First step is downloading and running Apache Qpid.
You can download Qpid from the project web site. Note, I have used Qpid java 0.18 for these samples, you can download that from here.

Next, unzip and start the broker.

Now to configure WSO2 CEP with Qpid JMS broker, follow the steps provided at Configuring JMS-Qpid Broker.

Test publishing and subscribing to JMS Map messages

Edit the jms-twitter-stockquote-analyser.xml provided at /samples/cep-samples/conf/buckets directory by changing the input and output broker names as, 

<bucket name="TwitterAndStockQuoteAnalyzer" xmlns="http://wso2.org/carbon/cep">
     ...
     <input topic="AllStockQuotes" brokerName="qpidJmsBroker">
     ...
     <input topic="TwitterFeed" brokerName="qpidJmsBroker">
     ...
     ...
     <output topic="PredictedStockQuotes" brokerName="qpidJmsBroker">
     ...
</bucket>

Here we are using the brokerName as "qpidJmsBroker", because we have define it so, when Configuring JMS-Qpid Broker.

Next copy past this file to /repository/deployment/server/cepbuckets directory. 

To test the broker download the client code from here, and unzip cep-210-jms-client-qpid.zip 

To subscribe to the PredictedStockQuotes topic, in the terminal go to cep-210-jms-client-qpid directory and run:

ant mapPredictedStockQuotesSubscriber

To publish to the AllStockQuotes topic, in another terminal from the same directory, run:

ant mapAllStockQuotesPublisher

To publish to the AllStockQuotes topic, in another terminal from the same directory, run:

ant mapTwitterFeedPublisher

Then you will be able to observe the outputs. Here the output will be similar to the output of Twitter and StockQuote Analyzer sample.

Test publishing and subscribing to JMS XML messages

Edit the xml-stockquote-anlyzer.xml provided at /samples/cep-samples/conf/buckets directory by changing the input and output broker names as, 

<cep:bucket name="XMLStockQuoteAnalyzer" xmlns="http://wso2.org/carbon/cep">
     ...
     <cep:input topic="AllStockQuotes" brokerName="qpidJmsBroker">
     ...
     ...
     <cep:output topic="FastMovingStockQuotes" brokerName="qpidJmsBroker">
     ...
</cep:bucket>

Here we are using the brokerName as "qpidJmsBroker", because we have define it so, when Configuring JMS-Qpid Broker.

Next copy past this file to /repository/deployment/server/cepbuckets directory, and make sure to delete other bucket configurations in order to avoid conflicts.

If you don't have the client code (used for the previous sample), download the client code from here, and unzip cep-210-jms-client-qpid.zip, 

To subscribe to the FastMovingStockQuotes topic, in the terminal go to cep-210-jms-client-qpid directory and run:

ant xmlFastMovingStockQuotesSubscriber

To publish to the AllStockQuotes topic, in another terminal from the same directory, run:

ant xmlAllStockQuotesPublisher

You will be able to observe the outputs. Here the output will be similar to the output of Stock Quote Analyzer sample.