How to configure a ESB proxy service as a consumer to listen to two message broker queues

In this blog post, we will look at how we can configure multiple transport receivers and senders with WSO2 ESB and configure a proxy service to have multiple transport receivers.

In order to test our scenario, we need to start two message broker instances.

Lets configure active MQ to run as two instances.

1. Download activemq and extract it.
2. Run the following command to create an instance of it.

$ ./activemq create instanceA
$ ./activemq create instanceB

Running these two commands will create two directories inside activemq bin directory with configuration files and start-up scripts duplicated within them. Now we can modify the configuration files to use different ports so that when we start the two mq instances, there wont be port conflicts.

Open InstanceB/conf/activemq.xml file and modify the ports under transportConnectors.

<transportConnectors>

               <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600"/>

            <transportConnector name="amqp" uri="amqp://0.0.0.0:5682?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600"/>

 </transportConnectors>

Now open jetty.xml in the same directory and modify the ui port from 8161 to different port.

Now we are ready to start the two activemq instances. 

cd instanceA/bin

./instanceA console

cd instanceB/bin

./instanceB console

Now we have two activemq instance running in console mode.

Log into activemq instanceA ui and create a queue named MyJMSQueue.
Similiary, log into activemq instanceB and create a queue with the same name.

Use http://localhost:8161/admin and username and password admin for defaults.

Now, we have done our configurations for activemq broker.

Now copy the following jar files to repository/components/lib directory of ESB.

activemq-broker-5.8.0.jar
activemq-client-5.8.0.jar
geronimo-j2ee-management_1.1_spec-1.0.1.jar
geronimo-jms_1.1_spec-1.1.1.jar
hawtbuf-1.9.jar


Configuring axis2.xml

Now go to repository/conf/axis2/axis2.xml and uncomment jms transport section for activemq and duplicate it with a transport named jms1. Make sure to update the provider url port with the value you specified in activemq.xml. My configuration looks like the following.

<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>

    <transportReceiver name="jms1" class="org.apache.axis2.transport.jms.JMSListener">

        <parameter name="myTopicConnectionFactory1" 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:61636</parameter>

                <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">TopicConnectionFactory</parameter>

                    <parameter name="transport.jms.ConnectionFactoryType" locked="false">topic</parameter>

        </parameter>

        <parameter name="myQueueConnectionFactory1" 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:61636</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:61636</parameter>

                <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">QueueConnectionFactory</parameter>

                    <parameter name="transport.jms.ConnectionFactoryType" locked="false">queue</parameter>

        </parameter>

Now start ESB and deploy the following proxy service.

<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="JMSListenerProxy"
       transports="jms jms1"
       startOnLoad="true"
       trace="disable">
   <description/>
   <target>
      <inSequence>
         <log level="full"/>
         <drop/>
      </inSequence>
   </target>
   <parameter name="transport.jms.Destination">MyJMSQueue</parameter>
</proxy>

Now if you publish message to queue MyJMSQueue of either activemq instance, you will notice that the message is consumed by our proxy service and logged.

How does it work ?

In our scenario, since we are going to have to have different configurations for transports jms and jms1, we cannot specify the connection factory details in the proxy service itself. Hence we have resorted to use the default configurations specified in the axis2.xml.

However, we can specify the jms destination name in our proxy service. This make sense as this kind of approach would only be required for mq high availability scenario and hence we can afford to have the same queue name for both message broker instances.