Archive for April 2013

First, to learn what is broker fail-over and how it is working with WSO2 Message Broker, read article "Configure Broker Fail-over in a WSO2 Message Broker and WSO2 ESB setup".

Why Broker Clustering?

Broker clustering is somewhat related to fail-over, but there are differences. In the above described fail-over case messages sent to the first broker is not visible to the second broker, as they are completely independent broker instances.

Thus when first broker goes down and it fails over to the second one, messages sent to the first broker is lost or message delivering order is broken.

For an example think of a system that in order message delivery is a mandatory requirement. If you receive messages, process it and insert processed information to a database, and for the second set of data to be inserted if it is mandatory first set of data is inserted, that has the above requirement. On the other hand, it cannot be neglected that HA must be there to get rid of single-point-of-failure.

There you need a Message Broker with proper clustering support.

Another scenario you might want clustering is to deal with "high messaging rates". When using one message broker instance if a particular "queue" gets many incoming messages and there are many subscribers for that queue, subscribers might starve waiting for a message to be delivered by the broker as the broker is hot with dumping incoming messages to the message store and also trying to deliver messages across all subscribers.

Scalable solution to above problem would be clustering. Some subscribers and publishers will be handled by a certain MB instance in the cluster. Whole load will be distributed among the nodes and as a result there will be no (or very little) net slowing down in publisher perspective or subscriber perspective.

If subscribers in a particular node consume messages fast, more messages should be dynamically routed to that node for delivery.

Further more, there should be a distributed algorithm to keep in order message delivery across the cluster.

WSO2 Message Broker 2.1.0 supports all above Fail-over and clustering scenarios and requirements.

A Broker Clustering Scenario with WSO2 MB

Setting up WSO2 MB Cluster

This is very straightforward and easy. There are various deployment patterns how WSO2 Message Broker can be clustered described here.

In this article we will follow simplest deployment pattern i.e "Start External Cassandra and ZooKeeper Servers and Point all Broker Nodes to Them"

Setup MB instances as described in the post "Configure Broker Fail-over in a WSO2 Message Broker and WSO2 ESB setup".
Now we will change the setup so that those 3 instances are members in a broker cluster rather than three standalone instances.
Start a Apache Cassandra Server on some machine as described here.
Start a Apache ZooKeeper Serveron some machine as described here.

As depicted in the same article, navigate to MBx/repository/conf/advanced/qpid-config.xml file and there you need to do two things.

Enable clustering
Change the server of the Zookeeper

<clustering>
 
          <enabled>true</enabled>
          <OnceInOrderSupportEnabled>false</OnceInOrderSupportEnabled>
          <externalCassandraServerRequired>true</externalCassandraServerRequired>
          <externalZookeeperServerRequired>true</externalZookeeperServerRequired>
          <coordination>
              <!-- Apache Zookeeper Address -->
              <ZooKeeperConnection>ZK_IP:2181</ZooKeeperConnection>
              <!-- Format yyyy-MM-dd HH:mm:ss -->
              <ReferenceTime>2012-02-29 08:08:08</ReferenceTime>
          </coordination>
....

Then we need to configure the message store. At each Message Broker node, modify $CARBON_HOME/repository/conf/advanced/qpid-virtualhosts.xml file as follows.

  <store>
     <class>org.wso2.andes.server.store.CassandraMessageStore</class>
     <username>admin</username>
     <password>admin</password>
     <cluster>ClusterOne</cluster>
     <idGenerator>org.wso2.andes.server.cluster.coordination.TimeStampBasedMessageIdGenerator</idGenerator>
     <connectionString>CASSANDRA_IP:9160</connectionString>
  </store>

Now as Cassandra server and zooKeeper server is running with necessary credentials we can start all three Message Broker servers.

Setting up WSO2 ESB

This is same as described at post "Configure Broker Fail-over in a WSO2 Message Broker and WSO2 ESB setup".

Testing the Setup

This is same as described in above article.
Now when MB1 is killed you will notice after the fail-over messages sent to MB1 is not lost. They are delivered back at some time later.

ESB Cluster with a MB Cluster

