3 Message Queue Clients: Design and Features

Developing Portable Clients

The Java Messaging Specification was developed to abstract access to message-oriented middleware systems (MOMs). A client that writes JMS code should be portable to any provider that implements this specification. If code portability is important to you, be sure that you do the following in developing clients:

• Make sure your code does not depend on extensions or features that are specific to Message Queue.

• Look up, using JNDI, (rather than instantiate) administered objects for connection factories and destinations.

  Administered objects encapsulate provider-specific implementation and configuration information. Besides allowing for portability, administered objects also make it much easier to share connection factories between applications and to tune a JMS application for performance and resource use. So, even if portability is not important to you, you might still want to leave the work of creating and configuring these objects to an administrator. For more information, see Looking Up a Connection Factory With JNDI and Looking Up a Destination With JNDI.

Choosing Messaging Domains

As described in "Messaging Domains" in Oracle GlassFish Server Message Queue Technical Overview, JMS supports two distinct message delivery models: point-to-point and publish/subscribe. These two message delivery models can be handled using different API objects—with slightly different semantics—representing different programming domains, as shown in Table 3-1, or they can be handled by base (unified domain) types.

Using the point-to-point or publish/subscribe domains offers the advantage of a clean API that prevents certain types of programming errors; for example, creating a durable subscriber for a queue destination. However, the non-unified domains have the disadvantage that you cannot combine point-to-point and publish/subscribe operations in the same transaction or in the same session. If you need to do that, you should choose the unified domain API.

The JMS 1.1 specification continues to support the separate JMS 1.02 programming domains. (The example applications included with the Message Queue product as well as the code examples provided in this book all use the separate JMS 1.02 programming domains.) You can choose the API that best suits your needs. The only exception are those developers needing to write clients for the Sun Java System Application Server 7 environment, as explained in the following note.

Connections and Sessions

A connection is a relatively heavy-weight object because of the authentication and communication setup that must be done when a connection is created. For this reason, it's a good idea to use as few connections as possible. The real allocation of work occurs in sessions, which are light-weight, single-threaded contexts for producing and consuming messages. When you are thinking about structuring your client, it is best to think of the work that is done at the session level.

A session

• Is a factory for its message producers and consumers

• Supplies provider-optimized message factories

• Supports a single series of transactions that combine work spanning its producers and consumers into atomic units

• Defines a serial order for the messages it consumes and the messages it produces

• Retains messages until they have been acknowledged

• Serializes execution of message listeners registered with its message consumers

The requirement that sessions be operated on by a single thread at a time places some restrictions on the combination of producers and consumers that can use the same session. In particular, if a session has an asynchronous consumer, it may not have any other synchronous consumers. For a discussion of the connection and session's use of threads, see Managing Client Threads. With the exception of these restrictions, let the needs of your application determine the number of sessions, producers, and consumers.

Producers and Consumers

Aside from the reliability your client requires, the design decisions that relate to producers and consumers include the following:

• Do you want to use a point-to-point or a publish/subscribe domain?

  There are some interesting permutations here. There are times when you would want to use publish/subscribe even when you have only one subscriber. On the other hand, performance considerations might make the point-to-point model more efficient than the publish/subscribe model, when the work of sorting messages between subscribers is too costly. Sometimes You cannot make these decisions cannot in the abstract, but must actually develop and test different prototypes.

• Are you using an asynchronous message consumer that does not receive messages often or a producer that is seldom used?

  Let the administrator know how to set the ping interval, so that your client gets an exception if the connection should fail. For more information see Using the Client Runtime Ping Feature.

• Are you using a synchronous consumer in a distributed application?

  You might need to allow a small time interval between connecting and calling the receiveNoWait() method in order not to miss a pending message. For more information, see Synchronous Consumption in Distributed Applications.

• Do you need message compression?

  Benefits vary with the size and format of messages, the number of consumers, network bandwidth, and CPU performance; and benefits are not guaranteed. For a more detailed discussion, see Message Compression.

Balancing Reliability and Performance

Reliable messaging is implemented in a variety of ways: through the use of persistent messages, acknowledgments or transactions, durable subscriptions, and connection failover.

In general, the more reliable the delivery of messages, the more overhead and bandwidth are required to achieve it. The trade-off between reliability and performance is a significant design consideration. You can maximize performance and throughput by choosing to produce and consume nonpersistent messages. On the other hand, you can maximize reliability by producing and consuming persistent messages in a transaction using a transacted session. For a detailed discussion of design options and their impact on performance, see Factors Affecting Performance.

JMS Threading Restrictions

The Java Messaging Specification mandates that a session not be operated on by more than one thread at a time. This leads to the following restrictions:

• A session may not have an asynchronous consumer and a synchronous consumer.

• A session that has an asynchronous consumer can only produce messages from within the onMessage() method (the message listener). The only call that you can make outside the message listener is to close the session.

