color IN ('red', 'green', 'white')

color = 'red' OR color = 'green' OR color = 'white'

size BETWEEN 6 AND 10

size>= 6 AND size <= 10

size> 6 OR color = 'red'

color = 'red' AND size> 6

MyMessage.setBooleanProperty("JMS_SUN_COMPRESS",true);

//topicSession and myTopic are assumed to have been created
topicPublisher publisher = topicSession.createPublisher(myTopic);
BytesMessage bytesMessage=topicSession.createBytesMessage();

//byteArray is assumed to have been created
bytesMessage.writeBytes(byteArray);

//instruct the client runtime to compress this message
bytesMessage.setBooleanProperty("JMS_SUN_COMPRESS", true);

//publish message to the myTopic destination
publisher.publish(bytesMessage);

//get uncompressed body size
int uncompressed=bytesMessage.getIntProperty("JMS_SUN_UNCOMPRESSED_SIZE");

//get compressed body size
int compressed=bytesMessage.getIntProperty("JMS_SUN_COMPRESSED_SIZE");

imqobjmgr add -t cf -l "cn=myConnectionFactory" \
    -o "imqAddressList=mq://jpgserv/jms" \
    -o "imqReconnect=true" \
    -o "imqReconnectAttempts=10"
    -j "java.naming.factory.initial =
                  com.sun.jndi.fscontext.RefFSContextFactory
    -j "java.naming.provider.url= file:///home/foo/imq_admin_objects"

imqobjmgr add -t cf -l "cn=myCF" \
    -o "imqAddressList=myhost1, mqtcp://myhost2:12345/jms" \
    -o "imqReconnect=true" \
    -o "imqReconnectRetries=5"
    -j "java.naming.factory.initial =
                  com.sun.jndi.fscontext.RefFSContextFactory
    -j "java.naming.provider.url= file:///home/foo/imq_admin_objects"

imqobjmgr add -t cf -l "cn=myConnectionFactory" \
    -o "imqAddressList=mq://myhost1/ssljms, \
            mq://myhost2/ssljms, \
            mq://myhost3/ssljms, \
            mq://myhost4/ssljms" \
    -o "imqReconnect=true" \
    -o "imqReconnectRetries=5" \
    -o "imqAddressListBehavior=RANDOM"
    -j "java.naming.factory.initial =
                  com.sun.jndi.fscontext.RefFSContextFactory
    -j "java.naming.provider.url= file:///home/foo/imq_admin_objects"

java —DimqAddressList="localhost:777"
                                  test.jmsclient.ha.FailoverQSender

java —DimqAddressList="localhost:777"
                                  test.jmsclient.ha.FailoverQReceiver

/*
 * @(#)FailoverQSender.java    1.2 07/04/20
 *
 * Copyright (c) 2000, 2019 Oracle and/or its licensees. All Rights Reserved.
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Distribution License v. 1.0, which is available at
 * http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * SPDX-License-Identifier: BSD-3-Clause
 *
 */
package test.jmsclient.ha;

import java.util.Date;
import jakarta.jms.*;
import com.sun.messaging.jms.Connection;
import com.sun.messaging.jms.notification.*;

/**
 *
 * This sample program uses a transacted session to send messages.
 * It is designed to run with test.jmsclient.ha.FailoverQReceiver
 * @version 1.0
 */