This is also now possible.

Single point-of-failure at Cassandra or Zookeeper??

Is problem back again? Maybe. But both Cassandra and ZooKeeper can be clustered, or connected in a ring model.

Where is in-order message delivery?

This is at an experimental level still. You can enable it and play with it. Navigate to MBx/repository/conf/advanced/qpid-config.xml file and enable it in all nodes. You need a restart of all nodes.

<clustering>
   <OnceInOrderSupportEnabled>true</OnceInOrderSupportEnabled>
</clustering>

What is Broker fail-over?

High availability (HA) is an important aspect associated with any server implementation that ensures a certain degree of operational continuity when unplanned downtime events impact parts of a system. JMS (Java Message Service) 1.1 does not define such instead most of the time they are vendor specific.

The idea behind fail-over is stopping single-point-of-failure in a system. As the broker is the middle man storing and forwarding the messages, if that server goes down entire message flowing of the system will go down no matter what other servers and functions are involved. Thus in order to make a robust messaging system it is mandatory to have a fail-over mechanism.

That is there are a few instances of Message Broker servers up and running in the system, and system generally uses one broker. If that broker goes down it automatically switch to the second broker and continue messaging. If the second fails it will try next and so on. Thus as a whole system will not have a downtime.

Fail-over scenario with WSO2 Message Broker and WSO2 ESB

Setting Up MB Instances In a Single Machine

Download MB 2.1.0 from here and extract it into a folder.
Make exact copy of MB to three places and rename them as MB1, MB2 and MB3. Those will be 3 separate MB instances.

Navigate to MB1/repository/conf/carbon.xml file and define an offset 1.

    
        <Ports>
        <!-- Ports offset. This entry will set the value of the ports defined below to
         the define value + Offset.
         e.g. Offset=2 and HTTPS port=9443 will set the effective HTTPS port to 9445
         -->
         <Offset>0</Offset>

Navigate to MB2/repository/conf/carbon.xml file and define an offset 2 as in the previous step.
Navigate to MB3/repository/conf/carbon.xml file and define an offset 3 as in the previous step.
Now our 3 MB instances are ready. Go to /bin folder of each MBx folder and start servers by typing in console wso2sever.sh (or wso2server.bat in Windows).
Above instances will start on ports 5673,5674 and 5675 respectively.

Note that in the above setup it will start 3 separate MB servers along with Cassandra servers dedicated for each. Those Cassandra servers will run in ports 9160, 9161 and 9162 respectively.

Setting Up MB Instances In different Machines

Download MB 2.1.0 from here and extract it into a folder.
Make exact copy of that folder in three machines. We will say those are MB1 in machine with IP1, MB2 in machine with IP2 and MB3 in machine with IP3.
Start all above instances by navigating to /bin folder and typing in console wso2sever.sh (or wso2server.bat in Windows). All severs will start with port offset 0, which is 5672.

Setting Up WSO2 ESB Instance

We will start ESB in default port (port offset 0). We can do that as we started all other servers with different port offsets in the singe machine setup. If you are using different machines use another machine to start WSO2 ESB.

For this you need to follow the WSO2 article on "Configuring WSO2 ESB with WSO2 Message Broker".
There is only a single difference to enable fail-over across the three brokers we have setup. That is when specifying

```
connectionfactory.ConnectionFactory
```

connectionfactory.QueueConnectionFactory

connectionfactory.TopicConnectionFactory

For an example, if you hope to use single machine MB setup described above

connectionfactory.QueueConnectionFactory = amqp://admin:admin@clientID/carbon?failover='roundrobin'&cyclecount='2'&brokerlist='tcp://localhost:5673 retries='5'&connectdelay='50';tcp://localhost:5674?retries='5'&connectdelay='50';tcp://localhost:5675?retries='5'&connectdelay='50''

If you hope to use the broker setup made across several machines as described above

connectionfactory.QueueConnectionFactory = amqp://admin:admin@clientID/carbon?failover='roundrobin'&cyclecount='2'&brokerlist='tcp://IP1:5672 retries='5'&connectdelay='50';tcp://IP2:5672?retries='5'&connectdelay='50';tcp://IP3:5672?retries='5'&connectdelay='50''