• A session may include any number of synchronous consumers, any number of producers, and any combination of the two. That is, the single-thread requirement cannot be violated by these combinations. However, performance may suffer.

The system does not enforce the requirement that a session be single threaded. If your client application violates this requirement, you will get a JMSIllegalState exception or unexpected results.

Thread Allocation for Connections

When the Message Queue client runtime creates a connection, it creates two threads: one for consuming messages from the socket, and one to manage the flow of messages for the connection. In addition, the client runtime creates a thread for each client session. Thus, at a minimum, for a connection using one session, three threads are created. For a connection using three sessions, five threads are created, and so on.

Managing threads in a JMS application often involves trade-offs between performance and throughput. Weigh the following considerations when dealing with threading issues.

• When you create several asynchronous message consumers in the same session, messages are delivered serially by the session thread to these consumers. Sharing a session among several message consumers might starve some consumers of messages while inundating other consumers. If the message rate across these consumers is high enough to cause an imbalance, you might want to separate the consumers into different sessions. To determine whether message flow is unbalanced, you can monitor destinations to see the rate of messages coming in. See Using the Metrics Monitoring API.

• You can reduce the number of threads allocated to the client application by using fewer connections and fewer sessions. However, doing this might slow your application's throughput.

• You might be able to use certain JVM runtime options to improve thread memory usage and performance. For example, if you are running on the Solaris platform, you may be able to run with the same number (or more) threads by using the following vm options with the client: Refer to the JDK documentation for details.

  • Use the Xss128K option to decrease the memory size of the heap.

  • Use the xconcurrentIO option to improve thread performance in the 1.3 VM.

Managing Memory

A client application running in a JVM needs enough memory to accommodate messages that flow in from the network as well as messages the client creates. If your client gets OutOfMemoryError errors, chances are that not enough memory was provided to handle the size or the number of messages being consumed or produced.

Your client might need more than the default JVM heap space. On most systems, the default is 64 MB but you will need to check the default values for your system.

Consider the following guidelines:

• Evaluate the normal and peak system memory footprints when sizing heap space.

• You can start by doubling the heap size using a command like the following:

  java -Xmx128m MyClass
  

• The best size for the heap space depends on both the operating system and the JDK release. Check the JDK documentation for restrictions.

• The size of the VM's memory allocation pool must be less than or equal to the amount of virtual memory that is available on the system.

Managing Message Size

In general, for better manageability, you can break large messages into smaller parts, and use sequencing to ensure that the partial messages sent are concatenated properly. You can also use a Message Queue JMS feature to compress the body of a message. This section describes the programming interface that allows you to compress messages and to compare the size of compressed and uncompressed messages.

Message compression and decompression is handled entirely by the client runtime, without involving the broker. Therefore, applications can use this feature with a pervious version of the broker, but they must use version 3.6 or later of the Message Queue client runtime library.

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");

Managing the Dead Message Queue

When a message is deemed undeliverable, it is automatically placed on a special queue called the dead message queue. A message placed on this queue retains all of its original headers (including its original destination) and information is added to the message's properties to explain why it became a dead message. An administrator or a developer can access this queue, remove a message, and determine why it was placed on the queue.

• For an introduction to dead messages and the dead message queue, see "Using the Dead Message Queue" in Oracle GlassFish Server Message Queue Administration Guide.

• For a description of the destination properties and of the broker properties that control the system's use of the dead message queue, see "Physical Destination Property Reference" in Oracle GlassFish Server Message Queue Administration Guide.

This section describes the message properties that you can set or examine programmatically to determine the following:

• Whether a dead message can be sent to the dead message queue.

• Whether the broker should log information when a message is destroyed or moved to the dead message queue.

• Whether the body of the message should also be stored when the message is placed on the dead message queue.

• Why the message was placed on the dead message queue and any ancillary information.

Message Queue 3.6 clients can set properties related to the dead message queue on messages and send those messages to clients compiled against earlier versions. However clients receiving such messages cannot examine these properties without recompiling against 3.6 libraries.

The dead message queue is automatically created by the broker and called mq.sys.dmq. You can use the message monitoring API, described in Using the Metrics Monitoring API, to determine whether that queue is growing, to examine messages on that queue, and so on.

You can set the properties described in Table 3-2 for any message to control how the broker should handle that message if it deems it to be undeliverable. Note that these message properties are needed only to override destination, or broker-based behavior.

The properties described in Table 3-3 are set by the broker for a message placed in the dead message queue. You can examine the properties for the message to retrieve information about why the message was placed on the queue and to gather other information about the message and about the context within which this action was taken.

Managing Physical Destination Limits

When creating a topic or queue destination, the administrator can specify how the broker should behave when certain memory limits are reached. Specifically, when the number of messages reaching a physical destination exceeds the number specified with the maxNumMsgs property or when the total amount of memory allowed for messages exceeds the number specified with the maxTotalMsgBytes property, the broker takes one of the following actions, depending on the setting of the limitBehavior property:

