Application server sample code

 

  1. The ConnectionConsumers get message references from the queue.
  2. Each ConnectionConsumer selects specific message references.
  3. The ConnectionConsumer buffer holds the selected message references.
  4. The ConnectionConsumer requests one or more ServerSessions from the ServerSessionPool.
  5. ServerSessions are allocated from the ServerSessionPool.
  6. The ConnectionConsumer assigns message references to the ServerSessions and starts the ServerSession threads running.
  7. Each ServerSession retrieves its referenced messages from the queue. It passes them to the onMessage method from the MessageListener that is associated with the JMS Session.
  8. After it completes its processing, the ServerSession is returned to the pool.

Normally, the appserver supplies ServerSessionPool and ServerSession functionality. However, WebSphere MQ JMS is supplied with a simple implementation of these interfaces, with program source. These samples are in the following directory, where <install_dir> is the installation directory for WebSphere MQ JMS: 

<install_dir>/samples/jms/asf

These samples enable you to use the WebSphere MQ JMS ASF in a standalone environment (that is, you do not need a suitable appserver). Also, they provide examples of how to implement these interfaces and take advantage of the WebSphere MQ JMS ASF. These examples are intended to aid both WebSphere MQ JMS users, and vendors of other application servers.

 

MyServerSession.java

This class implements the javax.jms.ServerSession interface. It associates a thread with a JMS session. Instances of this class are pooled by a ServerSessionPool (see MyServerSessionPool.java). As a ServerSession, it must implement the following two methods:

  • getSession(), which returns the JMS Session associated with this ServerSession
  • start(), which starts this ServerSession's thread and results in the JMS Session's run() method being invoked

MyServerSession also implements the Runnable interface. Therefore, the creation of the ServerSession's thread can be based on this class, and does not need a separate class.

The class uses a wait()-notify() mechanism that is based on the values of two boolean flags, ready and quit. This mechanism means that the ServerSession creates and starts its associated thread during its construction. However, it does not automatically execute the body of the run() method. The body of the run() method is executed only when the ready flag is set to true by the start() method. The ASF calls the start() method when it is necessary to deliver messages to the associated JMS session.

For delivery, the run() method of the JMS session is called. The WebSphere MQ JMS ASF will have already loaded the run() method with messages.

After delivery completes, the ready flag is reset to false, and the owning ServerSessionPool is notified that delivery is complete. The ServerSession then remains in a wait state until either the start() method is called again, or the close() method is invoked and ends this ServerSession's thread.

 

MyServerSessionPool.java

This class implements the javax.jms.ServerSessionPool interface, creating and controlling access to a pool of ServerSessions.

In this implementation, the pool consists of a static array of ServerSession objects that are created during the construction of the pool. The following four parameters are passed into the constructor:

  • javax.jms.Connection connection

    The connection used to create JMS sessions.

  • int capacity

    The size of the array of MyServerSession objects.

  • int ackMode

    The required acknowledge mode of the JMS sessions.

  • MessageListenerFactory mlf

    The MesssageListenerFactory that creates the message listener that is supplied to the JMS sessions. See MessageListenerFactory.java.

The pool's constructor uses these parameters to create an array of MyServerSession objects. The supplied connection is used to create JMS sessions of the given acknowledge mode and correct domain (QueueSessions for point-to-point and TopicSessions for publish/subscribe). The sessions are supplied with a message listener. Finally, the ServerSession objects, based on the JMS sessions, are created.

This sample implementation is a static model. That is, all the ServerSessions in the pool are created when the pool is created, and after this the pool cannot grow or shrink. This approach is just for simplicity. It is possible for a ServerSessionPool to use a sophisticated algorithm to create ServerSessions dynamically, as needed.

MyServerSessionPool keeps a record of which ServerSessions are currently in use by maintaining an array of boolean values called inUse. These booleans are all initialized to false. When the getServerSession method is invoked and requests a ServerSession from the pool, the inUse array is searched for the first false value. When one is found, the boolean is set to true and the corresponding ServerSession is returned. If there are no false values in the inUse array, the getServerSession method must wait() until notification occurs.

Notification occurs in either of the following circumstances:

  • The pool's close() method is called, indicating that the pool must be shut down.

  • A ServerSession that is currently in use completes its workload and calls the serverSessionFinished method. The serverSessionFinished method returns the ServerSession to the pool, and sets the corresponding inUse flag to false. The ServerSession then becomes eligible for reuse.

 

MessageListenerFactory.java

In this sample, a message listener factory object is associated with each ServerSessionPool instance. The MessageListenerFactory class represents a very simple interface that is used to obtain an instance of a class that implements the javax.jms.MessageListener interface. The class contains a single method: 

 javax.jms.MessageListener createMessageListener();

An implementation of this interface is supplied when the ServerSessionPool is constructed. This object is used to create message listeners for the individual JMS sessions that back up the ServerSessions in the pool. This architecture means that each separate implementation of the MessageListenerFactory interface must have its own ServerSessionPool.

WebSphere MQ JMS includes a sample MessageListenerFactory implementation, which is discussed in CountingMessageListenerFactory.java.

 

WebSphere is a trademark of the IBM Corporation in the United States, other countries, or both.

 

IBM is a trademark of the IBM Corporation in the United States, other countries, or both.