We will look into the parameters in the connection url above.

Brokerlist option

brokerlist='<broker url>[;<broker url>]'

The broker list defines the various brokers that can be used for this connection. A minimum of one broker url is required additional URLs are semi-colon(';') delimited.

Broker URL format

<transport>://<host>[:<port>][?<option>='<value>'[&<option>='<value>']]

Option	Default	Description
retries	1	The number of times to retry connection to this Broker
ssl	false	Use ssl on the connection
connecttimeout	30000	How long in (milliseconds) to wait for the connection to succeed
connectdelay	none	How long in (milliseconds) to wait before attempting to reconnect

Brokerlist failover option

failover='<method>[?<options>]'

Method	Description
singlebroker	This will only use the first broker in the list.
roundrobin	This method tries each broker in turn.
nofailover	[New in 0.5] This method disables all retry and failover logic.

The current defaults are naturally to use the 'singlebroker' when only one broker is present and the 'roundrobin' method with multiple brokers. The '''method''' value in the URL may also be any valid class on the classpath that implements the FailoverMethod interface.

Options

Option	Default	Description
cyclecount	1	The number of times to loop through the list of available brokers before failure.

Testing setup

When sending messages (you can configure SOAP UI to send messages continuously one after the other) to a queue using setup described in above article, kill MB1. You will notice fail-over happens at ESB to the second broker i.e MB2. Logs will be there at ESB side indicating fail-over occurred.
If some previous messages were there at MB1, they will not be delivered.

Most of the time solution to the above and many other problems is broker clustering. Read "Why Broker Fail-over is not Enough? Broker Clustering With WSO2 Message Broker" to know the advantages of clustering and WSO2 Message Broker's support on clustering.

What are JMS Transactions?

The theories and concepts behind JMS Transactions can be found in my earlier post titled "

JMS TRANSACTIONS, DELIVERY RELIABILITY AND REAL WORLD USE CASE"

This post is like a continuation of that post.

A Real World Scenario

I would like to highlight the scenario to use JMS transactions described in my previous post above as an entering to this article.

In real world messaging systems there can be bulk message processing part triggered around time that there is a less number of "user inputs" to the system (maybe at mid-night :D).

Get user records from database collected within the day.
Process them chunk by chunk.
Write processed data to a database.
There cannot be message duplication as one record should be processed only once and written to the database. Otherwise it would seem user has requested same thing twice (or many times).
There cannot be a message lost because what you process is the user requests. If a lost happens it is analogous to that user has not made any request.

We will see how to Setup WSO2 ESB for a situation like above using WSO2 Message Broker as the JMS provider. But following example does not involve all above, rather it introduce the core configurations.

Setup WSO2 Message Broker

Download WSO2 MB 2.1.0 and extract it to a folder. That folder is referred as <MB_Home> from here on.
Navigate to <MB_Home&gt/repository/conf folder and open carbon.xml file.

Define an offset 1. (We do this as we hope to start WSO2 ESB on the default port)

    
        <Ports>
        <!-- Ports offset. This entry will set the value of the ports defined below to
         the define value + Offset.
         e.g. Offset=2 and HTTPS port=9443 will set the effective HTTPS port to 9445
         -->
         <Offset>0</Offset>

Now start WSO2 MB using wso2server.sh on Linux or wso2server.bat on Windows.

Setup WSO2 ESB

Download WSO2 ESB 4.6.0 from here and extract it to a folder. That folder is referred as <ESB_Home> from here on.

Naviage to <ESB_Home> /repository/conf/axis2 folder and edit axis2.xml file.

Edit and add JMS Sender parameters

<transportSender name="jms" class="org.apache.axis2.transport.jms.JMSSender">
    <parameter name="default" locked="false">
 <parameter name="java.naming.factory.initial" locked="false">org.wso2.andes.jndi.PropertiesFileInitialContextFactory</parameter>
 <parameter name="java.naming.provider.url" locked="false">repository/conf/jndi.properties</parameter>
 <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">ConnectionFactory</parameter>
    </parameter>