• Slows message producers (FLOW_CONTROL)

• Throws out the oldest message in memory (REMOVE_OLDEST)

• Throws out the lowest priority message in memory (REMOVE_LOW_PRIORITY)

• Rejects the newest messages (REJECT_NEWEST)

If the default value REJECT_NEWEST is specified for the limitBehavior property, the broker throws out the newest messages received when memory limits are exceeded. If the message discarded is a persistent message, the producing client gets an exception which should be handled by resending the message later.

If any of the other values is selected for the limitBehavior property or if the message is not persistent, the application client is not notified if a message is discarded. Application clients should let the administrator know how they prefer this property to be set for best performance and reliability.

Using the Client Runtime Ping Feature

Message Queue defines a connection factory attribute for a ping interval. This attribute specifies the interval at which the client runtime should check the client's connection to the broker. The ping feature is especially useful to Message Queue clients that exclusively receive messages and might therefore not be aware that the absence of messages is due to a connection failure. This feature could also be useful to producers who don't send messages frequently and who would want notification that a connection they're planning to use is not available.

The connection factory attribute used to specify this interval is called imqPingInterval. Its default value is 30 seconds. A value of -1 or 0, specifies that the client runtime should not check the client connection.

Developers should set (or have the administrator set) ping intervals that are slightly more frequent than they need to send or receive messages, to allow time to recover the connection in case the ping discovers a connection failure. Note also that the ping may not occur at the exact time specified by the value you supply for interval; the underlying operating system's use of i/o buffers may affect the amount of time needed to detect a connection failure and trigger an exception.

A failed ping operation results in a JMSException on the subsequent method call that uses the connection. If an exception listener is registered on the connection, it will be called when a ping operation fails.

Preventing Message Loss for Synchronous Consumers

It is always possible that a message can be lost for synchronous consumers in a session using AUTO_ACKNOWLEDGE mode if the provider fails. To prevent this possibility, you should either use a transacted session or a session in CLIENT_ACKNOWLEDGE mode.

Synchronous Consumption in Distributed Applications

Because distributed applications involve greater processing time, such an application might not behave as expected if it were run locally. For example, calling the receiveNoWait method for a synchronous consumer might return null even when there is a message available to be retrieved.

If a client connects to the broker and immediately calls the receiveNoWait method, it is possible that the message queued for the consuming client is in the process of being transmitted from the broker to the client. The client runtime has no knowledge of what is on the broker, so when it sees that there is no message available on the client's internal queue, it returns with a null, indicating no message.

You can avoid this problem by having your client do either of the following:

• Use one of the synchronous receive methods that specifies a timeout interval.

• Use a queue browser to check the queue before calling the receiveNoWait method.

Delivery Mode (Persistent/Nonpersistent)

Persistent messages guarantee message delivery in case of broker failure. The broker stores these message in a persistent store until all intended consumers acknowledge that they have consumed the message.

Broker processing of persistent messages is slower than for nonpersistent messages for the following reasons:

• A broker must reliably store a persistent message so that it will not be lost should the broker fail.

• The broker must confirm receipt of each persistent message it receives. Delivery to the broker is guaranteed once the method producing the message returns without an exception.

• Depending on the client acknowledgment mode, the broker might need to confirm a consuming client's acknowledgment of a persistent message.

For both queues and topics with durable subscribers, performance was approximately 40% faster for non-persistent messages. We obtained these results using 10K-size messages and AUTO_ACKNOWLEDGE mode.

Use of Transactions

A transaction guarantees that all messages produced in a transacted session and all messages consumed in a transacted session will be either processed or not processed (rolled back) as a unit. Message Queue supports both local and distributed transactions.

A message produced or acknowledged in a transacted session is slower than in a non-transacted session for the following reasons:

• Additional information must be stored with each produced message.

• In some situations, messages in a transaction are stored when normally they would not be. For example, a persistent message delivered to a topic destination with no subscriptions would normally be deleted, however, at the time the transaction is begun, information about subscriptions is not available.

• Information on the consumption and acknowledgment of messages within a transaction must be stored and processed when the transaction is committed.

Acknowledgment Mode

Other than using transactions, you can ensure reliable delivery by having the client acknowledge receiving a message. If a session is closed without the client acknowledging the message or if the message broker fails before the acknowledgment is processed, the broker redelivers that message, setting a JMSRedelivered flag.

For a non-transacted session, the client can choose one of three acknowledgment modes, each of which has its own performance characteristics:

• AUTO_ACKNOWLEDGE. The system automatically acknowledges a message once the consumer has processed it. This mode guarantees at most one redelivered message after a provider failure.

• CLIENT_ACKNOWLEDGE. The application controls the point at which messages are acknowledged. All messages processed in that session since the previous acknowledgment are acknowledged. If the broker fails while processing a set of acknowledgments, one or more messages in that group might be redelivered.

  (Using CLIENT_ACKNOWLEDGE mode is similar to using transactions, except there is no guarantee that all acknowledgments will be processed together if a provider fails during processing.)