public class FailoverQSender
    implements ExceptionListener, EventListener, Runnable {
       //constant - commit property name
    public static final String COMMIT_PROPERTY_NAME = "COMMIT_PROPERTY";
    //constant - message counter
    public static final String MESSAGE_COUNTER = "counter";
    //constant - destination name
    public static final String TEST_DEST_NAME = "FailoverTestDest001";
    //queue connection
    QueueConnection conn = null;
    //session
    QueueSession session = null;
    //queue sender
    QueueSender sender = null;
    //queue destination
    Queue queue = null;

    //commmitted counter.
    private int commitCounter = 0;
    //current message counter
    private int currentCounter = 0;
    //set to true if the application is connected to the broker.
    private boolean isConnected = false;

    /**
     * Default constructor - do nothing.
     * Properties are passed in from init() method.
     */
    public FailoverQSender() {

    //set up JMS environment
         setup();
    }

    /**
     * Connection Exception listener.
     */
    public void onException (JMSException e) {

        //The run() method will exit.
        this.isConnected = false;

        log ("Exception listener is called.
              Connection is closed by MQ client runtime." );
        log (e);
    }

    /**
     * this method is called when a MQ connection event occurred.
     */
    public void onEvent (Event connectionEvent) {
        log(connectionEvent);
    }

    /**
     * Rollback the application data.
     *
     */
    private void rollBackApplication() {

        this.currentCounter = this.commitCounter;
        log ("Application rolled back., current (commit) counter: "
                                              + currentCounter);
    }

    /**
     * Roll back the current jms session.
     */
    private void rollBackJMS() {

         try {

               log("Rolling back JMS ...., commit counter: " + commitCounter);
               session.rollback();
         } catch (JMSException jmse) {
               log("Rollback failed");
               log(jmse);
               //application may decide to log and continue sending messages
               // without closing the application.
             close();
            }
   }
    /**
     * rollback application data and jms session.
     *
     */
    private void rollBackAll() {
        //rollback jms
            rollBackJMS();
          //rollback app data
            rollBackApplication();
    }

    /**
     * close JMS connection and stop the application
     *
     */
    private void close() {

        try {
          if ( conn != null ) {
                //close the connection
                conn.close();
              }
         } catch (Exception e) {
              //log exception
             log (e);
         } finally {
              //set flag to true. application thread will exit
             isConnected = false;
           }
    }

    /**
     * Send messages in a loop until the connection is closed.
     * Session is committed for each message sent.
     */
    public void run () {

        //start producing messages
        while (isConnected) {

          try {
           //reset message counter if it reaches max int value
                     checkMessageCounter();
                    //create a message
                     Message m = session.createMessage();
                    //get the current message counter value
                     int messageCounter = this.getMessageCounter();
                    //set message counter to message property
                     m.setIntProperty(MESSAGE_COUNTER, messageCounter);
                    //set commit property
                     m.setBooleanProperty(COMMIT_PROPERTY_NAME, true);
                    //send the message
                     sender.send(m);

                    log("Sending message: " + messageCounter +
                                  ", current connected broker: " +
               this.getCurrentConnectedBrokerAddress());

                    //commit the message
                     this.commit();

                    // pause 3 seconds
                     sleep(3000);

                } catch (TransactionRolledBackException trbe) {
                   //rollback app data
                    rollBackApplication();
                 } catch (JMSException jmse) {
                    if (isConnected == true) {
                     //rollback app data and JMS session
                       rollBackAll();
                     }
                }
           }
   }

  /**
     * Reset all counters if integer max value is reached.
     */
    private void checkMessageCounter() {

        if ( currentCounter == Integer.MAX_VALUE ) {
           currentCounter = 0;
           commitCounter = 0;
        }
    }

  /**
     * Set up testing parameters - connection, destination, etc
     */
 protected void setup() {
    try {
            //get connection factory
            com.sun.messaging.QueueConnectionFactory factory =
                            new com.sun.messaging.QueueConnectionFactory();
            //create a queue connection
            conn = factory.createQueueConnection();

            //set exception listener
            conn.setExceptionListener(this);

            //set event listener
            ( (com.sun.messaging.jms.Connection) conn).setEventListener(this);

            //get destination name
            String destName = TEST_DEST_NAME;

            //create a transacted session
            session = conn.createQueueSession(true, Session.AUTO_ACKNOWLEDGE);

            //get destination
            queue = session.createQueue(destName);
            //create queue sender
            sender = session.createSender(queue);

            //set isConnected flag to true.
            this.isConnected = true;

        } catch (JMSException jmse) {
            this.isConnected = false;
        }
    }

    /**
     * get the next message counter.
     */
    private synchronized int getMessageCounter () {
        return ++ currentCounter;
    }

    /**
     * commit the current transaction/session.
     */
    private void commit() throws JMSException {
        session.commit();
        this.commitCounter = currentCounter;

        log ("Transaction committed, commit counter: " +commitCounter);
    }

    /**
     * Get the current connencted broker address.
     */
    private String getCurrentConnectedBrokerAddress() {
        return ((com.sun.messaging.jms.Connection)conn).getBrokerAddress();
    }

    /**
     * log a string message.
     * @param msg
     */
    private synchronized void log (String msg) {
        System.out.println(new Date() + ": " + msg);
    }

    /**
     * Log an exception received.
     */
    private synchronized void log (Exception e) {
        System.out.println(new Date() + ": Exception:");
        e.printStackTrace();
    }
    /**
     * Log the specified MQ event.
     */
    private synchronized void log (Event event) {

      try {
        System.out.println(new Date() + ": Received MQ event notification.");
        System.out.println("*** Event code: " + event.getEventCode() );
        System.out.println("*** Event message: " + event.getEventMessage());
      } catch (Exception e) {
          e.printStackTrace();
      }
    }
    /**
     * pause the specified milli seconds.
     */
    private void sleep (long millis) {
      try {
         Thread.sleep(millis);
      } catch (java.lang.InterruptedException inte) {
         log (inte);
      }
    }
    /**
     * The main program.
     */
    public static void main (String args[]) {
        FailoverQSender fp = new FailoverQSender();
        fp.run();
    }
}

/*
 * @(#)FailoverQReceiver.java    1.4 07/04/20
 *
 * Copyright (c) 2000, 2019 Oracle and/or its licensees. All Rights Reserved.
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Distribution License v. 1.0, which is available at
 * http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * SPDX-License-Identifier: BSD-3-Clause
 *
 */
package test.jmsclient.ha;

import java.util.Date;
import java.util.Vector;
import jakarta.jms.*;
import com.sun.messaging.jms.notification.*;

/**
 * This sample program uses a transacted session to receive messages.
 * It is designed to run with test.jmsclient.ha.FailoverQSender.
 *
 * @version 1.0
 */
public class FailoverQReceiver
    implements ExceptionListener, EventListener, Runnable {

    //queue connection
    private QueueConnection conn = null;
    //queue session
    private QueueSession session = null;
    //qreceiver
    private QueueReceiver qreceiver = null;
    //queue destination
    private Queue queue = null;

    //commmitted counter.
    private int commitCounter = 0;
    //flag to indicate if the connection is connected to the broker.
    private boolean isConnected = false;
    //flag to indicate if current connection is to HA broker cluster.
    private boolean isHAConnection = false;
    //application data holder.
    private Vector data = new Vector();

    /**
     * Default constructor - JMS setup.
     */
    public FailoverQReceiver() {
        //set up JMS environment
              setup();
    }

    /**
     * Connection Exception listener.
     */
    public void onException (JMSException e) {

        //The run() method will exit.
        this.isConnected = false;

        log ("Exception listener is called. Connection is closed
                               by MQ client runtime." );
        log (e);
    }

    /**
     * log the connection event.
     */
    public void onEvent (Event connectionEvent) {
       log (connectionEvent);
    }

    /**
     * Roll back application data.
     */
    private void rollBackApplication() {
        //reset application data
            this.reset();

        log ("Rolled back application data, current commit counter:
             " + commitCounter);
    }

    /**
     * Clear the application data for the current un-committed transaction.
     */
    private void reset() {
        data.clear();
    }

    /**
     * Roll back JMS transaction and application.
     */
    private void rollBackAll() {
        try {
           //rollback JMS
               rollBackJMS();
              //rollback application data
            rollBackApplication();
        } catch (Exception e) {

            log ("rollback failed. closing JMS connection ...");

        //application may decide NOT to close connection if rollback failed.
            close();
        }
    }

    /**
     * Roll back jms session.
     */
    private void rollBackJMS() throws JMSException {
              session.rollback();
              log("JMS session rolled back ...., commit counter:
             " + commitCounter);

       }

  /**
     * Close JMS connection and exit the application.
    */
    private void close() {
        try {
            if ( conn != null ) {
                conn.close();
            }
        } catch (Exception e) {
            log (e);
        } finally {
            isConnected = false;
        }
    }

    /**
     * Receive, validate, and commit messages.
     */
    public void run () {

            //produce messages
            while (isConnected) {

                 try {
            //receive message
             Message m = qreceiver.receive();
            //process message -- add message to the data holder
             processMessage(m);
            //check if the commit flag is set in the message property
             if ( shouldCommit(m) ) {
                //commit the transaction
                commit(m);
             }

               } catch (TransactionRolledBackException trbe) {
                log ("transaction rolled back by MQ  ...");
                        //rollback application data
                rollBackApplication();
               } catch (JMSException jmse) {
                       //The exception can happen when receiving messages
              //and the connected broker is killed.
                if ( isConnected == true ) {
                             //rollback MQ and application data
                    rollBackAll();
                         }

                } catch (Exception e) {
                log (e);

                //application may decide NOT to close the connection
                //when an unexpected Exception occurred.
                close();
               }
            }
    }

    /**
     * Set up testing parameters - connection, destination, etc
     */
    protected void setup() {
      try {
               //get connection factory
               com.sun.messaging.QueueConnectionFactory factory =
                 new com.sun.messaging.QueueConnectionFactory();

               //create jms connection
               conn = factory.createQueueConnection();

               //set exception listener
               conn.setExceptionListener(this);

               //set event listener
               ( (com.sun.messaging.jms.Connection) conn).setEventListener(this);

               //test if this is a HA connection
               isHAConnection = ( (com.sun.messaging.jms.Connection)
                            conn).isConnectedToHABroker();
               log ("Is connected to HA broker cluster: " + isHAConnection);

               //get destination name
               String destName = FailoverQSender.TEST_DEST_NAME;

               //create a transacted session
               session = conn.createQueueSession(true, -1);

               //get destination
               queue = session.createQueue(destName);

               //create queue receiver
               qreceiver = session.createReceiver(queue);
               //set isConnected flag to true
            isConnected = true;
        //start the JMS connection
            conn.start();
            log("Ready to receive on destination: " + destName);
        } catch (JMSException jmse) {
               isConnected = false;
                log (jmse);
        close();
     }
    }

    /**
     * Check if we should commit the transaction.
     */
    private synchronized boolean shouldCommit(Message m) {

        boolean flag = false;

        try {
            //get the commit flag set by the FailoverQSender
            flag = m.getBooleanProperty(FailoverQSender.COMMIT_PROPERTY_NAME);

            if ( flag ) {
                //check if message property contains expected message counter
                validate (m);
            }

        } catch (JMSException jmse) {
            log (jmse);
        }

        return flag;
    }

    /**
     * A very simple validation only.  More logic may be added to validate
     * message ordering and message content.
     * @param m Message  The last message received for the current transaction.
     */
    private void validate (Message m) {

      try {
       //get message counter property
         int counter = m.getIntProperty(FailoverQSender.MESSAGE_COUNTER);
       //The counter is set sequentially and must be received in right order.
       //Each message is committed after validated.
            if (counter != (commitCounter + 1)) {
                        this.printData();
                        throw new RuntimeException("validation failed.");
                    }

            log ("messages validated.  ready to commit ...");
            } catch (JMSException jmse) {
              log (jmse);

            printData();

            throw new RuntimeException("Exception occurred during validation:
                                        " + jmse);
        }
    }

    /**
     * Get the message counter and put it in the data holder.
     * @param m the current message received
     */
    private synchronized void processMessage(Message m)  throws JMSException {

        // get message counter. this value is set by the FailoverQSender.
        int ct = m.getIntProperty(FailoverQSender.MESSAGE_COUNTER);
        // log the message
        log("received message: " + ct
                 +", current connected broker:
        " + this.getCurrentConnectedBrokerAddress());

        // saved the data in data holder.
        data.addElement(new Integer(ct));
    }

    /**
     * commit the current transaction.
     * @param m the last received message to be committed.
     * @throws JMSException if commit failed.
     */
    private void commit(Message m) throws JMSException {
     //commit the transaction
        session.commit();

        //get the current message counter
        int counter = m.getIntProperty(FailoverQSender.MESSAGE_COUNTER);
        //set the commit counter

        commitCounter = counter;
        //clear app data
        this.reset();

        log ("Messages committed, commitCounter: " + commitCounter);
    }

    /**
     * log exception.
     */
    private synchronized void log (Exception e) {
    System.out.println(new Date() + ": Exception Stack Trace: ");
        e.printStackTrace();
    }

    /**
     * log connection event.
     */
    private synchronized void log (Event event) {

        try {
            System.out.println(new Date()
                 + ": Received MQ event notification.");
            System.out.println("*** Event Code: " + event.getEventCode() );
            System.out.println("*** Event message: " + event.getEventMessage());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Log the specified message.
     */
    private void log (String msg) {
    System.out.println(new Date() + ": " + msg);
    }

    /**
     * print values stored in the data holder.
     *
     */
    private void printData() {

        for ( int i=0; i< data.size(); i++) {
            log (" *** data index " + i + " = " + data.elementAt(i) );
        }
    }

    private String getCurrentConnectedBrokerAddress() {
        return ((com.sun.messaging.jms.Connection)conn).getBrokerAddress();
    }
    /**
     * The main method.  This starts the failover queue receiver.
     */
    public static void main (String args[]) {
        FailoverQReceiver fqr = new FailoverQReceiver();
        fqr.run();
    }

}

java dura.example.FailoverProducer

java -DimqReconnectEnabled=true dura.example.FailoverProducer

/*
 * @(#)FailoverProducer.java    1.1 06/06/09
 * Copyright (c) 2000, 2019 Oracle and/or its licensees. All Rights Reserved.
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Distribution License v. 1.0, which is available at
 * http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * SPDX-License-Identifier: BSD-3-Clause
 *
 */

package dura.example;

import jakarta.jms.*;
import com.sun.messaging.ConnectionConfiguration;
import java.util.*;

public class FailoverProducer implements ExceptionListener {

    //connection factory
    private com.sun.messaging.TopicConnectionFactory factory;
    //connection
    private TopicConnection pconn = null;
    //session
    private TopicSession psession = null;
    //publisher
    private TopicPublisher publisher = null;
    //topic
    private Topic topic = null;
    //This flag indicates whether this test client is closed.
    private boolean isClosed = false;
    //auto reconnection flag
    private boolean autoReconnect = false;
    //destination name for this example.
    private static final String DURA_TEST_TOPIC = "DuraTestTopic";
    //the message counter property name
    public static final String MESSAGE_COUNTER = "MESSAGE_COUNTER";
    //the message in-doubt-bit property name
    public static final String MESSAGE_IN_DOUBT = "MESSAGE_IN_DOUBT";

    /**
     * Constructor.  Get imqReconnectEnabled property value from
     * System property.
     */
    public FailoverProducer () {

        try {
            autoReconnect =
            Boolean.getBoolean(ConnectionConfiguration.imqReconnectEnabled);
        } catch (Exception e) {
            this.printException(e);
        }

    }

    /**
     * Connection is broken if this handler is called.
     * If autoReconnect flag is true, this is called only
     * if no more retries from MQ.
     */
    public void onException (JMSException jmse) {
        this.printException (jmse);
    }

    /**
     * create MQ connection factory.
     * @throws JMSException
     */
    private void initFactory() throws JMSException {
        //get connection factory
        factory = new com.sun.messaging.TopicConnectionFactory();
    }

    /**
     * JMS setup.  Create a Connection,Session, and Producer.
     *
     * If any of the JMS object creation fails (due to system failure),
     * it retries until it succeeds.
     *
     */
    private void initProducer() {

        boolean isConnected = false;

        while ( isClosed == false && isConnected == false ) {

            try {
                println("producer client creating connection ...");

                //create connection
                pconn = factory.createTopicConnection();

                //set connection exception listener
                pconn.setExceptionListener(this);

                //create topic session
                psession = pconn.createTopicSession(false,
                    Session.CLIENT_ACKNOWLEDGE);

                //get destination
                topic = psession.createTopic(DURA_TEST_TOPIC);

                //publisher
                publisher = psession.createPublisher(topic);

                //set flag to true
                isConnected = true;

                println("producer ready.");
            }
            catch (Exception e) {

                println("*** connect failed ... sleep for 5 secs.");

                try {
                    //close resources.
                    if ( pconn != null ) {
                        pconn.close();
                    }
                    //pause 5 secs.
                    Thread.sleep(5000);

                } catch (Exception e1) {
                    ;
                }
            }
        }
    }

    /**
     * Start test.  This sends JMS messages in a loop (forever).
     */
    public void run () {

        try {
            //create MQ connection factory.
            initFactory();

            //create JMS connection,session, and producer
            initProducer();

            //send messages forever.
            sendMessages();
        } catch (Exception e) {
            this.printException(e);
        }
    }

    /**
     * Send persistent messages to a topic forever.  This shows how
     * to handle failover for a message producer.
     */
    private void sendMessages() {

        //this is set to true if send failed.
        boolean messageInDoubt = false;

        //message to be sent
        TextMessage m = null;

        //msg counter
        long msgcount = 0;

        while (isClosed == false) {

            try {

                /**
                 * create a text message
                 */
                m = psession.createTextMessage();

                /**
                 * the MESSAGE_IN_DOUBT bit is set to true if
                 * you get an exception for the last message.
                 */
                if ( messageInDoubt == true ) {
                    m.setBooleanProperty (MESSAGE_IN_DOUBT, true);
                    messageInDoubt = false;

                    println("MESSAGE_IN_DOUBT bit is set to true
                             for msg: " + msgcount);
                } else {
                    m.setBooleanProperty (MESSAGE_IN_DOUBT, false);
                }

                //set message counter
                m.setLongProperty(MESSAGE_COUNTER, msgcount);

                //set message body
                m.setText("msg: " + msgcount);

                //send the msg
                publisher.send(m, DeliveryMode.PERSISTENT, 4, 0);

                println("sent msg: " + msgcount);

                /**
                 * reset counetr if reached max long value.
                 */
                if (msgcount == Long.MAX_VALUE) {
                    msgcount = 0;

                    println ("Reset message counter to 0.");
                }

                //increase counter
                msgcount ++;

                Thread.sleep(1000);

            } catch (Exception e) {

                if ( isClosed == false ) {

                    //set in doubt bit to true.
                    messageInDoubt = true;

                    this.printException(e);

                    //init producer only if auto reconnect is false.
                    if ( autoReconnect == false ) {
                        this.initProducer();
                    }
                }
            }
        }
    }

    /**
     * Close this example program.
     */
    public synchronized void close() {

        try {
            isClosed = true;
            pconn.close();

            notifyAll();
        } catch (Exception e) {
            this.printException(e);
        }
    }

    /**
     * print the specified exception.
     * @param e the exception to be printed.
     */
    private void printException (Exception e) {
        System.out.println(new Date().toString());
        e.printStackTrace();
    }

    /**
     * print the specified message.
     * @param msg the message to be printed.
     */
    private void println (String msg) {
        System.out.println(new Date() + ": " + msg);
    }

    /**
     * Main program to start this example.
     */
    public static void main (String args[]) {
        FailoverProducer fp = new FailoverProducer();
        fp.run();
    }

}

java dura.example.FailoverConsumer

/*
 * @(#)FailoverConsumer.java    1.1 06/06/09
 * Copyright (c) 2000, 2019 Oracle and/or its licensees. All Rights Reserved.
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Distribution License v. 1.0, which is available at
 * http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * SPDX-License-Identifier: BSD-3-Clause
 *
 */
package dura.example;

import java.util.Date;
import jakarta.jms.Destination;
import jakarta.jms.ExceptionListener;
import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.Connection;
import jakarta.jms.MessageConsumer;
import jakarta.jms.Session;
import jakarta.jms.Topic;
import jakarta.jms.TransactionRolledBackException;
import com.sun.messaging.ConnectionConfiguration;

public class FailoverConsumer implements ExceptionListener, Runnable {

    //JMS connection
    private Connection conn = null;
    //JMS session
    private Session session = null;
    //JMS Message consumer
    private MessageConsumer messageConsumer = null;
    //JMS destination.
    private Destination destination = null;
    //flag indicates whether this program should continue running.
    private boolean isConnected = false;
    //destination name.
    private static final String DURA_TEST_TOPIC = "DuraTestTopic";
    //the commit counter, for information only.
    private long commitCounter = 0;

    /**
     * message counter property set by the producer.
     */
    public static final String MESSAGE_COUNTER = "MESSAGE_COUNTER";

    /**
     * Message in doubt bit set by the producer
     */
    public static final String MESSAGE_IN_DOUBT = "MESSAGE_IN_DOUBT";

    /**
     * receive time out
     */
    public static final long RECEIVE_TIMEOUT = 0;

    /**
     * Default constructor -
     * Set up JMS Environment.
     */
    public FailoverConsumer() {
       setup();
    }

    /*  Connection Exception listener.  This is called when connection
     *  breaks and no reconnect attempts are performed by MQ client runtime.
     */
    public void onException (JMSException e) {

        print ("Reconnect failed.  Shutting down the connection ...");

        /**
         * Set this flag to false so that the run() method will exit.
         */
        this.isConnected = false;
        e.printStackTrace();
    }

    /**
     * Best effort to roll back a jms session.  When a broker crashes, an
     * open-transaction should be rolled back.  But the re-started broker
     * may not have the uncommitted tranaction information due to system
     * failure.  In a situation like this, an application can just quit
     * calling rollback after retrying a few times  The uncommitted
     * transaction (resources) will eventually be removed by the broker.
     */
    private void rollBackJMS() {

        //rollback fail count
        int failCount = 0;

        boolean keepTrying = true;

        while ( keepTrying ) {

            try {

                print ("<<< rolling back JMS ...., consumer commit counter:
                          " +  this.commitCounter);

                session.rollback();

                print("<<< JMS rolled back ...., consumer commit counter:
                          " + this.commitCounter);
                keepTrying = false;
            } catch (JMSException jmse) {

                failCount ++;
                jmse.printStackTrace();

                sleep (3000); //3 secs

                if ( failCount == 1 ) {

                    print ("<<< rollback failed : total count" + failCount);
                    keepTrying = false;
                }
            }
        }
    }


    /**
     * Close the JMS connection and exit the program.
     *
     */
    private void close() {
        try {

            if ( conn != null ) {
                conn.close();
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            this.isConnected = false;
        }
    }

    /*Receive messages in a loop until closed.*/

    public void run () {

        while (isConnected) {

            try {

                /*receive message with specified timeout.*/

                Message m = messageConsumer.receive(RECEIVE_TIMEOUT);

                /* process the message. */
                processMessage(m);

                /* commit JMS transaction. */
                this.commit();

                /*increase the commit counter.*/
                this.commitCounter ++;

            } catch (TransactionRolledBackException trbe) {

                /**
                 * the transaction is rolled back
                 * a new transaction is automatically started.
                 */
                trbe.printStackTrace();
            } catch (JMSException jmse) {

                /* The transaction is in unknown state.
        * We need to roll back the transaction.*/

                jmse.printStackTrace();

                /* roll back if not closed.
                 */
                if ( this.isConnected == true ) {
                    this.rollBackJMS();
                }

            } catch (Exception e) {

                e.printStackTrace();

                /* Exit if this is an unexpected Exception.
                 */
                this.close();

            } finally {
                ;//do nothing
            }
        }

        print(" <<< consumer exit ...");
    }

    /*Set up connection, destination, etc*/
       /
    protected void setup() {

        try {

            //create connection factory
            com.sun.messaging.ConnectionFactory factory =
            new com.sun.messaging.ConnectionFactory();

            //set auto reconnect to true.
            factory.setProperty(ConnectionConfiguration.imqReconnectEnabled, "true");
            //A value of -1 will retry forever if connection is broken.
            factory.setProperty(ConnectionConfiguration.imqReconnectAttempts, "-1");
            //retry interval - every 10 seconds
            factory.setProperty(ConnectionConfiguration.imqReconnectInterval, "10000");
            //create connection
            conn = factory.createConnection();
            //set client ID
            conn.setClientID(DURA_TEST_TOPIC);

            //set exception listener
            conn.setExceptionListener(this);

            //create a transacted session
            session = conn.createSession(true, -1);

            //get destination
            destination = session.createTopic(DURA_TEST_TOPIC);

            //message consumer
            messageConsumer = session.createDurableSubscriber((Topic)destination,
                                                       DURA_TEST_TOPIC);
            //set flag to true
            this.isConnected = true;
            //we are ready, start the connection
            conn.start();

            print("<<< Ready to receive on destination: " + DURA_TEST_TOPIC);

        } catch (JMSException jmse) {
            this.isConnected = false;
            jmse.printStackTrace();

            this.close();
        }
    }

    /**
     * Process the received message message.
     *  This prints received message counter.
     * @param m the message to be processed.
     */
    private synchronized void processMessage(Message m) {

        try {
            //in this example, we do not expect a timeout, etc.
            if ( m == null ) {
                throw new RuntimeException ("<<< Received null message.
                                     Maybe reached max time out. ");
            }

            //get message counter property
            long msgCtr = m.getLongProperty (MESSAGE_COUNTER);

            //get message in-doubt bit
            boolean indoubt = m.getBooleanProperty(MESSAGE_IN_DOUBT);

            if ( indoubt) {
                print("<<< received message: " + msgCtr + ", indoubt bit is true");
            } else {
                print("<<< received message: " + msgCtr);
            }

        } catch (JMSException jmse) {
            jmse.printStackTrace();
        }
    }

    /**
     * Commit a JMS transaction.
     * @throws JMSException
     */
    private void commit() throws JMSException {
        session.commit();
    }

    /**
     * Sleep for the specified time.
     * @param millis sleep time in milli-seconds.
     */
    private void sleep (long millis) {
        try {
            Thread.sleep(millis);
        } catch (java.lang.InterruptedException inte) {
            print (inte);
        }
    }

    /**
     * Print the specified message.
     * @param msg the message to be printed.
     */
    private static void print (String msg) {
        System.out.println(new Date() + ": " + msg);
    }

    /**
     * Print Exception stack trace.
     * @param e the exception to be printed.
     */
    private static void print (Exception e) {
        System.out.print(e.getMessage());
        e.printStackTrace();
    }

    /**
     * Start this example program.
     */
    public static void main (String args[]) {
        FailoverConsumer fc = new FailoverConsumer();
        fc.run();
    }

}