</transportSender>

Edit and add JMS Receiver parameters

<transportReceiver name="jms" class="org.apache.axis2.transport.jms.JMSListener">
    <parameter name="myQueueConnectionFactory" locked="false">
        <parameter name="java.naming.factory.initial" locked="false">org.wso2.andes.jndi.PropertiesFileInitialContextFactory</parameter>
        <parameter name="java.naming.provider.url" locked="false">repository/conf/jndi.properties</parameter>
 <parameter name="transport.jms.SessionTransacted">true</parameter>
        <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">QueueConnectionFactory</parameter>
        <parameter name="transport.jms.ConnectionFactoryType" locked="false">queue</parameter>
    </parameter>
</transportReceiver>

Note that we have following parameter which says WSO2 ESB to create a "transacted session to the Message Broker".

<parameter name="transport.jms.SessionTransacted">true</parameter>

Navigate to <ESB_Home> /repository/conf/ folfer and edit jndi.properties file. We will point to the MB instance we are running and also define a topic to be used here.

# register some connection factories
# connectionfactory.[jndiname] = [ConnectionURL]
connectionfactory.ConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'
connectionfactory.QueueConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'

# register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
queue.myQueue = myQueue

# register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
#topic.MyTopic = example.MyTopic

Put client libs needed to connect to WSO2 MB to the folder <ESB_Home>/repository/components/lib folder. Get following two jars from <MB_Home>/client-libs and copy.

andes-client-0.13.wso2v7
geronimo-jms_1.1_spec-1.1.0.wso2v1

Start WSO2 ESB by navigating to <ESB_Home>/bin and typing wso2server.sh on Linux or wso2server.bat on Windows.

Using ESB management console create a proxy for sending messages to the queue This is called "StockQuoteProxy".

<proxy name="StockQuoteProxy" transports="http" startOnLoad="true">
<target>
    <endpoint>
        <address uri="jms:/myQueue?&amp;transport.jms.DestinationType=queue"/>
    </endpoint>
    <inSequence>
        <log level="custom">
            <property name="STATE" value="message is sent to queue"/>
        </log>
        <property name="OUT_ONLY" value="true"/>
        <property name="FORCE_SC_ACCEPTED" value="true" scope="axis2"/>
    </inSequence>
    <outSequence/>
</target>
</proxy>

The next step is to create a listener proxy for the JMS queue we use. As we defined transaction property in axis2.xml file there is no need to define it here in service level. When the message is received we send it to SimpleStockQuoteService service.

<proxy name="JMSQueueListenerProxy" transports="jms" startOnLoad="true">
<target>
    <inSequence onError="CustomFaultSequence">
        <property name="OUT_ONLY" value="true"/>
        <log level="custom">
            <property name="Transaction Action" value="Committed"/>
        </log>
        <send>
            <endpoint>
                <address uri="http://localhost:9000/services/SimpleStockQuoteService" format="pox"/>
            </endpoint>
        </send>
    </inSequence>
    <outSequence/>
</target>
<parameter name="transport.jms.ContentType">
    <rules>
        <jmsProperty>contentType</jmsProperty>
        <default>application/xml</default>
    </rules>
</parameter>
<parameter name="transport.jms.ConnectionFactory">myQueueConnectionFactory</parameter>
<parameter name="transport.jms.DestinationType">queue</parameter>
<parameter name="transport.jms.Destination">myQueue</parameter>
</proxy>

Now you will see a log getting printed in MB's console indicating a subscription was created. Note that in the inSequence of the listener proxy we have a fault sequence called "CustomFaultSequence". The rationale is if something happens during mediation in inSequence this gets called and there we rollback the transaction. Following is the CustomFaultSequence.

<sequence xmlns="http://ws.apache.org/ns/synapse" name="CustomFaultSequence">
    <log level="custom" separator=",">
        <property name="text" value="***FAult Handler***"/>
    </log>
    <property name="SET_ROLLBACK_ONLY" value="true" scope="axis2"/>
    <drop/>
</sequence>