• DUPS_OK_ACKNOWLEDGE. This mode instructs the system to acknowledge messages in a lazy manner. Multiple messages can be redelivered after a provider failure.

• NO_ACKNOWLEDGE In this mode, the broker considers a message acknowledged as soon as it has been written to the client. The broker does not wait for an acknowledgment from the receiving client. This mode is best used by typic subscribers who are not worried about reliability.

Performance is impacted by acknowledgment mode for the following reasons:

• Extra control messages between broker and client are required in AUTO_ACKNOWLEDGE and CLIENT_ACKNOWLEDGE modes. The additional control messages add processing overhead and can interfere with JMS payload messages, causing processing delays.

• In AUTO_ACKNOWLEDGE and CLIENT_ACKNOWLEDGE modes, the client must wait until the broker confirms that it has processed the client's acknowledgment before the client can consume more messages. (This broker confirmation guarantees that the broker will not inadvertently redeliver these messages.)

• The Message Queue persistent store must be updated with the acknowledgment information for all persistent messages received by consumers, thereby decreasing performance.

Durable vs. Nondurable Subscriptions

Subscribers to a topic destination have either durable and nondurable subscriptions. Durable subscriptions provide increased reliability at the cost of slower throughput for the following reasons:

• The Message Queue message broker must persistently store the list of messages assigned to each durable subscription so that should the broker fail, the list is available after recovery.

• Persistent messages for durable subscriptions are stored persistently, so that should a broker fail, the messages can still be delivered after recovery, when the corresponding consumer becomes active. By contrast, persistent messages for nondurable subscriptions are not stored persistently (should a broker fail, the corresponding consumer connection is lost and the message would never be delivered).

We compared performance for durable and non-durable subscribers in two cases: persistent and nonpersistent 10k-sized messages. Both cases use AUTO_ACKNOWLEDGE acknowledgment mode. We found a performance impact only in the case of persistent messages, which slowed messages conveyed to durable subscribers by about 30%.

Use of Selectors (Message Filtering)

Application developers can have the messaging provider sort messages according to criteria specified in the message selector associated with a consumer and deliver to that consumer only those messages whose property value matches the message selector. For example, if an application creates a subscriber to the topic WidgetOrders and specifies the expression NumberOfOrders>1000 for the message selector, messages with a NumberOfOrders property value of 1001 or more are delivered to that subscriber.

Creating consumers with selectors lowers performance (as compared to using multiple destinations) because additional processing is required to handle each message. When a selector is used, it must be parsed so that it can be matched against future messages. Additionally, the message properties of each message must be retrieved and compared against the selector as each message is routed. However, using selectors provides more flexibility in a messaging application and may lower resource requirements at the expense of speed.

Connection Events

The following table lists and describes the events that can be returned by the event listener.

Note that the JMS exception listener is not called when a connection event occurs. The exception listener is only called if the client runtime has exhausted its reconnection attempts. The client runtime always calls the event listener before the exception listener.

Creating an Event Listener

The following code example illustrates how you set a connection event listener. Whenever a connection event occurs, the event listener's onEvent method will be invoked by the client runtime.

//create an MQ connection factory.

com.sun.messaging.ConnectionFactory factory =
    new com.sun.messaging.ConnectionFactory();

//create an MQ connection.

com.sun.messaging.jms.Connection connection = 
    (com.sun.messaging.jms.Connection )factory.createConnection();

//construct an MQ event listener.  The listener implements 
//com.sun.messaging.jms.notification.EventListener interface.

com.sun.messaging.jms.notification.EventListener eListener = 
    new ApplicationEventListener();

//set event listener to the MQ connection.

connection.setEventListener ( eListener );

Event Listener Examples

In this example, an application chooses to have its event listener log the connection event to the application's logging system.

public class ApplicationEventListener implements
    com.sun.messaging.jms.notification.EventListener {

  public void onEvent ( com.sun.messaging.jms.notification.Event connEvent ) {
    log (connEvent);
  }
  private void log ( com.sun.messaging.jms.notification.Event connEvent ) {
    String eventCode = connEvent.getEventCode(); 
    String eventMessage = connEvent.getEventMessage();
    //write event information to the output stream.
  }
}

Consumer Events

The following table lists and describes the events that can be returned by the event listener.

Creating a Consumer Event Listener

The following code example illustrates how you set and remove a consumer event listener. Whenever a consumer event occurs, the event listener's onEvent method will be invoked by the client runtime.

//create an MQ connection factory.

com.sun.messaging.ConnectionFactory factory =
    new com.sun.messaging.ConnectionFactory();

//create an MQ connection.

com.sun.messaging.jms.Connection connection = 
    (com.sun.messaging.jms.Connection)factory.createConnection();

//create an MQ session

