In this post, we will refer WSO2 ESB V4.0.0 as ESB and the ESB installation directory as ESB_HOME.

Content

• Enabling 'JMS' Transport
• Configuring Qpid as the Message Store
• Sample Proxy Configuration for Qpid

Applies To :

1. Enabling 'JMS' transport in ESB

First, we need to download the ESB distribution. You can download WSO2 ESB 4.0.0 here. Extract the downloaded .zip into a directory. Go to ESB_HOME/repository/conf folder, where in the axis2 configuration file, you need to activate the 'JMS' transport sender and receiver in order to send and receive messages with Qpid as underlying JMS provider.

By default,in this(axis2.xml) file JMS transport is commented out. Do the changes as below;

1) At Transport In section;

 <!--Transport Ins --> 
 <!--Uncomment this and configure as appropriate for JMS transport support with Apache Qpid --> 
     <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.qpid.jndi.
              PropertiesFileInitialContextFactory</parameter> 
             <parameter name="java.naming.provider.url" locked="false">repository/conf/jndi.properties</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.qpid.jndi.
              PropertiesFileInitialContextFactory</parameter> 
             <parameter name="java.naming.provider.url" locked="false">repository/conf/jndi.properties</parameter> 
             <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">QueueConnectionFactory</parameter> 
             <parameter name="transport.jms.ConnectionFactoryType" locked="false">queue</parameter> 
         </parameter> 
     </transportReceiver> 

At Transport Out section;

 <!-- uncomment this and configure to use connection pools for sending messages--> 
     <transportSender name="jms" class="org.apache.axis2.transport.jms.JMSSender"/> 

In the above configuration, if you notice that, we point “jndi.properties file” as the value for “java.naming.provider.url” .

2) Add following lines to create connection factory, queue , topic etc.. at jndi.properties file(ESB_HOME/repository/conf/jndi.properties);

      connectionfactory.TopicConnectionFactory = amqp://admin:admin@carbon/carbon?brokerlist='tcp://localhost:5672' 
      connectionfactory.QueueConnectionFactory = amqp://admin:admin@carbon/carbon?brokerlist='tcp://localhost:5672' 
      queue.MyQueue = example.MyQueue 
      topic.MyTopic=example.MyTopic

Now, if you restart the ESB server, you will be able to send and receive messages from/to an JMS endpoint. You can find more detail documentation on JMS transport here[1].

2. Configuring Qpid as the Message-store

Message-store functionality is available in ESB 4.0.0 version to provide advanced enterprise messaging capability. There are two kind of implementations available;

• In-memory implementation
• Persistent JMS implementation

With in-memory implementation, we can store and froward messages efficiently and it is the fastest implementation. But, when we restart the server, messages which are stored in the queue before server goes down, will be lost. Which can be considered as a disadvantage in this kind of implementation.

To make queue(store) as durable, we can use JMS persistent implementation, which is written in a way to support common JMS providers, which follow JMS 1.1 spec.

Message Store does not need JMS transport sender/receiver enabled. But if you use "Send " mediator, you should enable JMS transport.

2.1 Sample Message Store Configuration for Qpid

Your Message Store configuration should have minimum three parameters;

      <messageStore xmlns="http://ws.apache.org/ns/synapse" class="org.wso2.carbon.message.store.persistence.jms.
       JMSMessageStore"  name="testMsgStore">
          <parameter name="java.naming.factory.initial">org.apache.qpid.jndi.PropertiesFileInitialContextFactory</parameter> 
          <parameter name="java.naming.provider.url">repository/conf/jndi.properties</parameter> 
          <parameter name="store.jms.destination">MyQueue</parameter> 
      </messageStore> 

The above configuration makes you to create a JMS based message store with Qpid, but the queue is not durable.

To make durable queue with Qpid, provide the value for “store.jms.destination” as follows;

 MyQueue;{create:always, node:{durable: True}}

Message Store parameter would be like;

 <parameter name="store.jms.destination">MyQueue;{create:always, node:{durable: True}}</parameter>

You can check whether the queue is durable or not via the jconsole.

There is a final configuration user should check.

In the ESB_HOME\repository\conf\advanced\qpid-virtualhosts.xml file, the 'carbon' virtual host points to "derby store"

<virtualhost> 
   <name>carbon</name> 
        <carbon> 
            <store> 
                <!--class>org.apache.qpid.server.store.MemoryMessageStore</class--> 
                <class>org.apache.qpid.server.store.DerbyMessageStore</class> 
                <environment-path>repository/database/qpid/derbystore</environment-path> 
            </store> 

So your connection factory url (which is provided via jndi.properties file)also should be like;

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

If you do all the above configurations, you will be able to connect to a durable Qpid queue..

3. Sample proxy configuration

This sample configuration shows, how an user can make Qpid as his JMS persistent Message store.

<proxy xmlns="http://ws.apache.org/ns/synapse" name="testJMSProxy" transports="https http" startOnLoad="true" trace="disable">
    <target>
        <inSequence>
            <property name="FORCE_SC_ACCEPTED" value="true" scope="axis2"/>
            <property name="target.endpoint" value="jmsEP" scope="default"/>
            <store messageStore="testMsgStore"/>
        </inSequence>
    </target>
</proxy>
<messageStore xmlns="http://ws.apache.org/ns/synapse" class="org.wso2.carbon.message.store.persistence.jms.JMSMessageStore"
 name="testMsgStore">
    <parameter name="java.naming.factory.initial">org.apache.qpid.jndi.PropertiesFileInitialContextFactory</parameter>
    <parameter name="java.naming.provider.url">repository/conf/jndi.properties</parameter>
    <parameter name="store.jms.destination">MyQueue;{create:always, node:{durable: True}}</parameter>
</messageStore>

<messageProcessor xmlns="http://ws.apache.org/ns/synapse" name="testMsgProcessor" messageStore="testMsgStore"
 class="org.apache.synapse.message.processors.forward.ScheduledMessageForwardingProcessor" />
<endpoint xmlns="http://ws.apache.org/ns/synapse" name="jmsEP">
    <address uri="https://localhost:9000/services/SimpleStockQuoteService"/>
</endpoint>

Author

Vijayaratha Vijayasingam, Software Engineer, WSO2 Inc

References:

[1]https://www.wso2.org/project/esb/java/4.0.0/docs/transports/transports-catalog.html

[2]https://www.wso2.org/project/esb/java/4.0.0/docs/message_store.html