Note the "SET_ROLLBACK_ONLY" property. This is mandatory for transactional rollback.

Now go to <ESB_Home>/samples/axis2Server/src/SimpleStockQuoteService and run ant to build SimpleStockQuoteService service. Then you can start the service by navigating to <ESB_Home>/samples/axis2Server/ and typing axis2server.sh (or axis2server.bat for Windows).
Now our setup is ready.

Test "Commit" Scenario

Using SOAP UI send following message to "StockQuoteProxy". You will see a log saying that message was sent to the queue in ESB's console.

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:ser="http://services.samples" xmlns:xsd="http://services.samples/xsd">
   <soapenv:Header/>
   <soapenv:Body>
      <ser:getQuote>
         <!--Optional:-->
         <ser:request>
            <!--Optional:-->
            <xsd:symbol>IBM</xsd:symbol>
         </ser:request>
      </ser:getQuote>
   </soapenv:Body>
</soapenv:Envelope>

As soon as the message is sent, it will be picked up by the listener proxy and forward the message to "SimpleStockQuoteService". If you log into MB's management console and check the message count it will be zero, as the sent message is consumed instantly. A log will get printed at SimpleStockQuoteService console as well.

Test "Rollback" Scenario

In order to test this we have to make a synapse exception in in-sequence and trigger "CustomFaultSequence". For demo purpose we can explicitly call CustomFaultSequence within in-sequence of JMSQueueListenerProxy.
```
 <sequence key="CustomFaultSequence"/>
```
Now send a message to "StockQuoteProxy" usin SOAP UI as described above.
CustomFaultSequence sequence will get called now and message will be rollbacked. If you check the message count of the queue it will be still '1'.

Will the roll-backed Message get redelivered?

There might be a requirement to get roll-backed message to be redelivered again so that ESB can try to process the message again. Those messages will have "redelivered" JMS header value. WSO2 MB 2.1.0 supports this feature. You can configure how many times broker should try to redeliver the message before it is dropped (dead lettered) using qipd-config file of Message Broker.

Navigate to <MB_Home>/repository/conf/advanced folder and edit "qpid-config" file to have desired number of message redelivers.

 
<!--Broker will drop the message after the configured number of delivery attempts for each message.-->
     <maximumNumberOfMessageDeliveryAttempts>10</maximumNumberOfMessageDeliveryAttempts>

You can adjust this demo to suit your messaging requirement, and use WSO2 Message Broker's JMS transaction capability to build a robust messaging system without message duplications and without message lost.

What are JMS Transactions

The most reliable way to produce a message is to send a PERSISTENT message within a transaction. JMS messages are PERSISTENT by default. A transaction is a unit of work into which you can group a series of operations, such as message sends and receives, so that the operations either all succeed or all fail. For details, see Specifying Message Persistence and Using JMS API Local Transactions.

The most reliable way to consume a message is to do so within a transaction, either from a queue or from a durable subscription to a topic. For details, see Creating Temporary Destinations, Creating Durable Subscriptions, and Using JMS API Local Transactions.

For other applications, a lower level of reliability can reduce overhead and improve performance. If your application does not use transactions, it can use one of these acknowledgement modes: auto, duplicates okay, and client.

If your application uses transactions, it can choose from these transaction options: transacted session, MDB with container-managed transaction demarcation (CMTD), and MDB with bean-managed transaction demarcation (BMTD).

Transaction options:

Transacted session: An application can participate in a transaction by creating a transacted session (or local transaction). The application completely controls the message delivery by either committing or rolling back the session.
Message-driven beans with CMTD: An MDB can participate in a container transaction by specifying CMTD in the XML deployment descriptor. The transaction commits upon successful message processing or the application can explicitly roll it back.
Message-driven beans with BMTD: An MDB can choose not to participate in a container transaction by specifying BMTD in the XML deployment descriptor. The MDB programmer has to design and code programmatic transactions.

One of important things to note here is that if you enable "JMS Transactions", acknowledge modes does not have any effect.

JMS support

Transactional send
Transactional receive
Transacted producers and transacted consumers can be grouped together in a single transaction if they are created from the same session object.