com.sun.messaging.jms.Session session = 
    (com.sun.messaging.jms.Session)connection.createSession(false,
    Session.AUTO_ACKNOWLEDGE);

//create a queue

com.sun.messaging.Queue queue =
    (com.sun.messaging.Queue)session.createQueue(strQueueName);

//construct an MQ event listener.  The listener implements 
//com.sun.messaging.jms.notification.EventListener interface.

com.sun.messaging.jms.notification.EventListener consEvtListener = 
    new MyConsumerEventListener();

//set consumer event listener.

connection.setConsumerEventListener
    ( (com.sun.messaging.Destination)queue, consEvtListener );

//perform activities

//remove consumer event listener.

connection.removeConsumerEventListener
    ( (com.sun.messaging.Destination)queue );

Consumer Event Listener Examples

In this example, an application chooses to have its event listener set a boolean flag to give ongoing consumer availability information.

public class MyEventListener implements
    com.sun.messaging.jms.notification.EventListener {

  boolean consumerReady = false;

  MyEventListener(){
    consumerReady = false;
  }

  public void onEvent(com.sun.messaging.jms.notification.Event evt) {

    if (evt.getEventCode().equals(
        com.sun.messaging.jms.notification.ConsumerEvent.CONSUMER_NOT_READY
        )) {
      synchronized(this){
        consumerReady=false;
      }
    } else if (evt.getEventCode().equals(
        com.sun.messaging.jms.notification.ConsumerEvent.CONSUMER_READY
        )) {
      synchronized(this){
        consumerReady=true;
      }
    }
  }
}

Enabling Auto-Reconnect

If you are using conventional clusters, you enable automatic reconnection by setting the connection factory imqReconnectEnabled attribute to true. If you are using a high availability cluster, the imqReconnectEnabled attribute is ignored; the client runtime will automatically reconnect to a backup broker if the connection is lost and not regained after no more than imqReconnectAttempts attempts. This applies to all deployment configurations: whether Message Queue is used stand alone or whether the connection is created through a resource adapter.

No matter which type of cluster you are using, you must also configure the connection factory administered object to specify the following information.

• A list of message-service addresses (using the imqAddressList attribute). Independently of the clustering model used, the client runtime uses this address list when it establishes the initial connection.

  When you connect to a conventional cluster, the client runtime also uses the address list when it tries to reestablish a connection to the message service: it attempts to connect to the brokers in the list until it finds (or fails to find) an available broker. If you specify only a single broker instance on the imqAddressList attribute, the configuration won't support recovery from hardware failure.

  When you specify more than one broker, you can decide whether to use parallel brokers or a broker cluster. In a parallel configuration, there is no communication between brokers, while in a broker cluster, the brokers interact to distribute message delivery loads. (Refer to "Cluster Message Delivery" in Oracle GlassFish Server Message Queue Technical Overview for more information on broker clusters.)

  • To enable parallel-broker reconnection, set theimqAddressListBehavior attribute to PRIORITY . Typically, you would specify no more than a pair of brokers for this type of reconnection. This way, the messages are published to one broker, and all clients fail over together from the first broker to the second.

  • To enable clustered-broker reconnection, set the imqAddressListBehavior attribute to RANDOM. This way, the client runtime randomizes connection attempts across the list, and client connections are distributed evenly across the broker cluster.

    Each broker in a cluster uses its own separate persistent store (which means that undelivered persistent messages are unavailable until a failed broker is back online). If one broker crashes, its client connections are reestablished on other brokers.

    If you use the high availability clustering model, the address list is dynamically updated to include the brokers that are connected to the highly available database serving the cluster. In this case, the client runtime and the brokers use an internal protocol to determine which broker takes over the persistent data of the failed broker. Therefore the imqAddressListBehavior property does not apply to this model.

• The number of iterations to be made over the list of brokers (using the imqAddressListIterations attribute) when attempting to create a connection or to reconnect.

  For high-availability clusters, the broker will attempt to reconnect forever (no matter what value you specify for this attribute). If the client does not want this behavior, it must explicitly close the connection.

• The number of attempts to reconnect to a broker if the first connection fails (using the imqReconnectAttempts attribute).

• The interval, in milliseconds, between reconnect attempts, using the imqReconnectInterval attribute. This attribute applies to both clustering models.

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"

Auto-Reconnect Behaviors

A broker treats an automatic reconnection as it would a new connection. When the original connection is lost, all resources associated with that connection are released. For example, in a broker cluster, as soon as one broker fails, the other brokers assume that the client connections associated with the failed broker are gone. After auto-reconnect takes place, the client connections are recreated from scratch.

Sometimes the client-side state cannot be fully restored by auto-reconnect. Perhaps a resource that the client needs cannot be recreated. In this case, the client runtime calls the client's connection exception handler and the client must take appropriate action to restore state. For additional information, see Handling Exceptions When Failover Occurs.

