Integrating HornetQ with WSO2 ESB

Introduction

HornetQ is a JMS compliant open source asynchronous messaging project from JBoss. WSO2 ESB provides a simplified configuration model to integrate any JMS compliant messaging system.  Recently I came across few discussion threads looking for a sample configuration between HorenetQ and WSO2 ESB, hence this is a short article to outline the steps. 

Sample Scenario

In this example, I am going to expose an ESB proxy service which accepts sample SOAP messages and push them to a JMS queue configured in HornetQ.  This ESB proxy will only execute an outward operation and will not return anything back to the caller.

Steps

1.       First, create a sample queue by editing $HORNET_HOME/config/stand-alone/non-clustered/hornetq-jms.xml.

NOTE : There are multiple execution modes supported by HornetQ, but I am only going to use the stand-alone configuration for this example.

<queue name="wso2">
      <entry name="/queue/mySampleQueue"/>
</queue>

 2.       Put the following 2 connection factory entries that will be required if the ESB needs to act as a JMS consumer.  However, this example will only cover the scenario of WSO2 ESB acting as a producer of which the incoming payload will simply be pushed to a queue.

<connection-factory name="QueueConnectionFactory">
      <xa>false</xa>
      <connectors>
         <connector-ref connector-name="netty"/>
      </connectors>
      <entries>
         <entry name="/QueueConnectionFactory"/>
      </entries>
</connection-factory>


<connection-factory name="TopicConnectionFactory">
      <xa>false</xa>
      <connectors>
         <connector-ref connector-name="netty"/>
      </connectors>
      <entries>
         <entry name="/TopicConnectionFactory"/>
      </entries>
</connection-factory>

 3.       Copy HoenetQ client JARs into $ESB_HOME/repository/components/lib.

NOTES :
[a.] There is a current limitation on the ESB that whenever multiple non-OSGi JARs having the same package names are copied, the process to convert them as OSGi bundle may drop these packages on subsequent JARs.  Hence I had to assemble those multiple client side JARs into a single JAR.  There are multiple ways to do that (i.e Maven JAR assembly plugin, etc.), and for the convenience I am attaching the assembled JAR for HornetQ 2.3.0.  You just have to copy this hornet-all.jar into $ESB_HOME/repository/components/lib if you decide to use this already assembled JAR.

Download the jar from here


[b.] If you opt to pack the JARs yourself, please make sure the remove the javax.jms package from this assembled JAR to avoid the carbon runtime from picking this implementation of JMS over the bundled-in distribution.

 4.       Uncomment the following line to enable JMS transportSender on axis2 core @ $ESB_HOME/repository/conf/axis2/axis2.xml

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


 5.       Let’s configure a simple proxy service as below which accept a simple SOAP message and push the SOAP envelope into the mySampleQueue. 

<proxy xmlns="http://ws.apache.org/ns/synapse" name="toJMSProxy" transports="https,http" statistics="disable" trace="disable" startOnLoad="true">
   <target>
      <inSequence>
         <property name="Accept-Encoding" scope="transport" action="remove"/>
         <property name="Content-Length" scope="transport" action="remove"/>
         <property name="Content-Type" scope="transport" action="remove"/>
         <property name="User-Agent" scope="transport" action="remove"/>
         <property name="OUT_ONLY" value="true"/>
         <send>
            <endpoint>
               <address uri="jms:/queue/mySampleQueue?transport.jms.ConnectionFactoryJNDIName=QueueConnectionFactory&java.naming.factory.initial=org.jnp.interfaces.NamingContextFactory&java.naming.provider.url=jnp://localhost:1099&transport.jms.DestinationType=queue"/>
            </endpoint>
         </send>
      </inSequence>
      <outSequence>
         <send/>
      </outSequence>
   </target>
   <publishWSDL uri="file:repository/samples/resources/proxy/sample_proxy_1.wsdl"/>
   <description>HornetQ-WSO2 ESB sample</description>
</proxy>

NOTES :
[a] We are creating a proxy for the SimpleStockQuoteService shipped with the ESB samples.  You don’t really need to have the actual service running for this example, as our plan is to push the incoming payload into the JMS queue (and not calling the actual back-end service). 

[b] Extra HTTP headers are removed to present only the XML content for validation. 

[c] This is only a one way operation pushing the payload into the queue, hence OUT_ONLY is set to true. 

[d] You may need to change the hostname, port etc. on the JMS endpoint string matching to your environment. 

6.       Start the WSO2 ESB server and use a client application of your choice to call the Proxy service endpoint.  Here I am using soapUI to create a request calling the placeOrder operation of the proxied SimpleStockQuoteService.

7.       Multiple ways to check if the payload was successfully pushed to the JMS queue in HornetQ, but I used a simple Java client to pull the message out from mySampleQueue.

public class HornetQJMSTestClient {

       public static void main(String[] args) throws Throwable {

              // Step 1. Create an initial context to perform the JNDI lookup.              Hashtable<String, String> env = new Hashtable<String, String>();
              env.put(Context.PROVIDER_URL, "jnp://localhost:1099");
              env.put(Context.INITIAL_CONTEXT_FACTORY,
                           "org.jnp.interfaces.NamingContextFactory");
              env.put(Context.URL_PKG_PREFIXES,
                           "org.jboss.naming:org.jnp.interfaces  ");
              Context ctx = new InitialContext(env);

              // Step 2. Lookup the connection factory              ConnectionFactory cf = (ConnectionFactory) ctx
                           .lookup("/ConnectionFactory");

              // Step 3. Lookup the JMS queue              Queue queue = (Queue) ctx.lookup("/queue/mySampleQueue ");

              // Step 4. Create the session
              Connection connection = cf.createConnection();
              Session session = connection.createSession(false,
                           Session.AUTO_ACKNOWLEDGE);

              // Step 5. Create a JMS Message Consumer to receive message
              MessageConsumer messageConsumer = session.createConsumer(queue);

              // Step 6. Start the Connection so that the server
              connection.start();

              // Step 7. Receive the message              TextMessage messageReceived = (TextMessage) messageConsumer
                           .receive(5000);
              System.out.println("Received message: " + messageReceived.getText());

              // clean up all the JMS resources              connection.close();

       }
 } 

NOTE :If you need to configure the WSO2 ESB as a JMS consumer, you will need to enable the transportReceiver block with HornetQ configuration parameters as follows on the axis2.xml file.


<transportReceiver name="jms"
class="org.apache.axis2.transport.jms.JMSListener">
<parameter name="myTopicConnectionFactory" locked="false">
<parameter name="java.naming.factory.initial" locked="false">org.jnp.interfaces.NamingContextFactory</parameter>
<parameter name="java.naming.factory.url.pkgs" locked="false">org.jboss.naming:org.jnp.interfaces</parameter>
<parameter name="java.naming.provider.url" locked="false">jnp://localhost:1099</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.jnp.interfaces.NamingContextFactory</parameter>
<parameter name="java.naming.factory.url.pkgs" locked="false">org.jboss.naming:org.jnp.interfaces</parameter>
<parameter name="java.naming.provider.url" locked="false">jnp://localhost:1099</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.jnp.interfaces.NamingContextFactory</parameter>
<parameter name="java.naming.factory.url.pkgs" locked="false">org.jboss.naming:org.jnp.interfaces</parameter>
<parameter name="java.naming.provider.url" locked="false">jnp://localhost:1099</parameter>
<parameter name="transport.jms.ConnectionFactoryJNDIName"
locked="false">QueueConnectionFactory</parameter>
<parameter name="transport.jms.ConnectionFactoryType"
locked="false">queue</parameter>
</parameter>

</transportReceiver>