Creating a JMS Transaction

Enabling a JMS transacted session happens as part of creating a Session object, as shown below.

// pub/sub connection creates a transacted TopicSession
javax.jms.TopicSession session=connect.createTopicSession(true,Session.AUTO_ACKNOWLEDGE);
// p2p connection creates a transacted QueueSession
javax.jms.QueueSession = 
 connect.createQueueSession(true,Session.AUTO_ACKNOWLEDGE);
            ...
}

The first parameter of createTopicSession() or createQueueSession() method is a Boolean indicating whether this is a transacted session. That is all we need to create a transactional session. There is no explicit begin() method. When a session is transacted, all messages sent or received using that session are automatically grouped in a transaction. The transaction remains open until either a session.rollback() or a session.commit() happens, at which point a new transaction is started. An additional Session method, isTransacted(), returns a Boolean true or false indicating whether the current session is transactional.

Distributed Transactions

Having all producers and all consumers participate in one global transaction would defeat the purpose of using a loosely coupled asynchronous messaging environment.

Sometimes it's necessary to coordinate the send or receipt of a JMS transaction with the update of another non-JMS resource, like a database or an EJB entity bean. This typically involves an underlying transaction manager that takes care of coordinating the prepare, commit, or rollback of each resource participating in the transaction. JMS provides JTA transaction interfaces for accomplishing this.

JMS providers that implement the JTA XA APIs can participate as a resource in a two-phase commit. The JMS specification provides XA versions of the following JMS objects : XAConnectionFactory, XAQueueConnection, XAQueueConnectionFactory, XAQueueSession, XASession, XATopicConnection, XATopicConnectionFactory, and XATopicSesion.

What JMS Transactions can do?

JMS transactions are very important when we come to message delivery reliability. Even though non-transactional JMS can provide a certain degree of message delivery reliability, in practical world even message acknowledgements can be lost inside a network.
Distributed transactions are important when you want to do a JMS transaction plus a Database transaction in a single transaction. A good article which discuss these can be found here.

Real World Use Case

In real world messaging systems there can be bulk message processing part triggered around time that there is a less number of "user inputs" to the system (maybe at mid-night :D).

Get user records from database collected within the day.
Process them chunk by chunk.
Write processed data to a database.
There cannot be message duplication as one record should be processed only once and written to the database. Otherwise it would seem user has requested same thing twice (or many times).
There cannot be a message lost because what you process is the user requests. If a lost happens it is analogous to that user has not made any request.

In such a situation we need a JMS broker in the middle and also JMS transactions to handle message rollback if the database write or processing failure happened. To make the system more robust we need to use "Broker Failover" allowing the system to use a backup JMS broker if current broker is crashed or unreachable.

We will see how to Setup WSO2 ESB for a situation like above using WSO2 Message Broker as the JMS provider. This is discussed in a separate post titled "Get JMS Local Transactions To Work With WSO2 ESB+WSO2 MEssage Broker Setup".

You might ask the obvious question where is WSO2 MB 2.1.0? OK, for now it is on the way. It will be released within two or three months from now on.

What are durable topic subscriptions

Durable queues keep messages around persistently for any suitable consumer to consume them. Durable queues do not need to concern themselves with which consumer is going to consume the messages at some point in the future. There is just one copy of a message that any consumer in the future can consume.

Durable topics however are different as they must logically persist an instance of each suitable message for every durable consumer - since each durable consumer gets their own copy of the message.

In synopsis:

Durable topic subscriptions persists the subscription on behalf of the client subscription when it is not online. In simple words, durable topics keep messages around persistently for any suitable consumer to consume them. Durable topic subscribers are used when an application needs to receive messages that are published even while the application is inactive.

We will see how to setup WSO2 ESB as a client who create durable topic subscriptions to WSO2 MB. The added advantage is if subscriber was down (maybe ESB instance which consume messages from topic is down temporarily) you will not loose messages that are being published to that topic (maybe from an event stream from CEP).

Setup WSO2 Message Broker