If the client is automatically-reconnected to a different broker instance, effects vary depending on the clustering model used.

• In a conventional cluster, persistent messages produced but not yet consumed may only be delivered to the consumer after the original broker recovers. Other state information held by the failed or disconnected broker can be lost. The messages held by the original broker, once it is restored, might be delivered out of order.

• In a high availability cluster, messages produced but not yet consumed continue to be delivered to the consumer without the original broker needing to recover.

A transacted session is the most reliable method of ensuring that a message isn't lost if you are careful in coding the transaction. If auto-reconnect happens in the middle of a transaction, any attempt to produce or consume messages will cause the client runtime to throw a JMSException. In this case, applications must call Session.rollback() to roll back the transaction.

The Message Queue client runtime may throw a TransactionRolledBackException when Session.commit() is called during or after a failover occurs. In this case, the transaction is rolled back and a new transaction is automatically started. Applications are not required to call Session.rollback() to rollback the transaction after receiving a TransactionRolledBackException.

The Message Queue client runtime may throw a JMSException when Session.commit() is called during or after a failover occurs. In this case, the transaction state is unknown (may or may not be committed). Applications should call Session.rollback() to roll back the uncommitted transaction.

If you are using a high availability cluster, the only time your transaction might wind up in an unknown state is if it is not possible to reconnect to any brokers in the cluster. This should happen rarely if ever. For additional information, see Handling Exceptions When Failover Occurs.

Automatic reconnection affects producers and consumers differently:

• During reconnection, producers cannot send messages. The production of messages (or any operation that involves communication with the message broker) is blocked until the connection is reestablished.

• For consumers, automatic reconnection is supported for all client acknowledgment modes. After a connection is reestablished, the broker will redeliver all unacknowledged messages it had previously delivered, marking them with a Redeliver flag. The client can examine this flag to determine whether any message has already been consumed (but not yet acknowledged). In the case of nondurable subscribers, some messages might be lost because the broker does not hold their messages once their connections have been closed. Any messages produced for nondurable subscribers while the connection is down cannot be delivered when the connections is reestablished. For additional information, see Handling Exceptions When Failover Occurs.

Auto-Reconnect Limitations

Notice the following points when using the auto-reconnect feature:

• Messages might be redelivered to a consumer after auto-reconnect takes place. In auto-acknowledge mode, you will get no more than one redelivered message. In other session types, all unacknowledged persistent messages are redelivered.

• While the client runtime is trying to reconnect, any messages sent by the broker to nondurable topic consumers are lost.

• Any messages that are in queue destinations and that are unacknowledged when a connection fails are redelivered after auto-reconnect. However, in the case of queues delivering to multiple consumers, these messages cannot be guaranteed to be redelivered to the original consumers. That is, as soon as a connection fails, an unacknowledged queue message might be rerouted to other connected consumers.

• In the case of a conventional broker cluster, the failure of the master broker prevents the following operations from succeeding on any other broker in the cluster:

  • Creating or destroying a new durable subscription.

  • Creating or destroying a new physical destination using the imqcmd create dst command.

  • Starting a new broker process. (However, the brokers that are already running continue to function normally even if the master broker goes down.)

    You can configure the master broker to restart automatically using Message Queue broker support for rc scripts or the Windows service manager.

• Auto-reconnect doesn't work if the client uses a ConnectionConsumer to consume messages. In that case, the client runtime throws an exception.

Handling Exceptions When Failover Occurs

Several kinds of exceptions can occur as a result of the client being reconnected after a failover. How the client application should handle these exceptions depends on whether a session is transacted, on the kind of exception thrown, and on the client's role--as producer or consumer. The following sections discuss the implications of these factors.

Independently of how the exception is raised, the client application must never call System.exit()to exit the application because this would prevent the Message Queue client runtime from reconnecting to an alternate or restarted broker.

When a failover occurs, exception messages may be shown on the application's console and recorded in the broker's log. These messages are for information only. They may be useful in troubleshooting, but minimizing or eliminating the impact of a failover is best handled preemptively by the application client in the ways described in the following sections.

/*
 * @(#)FailoverQSender.java    1.2 07/04/20
 *
 * Copyright 2000 Sun Microsystems, Inc. All Rights Reserved
 * SUN PROPRIETARY/CONFIDENTIAL
 * Use is subject to license terms.
 *
 */
package test.jmsclient.ha;

import java.util.Date;
import javax.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 2000 Sun Microsystems, Inc. All Rights Reserved
 * SUN PROPRIETARY/CONFIDENTIAL
 * Use is subject to license terms.
  */
package test.jmsclient.ha;

import java.util.Date;
import java.util.Vector;
import javax.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 2006 Sun Microsystems, Inc. All Rights Reserved
 * SUN PROPRIETARY/CONFIDENTIAL
 * Use is subject to license terms. */

package dura.example;

