Configure WSO2 ESB with jms, accept message to ESB and persist to database

Required
WSO2 ESB 4.0.3
Activemq-5.6.0

Steps
Download ActiveMQ from here http://activemq.apache.org/
Download WSO2 ESB from here http://wso2.com/products/enterprise-service-bus/
Download Mysql and connector jars from http://dev.mysql.com/downloads/
Setup ActiveMQ
Extract ActiveMQ to /opt folder
/opt/apache-activemq-5.6.0
run activeMQ using type command /opt/apache-activemq-5.6.0/bin/activemq

Setup WSO2 ESB
First we need to copy following activeMQ jars available in /opt/apache-activemq-5.6.0/lib into repository/lib folder of ESB distribution home. From here onwards i will refer that place as ESB_HOME.

Copy these 2 jars
geronimo-j2ee-management_1.1_spec-1.0.1.jar
activemq-core-5.6.0.jar

Since we are using mysql database download and copy mysql jar as well
mysql-connector-java-5.1.12-bin.jar

Now we need to configure jms transport from ESB. open ESB_HOME/repository/conf/axis2.xml file and Enable jms transport receiver by uncommenting following section

<transportReceiver name="jms" class="org.apache.axis2.transport.jms.JMSListener">
<parameter name="myTopicConnectionFactory" locked="false">
<parameter name="java.naming.factory.initial"
locked="false">org.apache.activemq.jndi.ActiveMQInitialContextFactory</parameter>
<parameter name="java.naming.provider.url" locked="false">tcp://localhost:61616</
parameter>
<parameter name="transport.jms.ConnectionFactoryJNDIName"
locked="false">TopicConnectionFactory</parameter>
<parameter name="transport.jms.ConnectionFactoryType" locked="false">topic</
parameter>
</parameter>
<parameter name="myQueueConnectionFactory" locked="false">
<parameter name="java.naming.factory.initial"
locked="false">org.apache.activemq.jndi.ActiveMQInitialContextFactory</parameter>
<parameter name="java.naming.provider.url" locked="false">tcp://localhost:61616</
parameter>
<parameter name="transport.jms.ConnectionFactoryJNDIName"
locked="false">QueueConnectionFactory</parameter>
<parameter name="transport.jms.ConnectionFactoryType"
locked="false">queue</parameter>
</parameter>
<parameter name="default" locked="false">
<parameter name="java.naming.factory.initial"
locked="false">org.apache.activemq.jndi.ActiveMQInitialContextFactory</parameter>
<parameter name="java.naming.provider.url" locked="false">tcp://localhost:61616</
parameter>
<parameter name="transport.jms.ConnectionFactoryJNDIName"
locked="false">QueueConnectionFactory</parameter>
<parameter name="transport.jms.ConnectionFactoryType"
locked="false">queue</parameter>
</parameter>
</transportReceiver>

Enable jms transport sender by uncomment following configuration

<transportSender name="jms" class="org.apache.axis2.transport.jms.JMSSender"/>

So now we have enabled jms transport for WSO2 ESB.
Now we will define new proxy service to consume messages through jms transport. Create new proxy service as follows.

 

 

 

                    step5: edit out sequence(we don’t have to do anything here)

Now we have completed proxy service.Next we will see how to invoke this service.

First build sample
Go to ESB_HOME/samples/axis2Server/src/SimpleStockQuoteService
run ant command > ant
Next we have to start Axis2Service for that go to ESB/samples/axis2Server.
run command > sh axis2Server.sh

So now we have proxy service with following sub items
01.Log full message
02.Extract some parameter in message body using xpath expression
xmlns:m="http://services.samples"
xmlns:m0="http://schemas.xmlsoap.org/soap/envelope/" expression="//m0:Body/m:placeOrder/
m:order/m:symbol" type="VARCHAR"
03.Persist that parameter to database using db report mediator(We can write our own mediator
using class mediator to store message or part of it in any given store)
04.Send to some back end service or drop message(in this particular use case we can drop
message since we are not going to send it to actual back end anyway i have added that as well)
Here is the synapse configuration for proxy service.

<proxy name="StockQuoteProxy" transports="jms" startOnLoad="true" trace="disable">
<target>
<endpoint>
<address uri="http://localhost:9000/services/SimpleStockQuoteService"/>
</endpoint>
<inSequence>
<log level="full"/>
<dbreport>
<connection>
<pool>
<password>root</password>
<user>root</user>
<url>jdbc:mysql://localhost/testdb</url>
<driver>com.mysql.jdbc.Driver</driver>
</pool>
</connection>
<statement>
<sql>call addmessage(?)</sql>
<parameter xmlns:m="http://services.samples" xmlns:m0="http://
schemas.xmlsoap.org/soap/envelope/" expression="//m0:Body/m:placeOrder/m:order/
m:symbol" type="VARCHAR"/>
</statement>
</dbreport>
<property name="OUT_ONLY" value="true"/>
</inSequence>
<outSequence><send/></outSequence>
</target>
<publishWSDL uri="file:repository/samples/resources/proxy/sample_proxy_1.wsdl"/>
<parameter name="transport.jms.ContentType">
<rules>
<jmsProperty>contentType</jmsProperty>
<default>application/xml</default>
</rules>
</parameter>
</proxy>

Create test_db and execute following commands to create table and stored procedure.

CREATE TABLE `testdb`.`messages` (
`message_id` int(10) NOT NULL AUTO_INCREMENT,
`message` text,
PRIMARY KEY (`message_id`)
) ENGINE=MyISAM AUTO_INCREMENT=44 DEFAULT CHARSET=latin1
CREATE DEFINER=`root`@`localhost` PROCEDURE `testdb`.`addmessage`(message_txt
VARCHAR(10))
INSERT INTO messages (message) values (message_txt)
Now we can invoke proxy service from the axis2 Client available inside wso2 ESB.
Goto ESB_HOME/samples/axis2Client and type following command.
>ant jmsclient -Djms_type=pox -Djms_dest=dynamicQueues/StockQuoteProxy -
Djms_payload=dddddd
So now you will see message log in ESB console and then persist extracted parameter to
database and message send to actual back end. If you want to drop message drop it from proxy
configuration.
ESB-log
=======
[2012-08-28 12:11:43,798] INFO - LogMediator To: http://localhost:9000/services/
SimpleStockQuoteService, WSAction: urn:placeOrder, SOAPAction: urn:placeOrder,
MessageID: urn:uuid:ed0de324-79eb-4101-9a0b-c201971910fd, Direction: request
To : http://localhost:9000/services/SimpleStockQuoteService
WSAction : urn:placeOrder
SOAPAction : urn:placeOrder
MessageID : urn:uuid:ed0de324-79eb-4101-9a0b-c201971910fd
Body : 70.25</
m0:price>200MSFT</
m0:placeOrder>
[2012-08-28 12:11:43,801] INFO - PlaceStockOrderTask placed order symbol:MSFT
quantity:200 price:70.25
[2012-08-28 12:11:45,213] INFO - LogMediator To: , MessageID: ID:sanjeewa-TECRA-
M11-47367-1346181105012-1:1:1:1:1, Direction: request, Envelope: encoding='utf-8'?>

150.33783033599516
6095
MSFTddiiActual axis2 backend log
==================
Tue Aug 28 12:11:45 PDT 2012 samples.services.SimpleStockQuoteService :: Accepted order
#76 for : 6095 stocks of MSFTddii at $ 150.33783033599516
Table entry in database
=================
If you go to database and see there will be entry for stock symbol you entered