Download WSO2 MB 2.1.0 and extract it to a folder. That folder is referred as <MB_Home> from here on.
Navigate to <MB_Home&gt/repository/conf folder and open carbon.xml file.

Define an offset 1. (We do this as we hope to start WSO2 ESB on the default port)

    
        <Ports>
        <!-- Ports offset. This entry will set the value of the ports defined below to
         the define value + Offset.
         e.g. Offset=2 and HTTPS port=9443 will set the effective HTTPS port to 9445
         -->
         <Offset>0</Offset>

Now start WSO2 MB using wso2server.sh on Linux or wso2server.bat on Windows.

Setup WSO2 ESB

Download WSO2 ESB 4.6.0 from here and extract it to a folder. That folder is referred as <ESB_Home> from here on.

Naviage to <ESB_Home> /repository/conf/axis2 folder and edit axis2.xml file.

Edit and add JMS Sender parameters

<transportSender name="jms" class="org.apache.axis2.transport.jms.JMSSender">
    <parameter name="default" locked="false">
 <parameter name="java.naming.factory.initial" locked="false">org.wso2.andes.jndi.PropertiesFileInitialContextFactory</parameter>
 <parameter name="java.naming.provider.url" locked="false">repository/conf/jndi.properties</parameter>
 <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">ConnectionFactory</parameter>
    </parameter>
</transportSender>

Edit and add JMS Receiver parameters

<transportReceiver name="jms" class="org.apache.axis2.transport.jms.JMSListener">
    <parameter name="myTopicConnectionFactory" locked="false">
       <parameter name="java.naming.factory.initial" locked="false">org.wso2.andes.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>
</transportReceiver>

Navigate to <ESB_Home> /repository/conf/ folfer and edit jndi.properties file. We will point to the MB instance we are running and also define a topic to be used here.

# register some connection factories
# connectionfactory.[jndiname] = [ConnectionURL]
connectionfactory.ConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'
connectionfactory.TopicConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'

# register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
#queue.JMSMS=JMSMS

# register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
topic.myTopic = myTopic

Put client libs needed to connect to WSO2 MB to the folder <ESB_Home>/repository/components/lib folder. Get following two jars from <MB_Home>/client-libs and copy.

andes-client-0.13.wso2v7
geronimo-jms_1.1_spec-1.1.0.wso2v1

Start WSO2 ESB by navigating to <ESB_Home>/bin and typing wso2server.sh on Linux or wso2server.bat on Windows.

Using ESB management console create a proxy for sending messages to the topic.

<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="StockQuoteProxy"
       transports="http"
       startOnLoad="true">
   <target>
      <endpoint>
         <address uri="jms:/myTopic?&amp;transport.jms.DestinationType=topic"/>
      </endpoint>
      <inSequence>
         <log level="custom">
            <property name="STATE" value="message is sent to queue"/>
         </log>
         <property name="OUT_ONLY" value="true"/>
         <property name="FORCE_SC_ACCEPTED" value="true" scope="axis2"/>
      </inSequence>
      <outSequence/>
   </target>
</proxy>

Using ESB management console create a proxy for consuming messages from the topic.

<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="JMSQueueListenerProxy"
       startOnLoad="true">
   <target>
      <inSequence>
         <property name="OUT_ONLY" value="true"/>
         <log level="full"/>
         <drop/>
      </inSequence>
      <outSequence/>
   </target>
   <parameter name="transport.jms.ContentType">
      <rules>
         <jmsProperty>contentType</jmsProperty>
         <default>application/xml</default>
      </rules>
   </parameter>
   <parameter name="transport.jms.ConnectionFactory">myTopicConnectionFactory</parameter>
   <parameter name="transport.jms.DestinationType">topic</parameter>
   <parameter name="transport.jms.Destination">myTopic</parameter>
</proxy>

you will see following log message at MB's Console

[2013-04-21 09:20:36,531]  INFO {org.wso2.andes.server.store.CassandraMessageStore} -  Created Topic : myTopic
[2013-04-21 09:20:36,539]  INFO {org.wso2.andes.server.store.CassandraMessageStore} -  Registered Subscription TopicNodeQueue_1 for Topic myTopic