import javax.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 2006 Sun Microsystems, Inc. All Rights Reserved
 * SUN PROPRIETARY/CONFIDENTIAL
 * Use is subject to license terms.
 *
 */
package dura.example;

import java.util.Date;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Connection;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.Topic;
import javax.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();
    }

}

Using Client Acknowledge Mode

For more flexibility, Message Queue lets you customize the JMS client-acknowledge mode. In client-acknowledge mode, the client explicitly acknowledges message consumption by invoking the acknowledge() method of a message object. The standard behavior of this method is to cause the session to acknowledge all messages that have been consumed by any consumer in the session since the last time the method was invoked. (That is, the session acknowledges the current message and all previously unacknowledged messages, regardless of who consumed them.)

In addition to the standard behavior specified by JMS, Message Queue lets you use client-acknowledge mode to acknowledge one message at a time.

Observe the following rules when implementing custom client acknowledgment:

• To acknowledge an individual message, call the acknowledgeThisMessage() method. To acknowledge all messages consumed so far, call the acknowledgeUpThroughThisMessage() method. Both are shown in the following code example.

  public interface com.sun.messaging.jms.Message {
            void acknowledgeThisMessage() throws JMSException;
            void acknowledgeUpThroughThisMessage() throws JMSException;
  }
  

• When you compile the resulting code, include both imq.jar and jms.jar in the class path.

• Don't call acknowledge(), acknowledgeThisMessage() , or acknowledgeUpThroughThisMessage() in any session except one that uses client-acknowledge mode. Otherwise, the method call is ignored.

• Don't use custom acknowledgment in transacted sessions. A transacted session defines a specific way to have messages acknowledged.

If a broker fails, any message that was not acknowledged successfully (that is, any message whose acknowledgment ended in a JMSException) is held by the broker for delivery to subsequent clients.

Example 3-6 demonstrates both types of custom client acknowledgment.

...
import javax.jms.*;
...[Look up a connection factory and create a connection.]

    Session session = connection.createSession(false,
                       Session.CLIENT_ACKNOWLEDGE);

...[Create a consumer and receive messages.]

    Message message1 = consumer.receive();
    Message message2 = consumer.receive();
    Message message3 = consumer.receive();

...[Process messages.]

...[Acknowledge one individual message.
   Notice that the following acknowledges only message 2.]

    ((com.sun.messaging.jms.Message)message2).acknowledgeThisMessage();

...[Continue. Receive and process more messages.]

    Message message4 = consumer.receive();
    Message message5 = consumer.receive();
    Message message6 = consumer.receive();

...[Acknowledge all messages up through message 4. Notice that this 
    acknowledges messages 1, 3, and 4, because message 2 was acknowledged  
    earlier.]

    ((com.sun.messaging.jms.Message)message4).acknowledgeUpThroughThisMessage();
...[Continue. Finally, acknowledge all messages consumed in the session. 
    Notice that this acknowledges all remaining consumed messages, that is,  
    messages 5 and 6, because this is the standard behavior of the JMS API.] 

    message5.acknowledge();

Using No Acknowledge Mode

No-acknowledge mode is a nonstandard extension to the JMS API. Normally, the broker waits for a client acknowledgment before considering that a message has been acknowledged and discarding it. That acknowledgment must be made programmatically if the client has specified client-acknowledge mode or it can be made automatically, by the session, if the client has specified auto-acknowledge or dups-OK-acknowledge. If a consuming client specifies no-acknowledge mode, the broker discards the message as soon as it has sent it to the consuming client. This feature is intended for use by nondurable subscribers consuming nonpersistent messages, but it can be used by any consumer.

Using this feature improves performance by reducing protocol traffic and broker work involved in acknowledging a message. This feature can also improve performance for brokers dealing with misbehaving clients who do not acknowledge messages and therefore tie down broker memory resources unnecessarily. Using this mode has no effect on producers.

You use this feature by specifying NO_ACKNOWLEDGE for the acknowledgeMode parameter to the createSession, createQueueSession, or createTopicSession method. No-acknowledge mode must be used only with the connection methods defined in the com.sun.messaging.jms package. Note however that the connection itself must be created using the javax.jms package.

The following are sample variable declarations for connection, queueConnection and topicConnection:

javax.jms.connection Connection;
javax.jms.queueConnection queueConnection
javax.jms.topicConnection topicConnection

The following are sample statements to create different kinds of no-acknowledge sessions:

//to create a no ack session
Session noAckSession  =
     ((com.sun.messaging.jms.Connection)connection)
    .createSession(com.sun.messaging.jms.Session.NO_ACKNOWLEDGE);

// to create a no ack topic session
TopicSession noAckTopicSession  =
     ((com.sun.messaging.jms.TopicConnection) topicConnection)
    .createTopicSession(com.sun.messaging.jms.Session.NO_ACKNOWLEDGE);

//to create a no ack queue session
QueueSession noAckQueueSession  =
     ((com.sun.messaging.jms.QueueConnection) queueConnection)
    .createQueueSession(com.sun.messaging.jms.Session.NO_ACKNOWLEDGE);

Specifying no-acknowledge mode for a session results in the following behavior:

• The client runtime will throw a JMSException if Session.recover() is called.

• The client runtime will ignore a call to the Message.acknowledge() method from a consumer.

• Messages can be lost. As opposed to dups-OK-acknowledge, which can result in duplicate messages being sent, no-acknowledge mode bypasses checks and balances built into the system and may result in message loss.

Logging Name Spaces, Levels, and Activities

The Message Queue provider defines a set of logging name spaces associated with logging levels and logging activities that allow Message Queue clients to log connection and session events when a logging configuration is appropriately set.

The root logging name space for the Message Queue client runtime is defined as javax.jms. All loggers in the Message Queue client runtime use this name as the parent name space.

The logging levels used for the Message Queue client runtime are the same as those defined in the java.util.logging.Level class. This class defines seven standard log levels and two additional settings that you can use to turn logging on and off.

OFF

Turns off logging.

SEVERE

Highest priority, highest value. Application-defined.

WARNING

Application-defined.

INFO

Application-defined.

CONFIG

Application-defined

FINE

Application-defined.

FINER

Application-defined.

FINEST

Lowest priority, lowest value. Application-defined.

ALL

Enables logging of all messages.

In general, exceptions and errors that occur in the Message Queue client runtime are logged by the logger with the javax.jms name space.

• Exceptions thrown from the JVM and caught by the client runtime, such as IOException, are logged by the logger with the logging name space javax.jms at level WARNING.

• JMS exceptions thrown from the client runtime, such as IllegalStateException, are logged by the logger with the logging name space javax.jms at level FINER.

• Errors thrown from the JVM and caught by the client runtime, such as OutOfMemoryError, are logged by the logger with the logging name space javax.jms at level SEVERE.

The following tables list the events that can be logged and the log level that must be set to log events for JMS connections and for sessions.

The following table describes log levels and events for connections.

For sessions, the following information is recorded in the log record.

• Each log record for a message delivered to a consumer includes ConnectionID, SessionID, and ConsumerID.

• Each log record for a message sent by a producer includes ConnectionID, SessionID, ProducerID, and destination name.

The table below describes log levels and events for sessions.

By default, the output log level is inherited from the JRE in which the application is running. Check the JRE_DIRECTORY/lib/logging.properties file to determine what that level is.

You can configure logging programmatically or by using configuration files, and you can control the scope within which logging takes place. The following sections describe these possibilities.

Using the JRE Logging Configuration File

The following example shows how you set logging name spaces and levels in the JRE_DIRECTORY/lib/logging.properties file, which is used to set the log level for the Java runtime environment. All applications using this JRE will have the same logging configuration. The sample configuration below sets the logging level to INFO for the javax.jms.connection name space and specifies that output be written to java.util.logging.ConsoleHandler.

#logging.properties file.
# "handlers" specifies a comma separated list of log Handler 
# classes. These handlers will be installed during VM startup.
# Note that these classes must be on the system classpath.
# By default we only configure a ConsoleHandler, which will only
# show messages at the INFO and above levels.

    handlers= java.util.logging.ConsoleHandler

# Default global logging level.
# This specifies which kinds of events are logged across
# all loggers. For any given facility this global level
# can be overriden by a facility-specific level.
# Note that the ConsoleHandler also has a separate level
# setting to limit messages printed to the console.

    .level= INFO

# Limit the messages that are printed on the console to INFO and above.

    java.util.logging.ConsoleHandler.level = INFO
    java.util.logging.ConsoleHandler.formatter = 
                                    java.util.logging.SimpleFormatter

# The logger with javax.jms.connection name space will write
# Level.INFO messages to its output handler(s). In this configuration 
# the ouput handler is set to java.util.logging.ConsoleHandler.

    javax.jms.connection.level = INFO

Using a Logging Configuration File for a Specific Application

You can also define a logging configuration file from the java command line that you use to run an application. The application will use the configuration defined in the specified logging file. In the following example, configFile uses the same format as defined in the JRE_DIRECTORY/lib/logging.properties file.

java -Djava.util.logging.config.file=configFile MQApplication

Setting the Logging Configuration Programmatically

The following code uses the java.util.logging API to log connection events by changing the javax.jms.connection name space log level to FINE. You can include such code in your application to set logging configuration programmatically.

import java.util.logging.*;
//construct a file handler and output to the mq.log file 
//in the system's temp directory.

    Handler fh = new FileHandler("%t/mq.log");
    fh.setLevel (Level.FINE);

//Get Logger for "javax.jms.connection" domain.

    Logger logger = Logger.getLogger("javax.jms.connection");
    logger.addHandler (fh);

//javax.jms.connection logger would log activities   
//with level FINE and above.

    logger.setLevel (Level.FINE);