Testing the setup

Send the following message to StockQuoteProxy in ESB using soap UI.

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:ser="http://services.samples" xmlns:xsd="http://services.samples/xsd">
   <soapenv:Header/>
   <soapenv:Body>
      <ser:getQuote>
         <!--Optional:-->
         <ser:request>
            <!--Optional:-->
            <xsd:symbol>IBM</xsd:symbol>
         </ser:request>
      </ser:getQuote>
   </soapenv:Body>
</soapenv:Envelope>

You will see immediately message being consumed by the listener proxy.

[2013-04-21 09:38:54,548]  INFO - LogMediator To: , WSAction: "urn:getQuote", SOAPAction: "urn:getQuote", MessageID: ID:e1835e06-a8ba-3630-8124-31920227607a, Direction: request, Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><soapenv:Envelope xmlns:xsd="http://services.samples/xsd" xmlns:ser="http://services.samples"><soapenv:Body>
      <ser:getQuote>
         <!--Optional:-->
         <ser:request>
            <!--Optional:-->
            <xsd:symbol>IBM</xsd:symbol>
         </ser:request>
      </ser:getQuote>
   </soapenv:Body></soapenv:Envelope></soapenv:Body></soapenv:Envelope>

Now we will make the subscription durable by adding two additional following properties to JMSQueueListenerProxy. Here transport.jms.DurableSubscriberName is the one used to identify the durable topic subscriber.

  
<parameter name="transport.jms.SubscriptionDurable">true</parameter>
<parameter name="transport.jms.DurableSubscriberName">mySub1</parameter>

Following log message will be there at MB's console

[2013-04-21 09:34:55,508]  INFO {org.wso2.andes.server.cassandra.DefaultClusteringEnabledSubscriptionManager} -  Binding Subscription 1 to queue clientID:mySub1

Now send two more messages as before and you will observe even now it behaves as a normal topic subscription. As soon as you publish the message consumer consumes it.
Now "cut" JMSQueueListenerProxy from source view and save it to some text file. Save the modified source.

Now there is not subscriber listening to the messages published for the topic. But do not worry, messages will not be lost. we made it durable prior to closing it!! Now send three more messages to the StockQuoteProxy using soap UI.

[2013-04-21 09:40:05,205]  INFO - LogMediator STATE = message is sent to queue
[2013-04-21 09:40:06,366]  INFO - LogMediator STATE = message is sent to queue
[2013-04-21 09:40:07,710]  INFO - LogMediator STATE = message is sent to queue

Then we will register the topic subscriber again. Paste it from the file you kept it before removing from source view. Important thing to note here is that you need to use exactly same "DurableSubscriberName" you used before.
You will see 3 messages we sent while listener proxy was absent, are immediately consumed. You have not lost any sing message........

Why Broker Clustering?

A Broker Clustering Scenario with WSO2 MB

Testing the Setup

ESB Cluster with a MB Cluster

Single point-of-failure at Cassandra or Zookeeper??

Where is in-order message delivery?

What is Broker fail-over?

Fail-over scenario with WSO2 Message Broker and WSO2 ESB

Setting Up MB Instances In a Single Machine

Setting Up MB Instances In different Machines

Setting Up WSO2 ESB Instance

Brokerlist option

Broker URL format

Brokerlist failover option

Options

Testing setup

What are JMS Transactions?

JMS TRANSACTIONS, DELIVERY RELIABILITY AND REAL WORLD USE CASE"

A Real World Scenario

Setup WSO2 Message Broker

Setup WSO2 ESB

Test "Commit" Scenario

Test "Rollback" Scenario

Will the roll-backed Message get redelivered?

What are JMS Transactions

Creating a JMS Transaction

Distributed Transactions

What JMS Transactions can do?

Real World Use Case

What are durable topic subscriptions

Setup WSO2 Message Broker

Setup WSO2 ESB

Testing the setup

About Me

About

Popular Posts

Labels Cloud

Total Pageviews

Blog Archive

My Favorite Tech Blogs

Featured

Most Popular

About me

Featured

Instagram