The real power of JMS is derived from the fact that the communication format is a message, which provides a level of abstraction that enables you to separate the details about the destination system from the source application code. For example, using JMS, a component can send a message to a destination, and the recipient can retrieve it from that destination. However, the sender and receiver do not have to be available at the same time in order to communicate, or need to know anything about each other. They both only need to know what message format is being used to communicate and what destination to use to send and receive the message. In this respect, messaging differs from tightly coupled technologies such as Remote Method Invocation (RMI) or Remote Procedure Calls (RPC), which both require an application to know a remote application’s interface (methods).

There are many scenarios where JMS messaging is a preferred communication means over a tightly coupled technology API:

You will find the concept of messaging very prevalent in manufacturing systems, financial services applications, health services applications, auctioning and community-based trading, workflow systems, supply-chain systems, and many others that need to leverage loosely coupled and asynchronous communication.

Also, with the industrywide adoption of XML, there are several organizations and industry consortia working on developing common XML DTDs and schemas for e-commerce interchange. In fact, BEA WebLogic Integration (WLI) includes B2Bi and Business Process Management (BPM) functionality based largely on XML delivered over JMS. To learn more about the features and capabilities of WebLogic Integration, see Chapter 33, “Introducing WebLogic Integration (WLI).”

Messaging systems are divided into two categories or domains. These are the Point-to-Point (PTP) and Publish/Subscribe (Pub/Sub) domains. The JMS specification does not require JMS providers to support both messaging domains. However, WebLogic Server, and most competing products, implement both.

The Point-to-Point Messaging Domain

The defining concept in the point to point domain is the message queue. Producers send messages to a queue, which serializes the messages into a linear order. Hence, when consumers receive messages from the queue, the messages are taken in a first-in, first-out (FIFO) order, as illustrated in Figure 15.2. Every message in the queue is associated with a single producer and more importantly, a single consumer. However, multiple producers and consumers can share a single message queue. Queues have the added advantage of holding messages until they are consumed. Many applications use message queues as a way of deferring certain non-essential or time-consuming tasks (like logging) to be handled when resources are available.

Figure 15.2. The characteristics of a point-to-point messaging domain.

The JMS API does not define how queues are created or managed. In WebLogic Server, developers and administrators use the WebLogic Administration Console to create, manage, and configure message queues. Administrators will normally create and configure queues and developers will write clients that access them.

The Publish/Subscribe Messaging Domain

In the Publish/Subscribe domain the topic is the analog of the queue. Message producers publish messages to a topic and multiple consumers (called subscribers) receive messages from the topic, as illustrated in Figure 15.3.

Figure 15.3. The characteristics of the publish/subscribe messaging domain.

Note

Having publishers publish to a topic instead of a list of subscribers—and the anonymity between publishers and subscribers—decouples the publishers and subscribers.

Topics retain messages for as long as it takes to distribute them. However, there is a timing dependency between publishers and subscribers, where a topic can only consume messages after a client has created a subscription to that topic, and the subscriber must be active in order for it to consume messages. The JMS API relaxes this timing dependency by enabling clients to create durable subscribers, which guarantees messages will be delivered to them, even if the subscribers are inactive for a period. When a disconnected durable subscriber comes back online, the topic delivers all the waiting messages to that durable subscriber. Hence, when all messages for a topic must be received, a durable subscriber should be used. If missed messages are tolerable, non-durable subscribers should be used.

Note

Support for durable subscribers is applicable only to the Pub/Sub messaging domain.

A stock ticker application is an example of a scenario where pub/sub messaging domain would be applicable, as a topic can be created for each stock. When the price of the stock changes, the new price is reflected in the topic. The MOM would at this point be responsible for broadcasting the stock price (message) to the connected subscribers to that specific topic (stock).

In some non–JMS-based broadcast messaging systems, subscribers receive all messages within the system. The client then has to implement some sort of filtering solution. In fact it is often the case that every client must filter out important messages from noise. The problem with such a system is twofold. First, every client must do this filtering. Second, the message server has to deliver messages to clients who will eventually filter them out. This wasted effort has tremendous performance implications.

JMS enables message consumers to register a message selector with the JMS provider, which defines filtering logic indicating which messages will be delivered to the consuming client. Importantly, this filtering is performed by the JMS provider and not by the client. Therefore the server does not have to send any unnecessary messages.

Note

Message selectors are not specific to the publish/subscribe domain.

A header contains values used by both clients and providers to identify and route messages, for example the destination, a reply-to destination, the message type, and the message expiration time. All messages support the same standard set of header fields as described in Table 15.2.

These fields contain a set of application-defined name/value pairs that are accessible via the getter and setter methods on the Message object.

Although message property fields may be used for application-specific purposes, JMS provides them primarily for use in conjunction with message selectors for filtering messages by the consumer.

A JMS message body contains the content being delivered from a producer to a consumer. JMS provides five types of message bodies that extend javax.jms.Message, which consists of message headers and properties, but no message body. The message types supported by WebLogic JMS are described in Table 15.3.

A JMS Connection Factory is an abstraction that hides provider-specific connection configuration information from clients and enables JMS applications to create a Connection. Typically, it is the responsibility of the WebLogic Administrator to define and configure one or more uniquely named connection factories via the WebLogic Administration Console. Upon startup WebLogic Server binds the connection factories to the WebLogic JNDI namespace, which enables JMS clients the capability to look up a connection factory.

WebLogic JMS defines one default connection factory that can be looked up using the JNDI name, weblogic.jms.ConnectionFactory. You only need to define a connection factory if this default provided by WebLogic JMS is not suitable for your application.

The ConnectionFactory class does not itself define any messaging methods. For this reason its subclasses define methods for the respective messaging models (PTP and Pub/Sub) as follows:

A connection object encapsulates a virtual connection between an application and the JMS provider, and is used to create a Session for producing and consuming messages. Like connection factories, connections come in two forms—the QueueConnection (PTP) or the TopicConnection (Pub/Sub) interface. For example, after you have a QueueConnectionFactory or TopicConnectionFactory object, you can use it to create a connection.

Clients use a Session to send or receive messages. Sessions are single-threaded. The Session interfaces (QueueSession and TopicSession) provide methods to create MessageProducers and MessageConsumers. Sessions order the messages sent to them and also persist messages until they are delivered. In a non-transacted session, the application creating the session can select one of the following acknowledgement modes for its messages:

In a transacted session, only one transaction is active at any given time. Any messages sent or received during a transaction are treated as an atomic unit and the acknowledgement mode is ignored. However, when a transaction commits, all the messages that the application received during the transaction are acknowledged by the messaging system. If an application rolls back a transaction, the messages that the application received during the transaction are not acknowledged and messages it sent are discarded.

Because sessions represent resources, like MessageServer connections, it is important to invoke close() on a connection to ensure that critical resources are released before the garbage collector frees the associated object.

A destination is the object a JMS client uses to specify the target of the messages it produces and the source of the messages it consumes. In a PTP messaging domain, destinations are called Queues, and in the Pub/Sub messaging domain, destinations are called Topics.

Similar to a connection factory, a WebLogic administrator defines and configures the destination and WebLogic Server adds it to the WebLogic JNDI namespace during startup.

The JMS API also enables you to create destinations (TemporaryQueue and TemporaryTopic objects) that last only for the duration of the connection in which they are created. Temporary destinations are private to the application in which they are created. Any session created by the same connection as the temporary destination can create consumers for it. However, Sessions created by other connections cannot. If you close the connection that the temporary destination belongs to, the destination is closed and its contents are lost. If you create a temporary destination and specify it as the value of the JMSReplyTo message header field, the consumer of the message can use the JMSReplyTo field as the destination to which it sends a reply and can also reference the original request by setting the JMSCorrelationID header field of the reply message to the value of the JMSMessageID header field of the request.

JMS Queues and Topics extend javax.jms.Destination with the following Destination subclasses:

MessageProducer is an interface that encapsulates the functionality necessary to send messages. There are two specialized versions of MessageProducers for the two messaging domains.

The QueueSender interface extends MessageProducer. PTP clients use QueueSender to send messages. When a client creates a QueueSender it is bound to a queue. The interface also has several overloaded send() methods. TopicPublisher is the Pub/Sub version of MessageProducer.

Clients use the MessageConsumer interface to receive messages. PTP clients use the QueueReceiver interface and Pub/Sub clients use topic subscriber.

Because JMS is asynchronous, the MessageConsumer interface provides asynchronous and synchronous receive methods to give developers a degree of control over the way messages are consumed. Table 15.4 describes the four ways to receive a message in JMS.

If your WebLogic JMS server will be supporting durable subscriptions by using the JMS Paging feature to free memory or you want in-flight messages to persist across server restarts, messages stored in memory must be written out to persistent storage. To enable persistent messaging you will need to perform the following tasks:

However, before configuring JMS stores, it is important you take the following into consideration:

For storing persistent messages in a disk-based file, you must create a JMS file store by following these steps:

If you intend to leverage the JMS Paging feature, you must also repeat the above steps to configure an additional “paging” JMS file store for temporarily swapping message bodies out to disk when WebLogic Server’s memory allocation is exhausted. Hence, you could configure two file stores such as JMSFileStore and JMSPagingStore to take into account message persistence as well as paging.

WebLogic Server creates the JMS file store files in their specified directories once the file store has been assigned to an active JMS server that has queues or topics assigned to it receiving messages from a message producer.

For storing persistent messages in a JDBC-accessible database, you must create a JMS JDBC store using the following steps:

WebLogic Server will automatically create the JMS tables in the target database after the JMS JDBC Store has been assigned to a JMS server and WebLogic Server rebooted. During WebLogic Server’s initialization process, WebLogic Server uses the assigned connection pool to connect to the target database and create the JMSSTATE and JMSSTORE tables with the appropriate prefixes in the default schema defined for the connecting user.

After you have created your JMS stores, you can view them by clicking the Stores node in the Administration Console. Figure 15.8 shows three stores—JMSFileStore, JMSPagingStore, and JMSJDBCStore.

Figure 15.8. Viewing your JMS stores using the Administration Console.

A WebLogic JMS server manages connections and message requests on behalf of JMS clients (JMS Producers and Consumers). Before you can configure any JMS Destinations (Queues or Topics), you must first configure a JMS server.

To create a JMS server using the Administration Console, follow these steps:

Your new JMS server will be added to the JMS servers node under the JMS node in the left pane of the Administration Console. By default, WebLogic Server also creates a Destination and Session Pool node under your new JMS server to enable you to configure associated Queues, Topics, and Session Pools, as shown in Figure 15.10.

Figure 15.10. Reviewing your configured JMS servers using the Administration Console.

After defining a JMS server, you must configure one or more destinations for your JMS server. The steps to create a JMS queue or topic associated with your JMS server are very similar using the Administration Console, as follows:

You will need to reboot your WebLogic Server once you have configured your JMS server’s destinations.

Configuring JMS Server and Destination Thresholds

After you have created a JMS server and an associated destination object, you can specify thresholds and quotas to control the size and number of messages stored in the JMS server and destination objects, respectively.

You can use quotas to limit the number of messages or the size of all messages that can be stored. Messages sent after a quota has been exceeded are rejected and the sender is notified through an exception. Quotas can be used to control individual applications from consuming all resources within WebLogic JMS.

Thresholds are used to signal high or low load congestion periods based upon the number or size of the messages stored in the JMS server or destination. Upon exceeding an upper or lower threshold, triggered events are launched; for example messages are logged on the WebLogic Server indicating the respective threshold condition being exceeded.

To configure the thresholds and quotas for your JMS server or destination, select your JMS server or Destination node in the Administration Console, and then select the Threshold & Quotas tab. As shown in Figure 15.13, there are threshold and quota attributes related to the number of messages and size of messages.

Figure 15.13. Configuring the threshold and quota attributes for your JMS topics and queues.

The default values, which are set to –1, imply there is no limit to the size or number of messages in your JMS server or destination object. You should understand the messaging behavior of your JMS application and the performance of your WebLogic Server under expected normal, off-peak, and peak operational conditions before you modify these default settings.

Note

▸ To learn how to performance test your WebLogic Server, see “Performance Tuning the WebLogic Server,” p. 1011.

You can define and configure one or more uniquely named connection factories to create JMS connections with predefined attributes. Once created, WebLogic Server adds the connection factories to the JNDI namespace during startup, which JMS clients can retrieve using WebLogic JNDI. By default WebLogic JMS defines one connection factory, weblogic.jms.ConnectionFactory. However, all configuration attributes are set to their default values for this default connection factory, preventing you from having any control over the JMS server on which the connection factory may be deployed. For this reason, if you would like to target a connection factory to a particular JMS server, you must create a new connection factory and specify the appropriate JMS server target(s).

To create a new JMS connection factory, follow these steps:

You will now need to reboot WebLogic Server to bind the JNDI name for this newly created JMS connection factory to WebLogic JNDI in order for message producers and consumers to gain a connection to your JMS server.

Once you have configured your JMS server, you can monitor run-time statistics related to your JMS server and all its associated destinations and session pools using the following steps:

JMS statistics continue to increment as long as the server is running. Statistics are only reset when WebLogic Server is rebooted.

You can also monitor durable subscribers to the destinations associated with your JMS server using the following steps:

PTP and Pub/Sub clients are similar. They follow similar steps using different interfaces. PTP clients must acquire a QueueConnectionFactory and then use it to create a QueueConnection. The connection then serves as a session factory and creates a QueueSession.

PTP clients use the following code to acquire a connection factory, a connection, and a session:

// Lookup a Connection Factory in JNDI
QueueConnectionFactory qCF = (QueueConnectionFactory)ic.lookup("javax.jms.QueueConnectionFactory");

// Create a Connection
QueueConnection qCon = qCF.createQueueConnection();

// Create a Session
QueueSession qSession = qCon.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

PTP producers need to look up a queue in JNDI. The JNDI name is designated when the queue is created in the WebLogic Administration Console:

// Lookup the Queue
Queue queue = (Queue)ic.lookup("book.jms.BookQueue");

After getting the queue, a producer needs to create a QueueSender to actually send messages. Clients use the session to create a sender bound to the queue:

// Create a QueueSender bound to the Queue
QueueSender qSender = qSession.createSender(queue);

Next the client creates a message. In this example the client creates a TextMessage and initializes it with “Hello World!”:

// Create a TextMessage
TextMessage mes = qSession.createTextMessage("Hello World!");

Finally the client starts the connection and sends the message five times. Then the client closes the connection.

 qCon.start() ;
    for (int i = 0; i<5; i++ ) {
      qSender.send(mes);
    }
 finally {
   try {
        qCon.close();
        qSender.close();
        qSession.close();
   }
   catch (JMSException je) {
       System.out.println(je);
   }
 }

Listing 15.2 provides the complete PTPProducer JMS application.

import javax.naming.*;
import javax.jms.*;
import java.util.*;

public class PTPProducer {

  private final static String NAMING_FACTORY=
                       "weblogic.jndi.WLInitialContextFactory";
  private final static String QCON_FACTORY="javax.jms.QueueConnectionFactory";
  private final static String MESSAGE_QUEUE="book.jms.BookQueue";

public static void main ( String[] argv )
      throws NamingException, JMSException
  {
      QueueConnection qCon = null;
      QueueSender qSender = null;
      QueueSession qSession = null;

    try {

      // Create a JNDI initial context
      Properties props = new Properties() ;
      props.put( Context.INITIAL_CONTEXT_FACTORY, NAMING_FACTORY);
      props.put( Context.PROVIDER_URL, "t3://localhost:7001");
      InitialContext ic = new InitialContext(props);

      // Look up a Connection Factory in JNDI
      QueueConnectionFactory qCF =
                 (QueueConnectionFactory)ic.lookup(QCON_FACTORY);

      // Create a Connection
      qCon = qCF.createQueueConnection();

      // Create a Session
      qSession =
              qCon.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

      // Lookup the Queue
      Queue queue = (Queue)ic.lookup(MESSAGE_QUEUE);

      // Create a QueueSender bound to the Queue
      qSender = qSession.createSender(queue);

      // Create a TextMessage
      TextMessage mes = qSession.createTextMessage("Hello World!");

      // Start the Queue, send the message
      qCon.start()
         for (int i = 0; i<5; i++ ) {
           qSender.send(mes);
           System.out.println("Sending Hello World - " + i +
                              " to the JMS Queue");
         }
    }
    catch ( NamingException e ) {
           System.out.println(e);
    }
    finally {
       //Release all JMS resources
        try {
             qCon.close();
             qSender.close();
             qSession.close();
        }
        catch (JMSException je) {
            System.out.println(je);
        }
    }
  }
}

PTP consumers share the basic steps with PTP producers. The differences begin after acquiring a queue. A PTP consumer must get a QueueReceiver to interact with the queue as follows:

// Create a QueueReceiver bound to the Queue
QueueReceiver qReceiver = qSession.createReceiver(queue);

The PTP consumer uses a blocking read in an endless loop to receive messages from the queue. Listing 15.3 provides the entire PTPConsumer JMS application.

import javax.naming.*;
import javax.jms.*;
import java.util.*;

public class PTPConsumer {

  private final static String NAMING_FACTORY=
                             "weblogic.jndi.WLInitialContextFactory";
  private final static String QCON_FACTORY="javax.jms.QueueConnectionFactory";
  private final static String MESSAGE_QUEUE="book.jms.BookQueue";

  public static void main ( String[] argv ){

    try {

      // Create a JNDI initial context
      Properties props = new Properties() ;
      props.put( Context.INITIAL_CONTEXT_FACTORY, NAMING_FACTORY);
      props.put( Context.PROVIDER_URL, "t3://localhost:7001");
      InitialContext ic = new InitialContext(props);

      // Look up a Connection Factory in JNDI
      QueueConnectionFactory qCF =
                             (QueueConnectionFactory)ic.lookup(QCON_FACTORY);

      // Create a Connection
      QueueConnection qCon = qCF.createQueueConnection();

      // Create a Session
      QueueSession qSession =
                        qCon.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

      // Look up the Queue
      Queue queue = (Queue)ic.lookup(MESSAGE_QUEUE);

      // Create a QueueReceiver bound to the Queue
      QueueReceiver qReceiver = qSession.createReceiver(queue);

      // Start the Queue, send, and close
      qCon.start() ;

      TextMessage msg;

         while( true ) {
           msg = (TextMessage)qReceiver.receive();
           System.out.println(msg.getText()) ;
          }
    } catch ( NamingException e1 ) {
      System.out.println(e1);
    } catch ( JMSException e2 ) {
      System.out.println(e2);
    }
  }
}

To run the PTPProducer and PTPConsumer JMS applications, follow these steps:

Figure 15.21 shows the PTPProducer sending messages and the PTPConsumer receiving messages.

Figure 15.21. Running the PTPProducer and PTPConsumer applications.

JMS Clients are not limited to blocking receive calls. Clients may create and register listeners to listen asynchronously for JMS messages. These listeners are classes that implement javax.jms.MessageListener. This interface has only one method, public void onMessage(Message msg). Using a message listener requires a JMS consumer to first register as a listener for a specific destination with WebLogic JMS. After a message is delivered to this destination, WebLogic JMS calls the attached listener and passes the message to its onMessage() method, which is then processed by the consumer to that destination.

The following code sample shows how a listener implements MessageListener, which is a local class:

// Start the Queue
qCon.start() ;

class Listener implements MessageListener {
  public void onMessage( Message msg ) {
    try {
      System.out.println(((TextMessage)msg).getText()) ;
    } catch (JMSException ex) {
      System.out.println(ex);
    }
  }
}
qReceiver.setMessageListener(new Listener()) ;
while(true);

Listing 15.4 shows the PTPConsumerAsync JMS application with a complete implementation of the asynchronous listener.

import javax.naming.*;
import javax.jms.*;
import java.util.*;

public class PTPConsumerAsync {

  private final static String NAMING_FACTORY=
                                     "weblogic.jndi.WLInitialContextFactory";
  private final static String QCON_FACTORY="javax.jms.QueueConnectionFactory";
  private final static String MESSAGE_QUEUE="book.jms.BookQueue";

  public static void main ( String[] argv ){

    try {

      // Create a JNDI  initial context
      Properties props = new Properties() ;
      props.put( Context.INITIAL_CONTEXT_FACTORY, NAMING_FACTORY);
      props.put( Context.PROVIDER_URL, "t3://localhost:7001");
      InitialContext ic = new InitialContext(props);

      // Look up a Connection Factory in JNDI
      QueueConnectionFactory qCF =
                             (QueueConnectionFactory)ic.lookup(QCON_FACTORY);

      // Create a Connection
      QueueConnection qCon = qCF.createQueueConnection();

      // Create a Session
      QueueSession qSession =
                      qCon.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

      // Look up the Queue
      Queue queue = (Queue)ic.lookup(MESSAGE_QUEUE);

      // Create a QueueReceiver bound to the Queue
      QueueReceiver qReceiver = qSession.createReceiver(queue);

      // Start the Queue, send, and close
      qCon.start() ;
      class Listener implements MessageListener {
        public void onMessage( Message msg ) {
          try {
            System.out.println(((TextMessage)msg).getText());
          } catch (JMSException ex) {
            System.out.println(ex);
          }
        }
      }
      qReceiver.setMessageListener(new Listener());
      while(true);
    } catch ( NamingException e1 ) {
      System.out.println(e1);
    } catch ( JMSException e2 ) {
      System.out.println(e2);
    }
  }
}

Alternatively, an easier and more scalable means to receiving asynchronous messages is by configuring JMS Session Pools (a WebLogic JMS feature), which can receive messages from a destination and forward them to a server-side message listener class that you provide to process messages. The message listener class processes the message through a single-threaded onMessage() method. It is important to note that the message listener class does not need to implement the code to create server objects, such as connection factories and sessions, because they are created by the Session Pool. For example, Listing 15.5 provides the code for a simple SessionConsumer class that can be used in conjunction with JMS Session Pools.

import javax.jms.*;

public class SessionConsumer implements MessageListener {
        public void onMessage( Message msg ) {
          try {
            System.out.println(((TextMessage)msg).getText());
          } catch (JMSException ex) {
            System.out.println(ex);
          }
        }
}

After you have compiled your message listener class, it is important you deploy it to the WebLogic classpath. You can also modify your operating system or WebLogic classpath variable to include the directory where your class files reside.

To create a WebLogic session pool using the Administration Console, follow these steps:

After you have defined a session pool, you must configure one or more connection consumers (queues or topics) that will be used to retrieve server sessions and process messages. To configure connection consumers using the Administration Console, follow these steps:

To test the operation of a session pool, you can run the PTPProducer application in Listing 15.2, which will send messages to the book.jms.BookQueue. The message listener (SessionConsumer) class will process and output the messages to WebLogic command shell, as shown in Figure 15.24.

Figure 15.24. Testing the operation of a session pool.

Pub/Sub producers need to look up a topic in JNDI as follows:

// Look up the Topic
Topic topic = (Topic)ic.lookup("book.jms.BookTopic");

The JNDI name is designated when the topic is created via the WebLogic Administration Console.

After getting the topic, a producer needs to create a TopicPublisher to actually send messages. Clients use the session to create a publisher bound to the topic:

// Create a TopicPublisher bound to the Topic
TopicPublisher tPublisher = tSession.createPublisher(topic);

Next the client creates a message. In this example the client creates a TextMessage and initializes it with “Hello World!”:

// Create a TextMessage
TextMessage mes = qSession.createTextMessage("Hello World!");

Finally the client starts the connection and sends the message five times. Then the client closes the connection:

// Start the Queue, send, and close
qCon.start();
for(int i = 0; i<5; i++) {
  tPublisher.publish(mes);
}
qCon.close();

Listing 15.6 is the complete PubSubProducer application.

import javax.naming.*;
import javax.jms.*;
import java.util.*;

public class PubSubProducer {

  private final static String NAMING_FACTORY=
                                     "weblogic.jndi.WLInitialContextFactory";
  private final static String TCON_FACTORY="javax.jms.TopicConnectionFactory";
  private final static String MESSAGE_TOPIC="book.jms.BookTopic";

  public static void main ( String[] argv ){

    try {

      // Create a JNDI  initial context
      Properties props = new Properties() ;
      props.put( Context.INITIAL_CONTEXT_FACTORY, NAMING_FACTORY);
      props.put( Context.PROVIDER_URL, "t3://localhost:7001");
      InitialContext ic = new InitialContext(props);

      // Look up a Connection Factory in JNDI
      TopicConnectionFactory tCF =
                           (TopicConnectionFactory)ic.lookup(TCON_FACTORY);

      // Create a Connection
      TopicConnection tCon = tCF.createTopicConnection();

      // Create a Session
      TopicSession tSession =
                   tCon.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);

      // Look up the Topic
      Topic topic = (Topic)ic.lookup(MESSAGE_TOPIC) ;

      // Create a TopicPublisher bound to the Topic
      TopicPublisher tPublisher = tSession.createPublisher(topic);

      // Create a TextMessage
      TextMessage mes = tSession.createTextMessage("Hello World!");

      // Start the Topic, send the message, and close the topic
      tCon.start() ;

      for(int i = 0; i<5; i++ ) {
        tPublisher.publish(mes);
        System.out.println("  Sending Hello World - " + i +
                               " to the JMS Topic");
      }
         tCon.close() ;

    } catch ( NamingException e1 ) {
      System.out.println(e1);
    } catch ( JMSException e2 ) {
      System.out.println(e2);
    }
  }
}

Pub/Sub consumers share the basic steps with Pub/Sub producers. The differences begin after acquiring a topic. A Pub/Sub consumer must get a TopicSubscriber to interact with the topic:

// Create a TopicSubscriber bound to the Topic
TopicSubscriber tSubscriber = tSession.createSubscriber(topic);

The Pub/Sub consumer uses a blocking read in an endless loop to receive messages from the queue. Listing 15.7 is the entire PubConsumer application.

import javax.naming.*;
import javax.jms.*;
import java.util.*;

public class PubSubConsumer {

  private final static String NAMING_FACTORY=
                                     "weblogic.jndi.WLInitialContextFactory";
  private final static String TCON_FACTORY="javax.jms.TopicConnectionFactory";
  private final static String MESSAGE_TOPIC="book.jms.BookTopic";

  public static void main ( String[] argv ){

    try {

      // Create a JNDI initial context
      Properties props = new Properties() ;
      props.put( Context.INITIAL_CONTEXT_FACTORY, NAMING_FACTORY);
      props.put( Context.PROVIDER_URL, "t3://localhost:7001");
      InitialContext ic = new InitialContext(props);

      // Look up a Connection Factory in JNDI
      TopicConnectionFactory tCF =
                               (TopicConnectionFactory)ic.lookup(TCON_FACTORY);

      // Create a Connection
      TopicConnection tCon = tCF.createTopicConnection();

      // Create a Session
      TopicSession tSession =
                        tCon.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);

      // Look up the Topic
      Topic topic = (Topic)ic.lookup(MESSAGE_TOPIC);

      // Create a TopicSubscriber bound to the Topic
      TopicSubscriber tSubscriber = tSession.createSubscriber(topic);

      // Start the Topic, send the message, and close the topic
      tCon.start() ;

      TextMessage msg;

      while( true ) {
        msg = (TextMessage)tSubscriber.receive();
        System.out.println(msg.getText()) ;
      }

    } catch ( NamingException e1 ) {
      System.out.println(e1);
    } catch ( JMSException e2 ) {
      System.out.println(e2);
    }
  }
}

To run the PubSubProducer and PubSubConsumer JMS applications:

Figure 15.25 shows both the PubSubProducer and three PubSubConsumer applications. All the consumers receive all the messages placed in the topic.

Figure 15.25. Running the PubSubProducer and PubSubConsumer Applications.

Not all JMS clients need every message on a queue or topic. Clients may create QueueReceiver or TopicSubscriber objects with message selectors. A message selector acts as a filter to control which messages get delivered. Message selectors are SQL-like clauses that evaluate to true or false. If a selector is true the message will be delivered. Selector expressions may use standard JMS headers or optional properties, including user settable properties, to use in the Boolean evaluation. Message selectors cannot filter a message based upon the contents of the message body.

The PubSubProducer application (refer to Listing 15.5) can be adapted to set an optional property field. In this case each text message will get a message number. The name of the property is mesnum:

// set the mesnum property for selectors
mes.setIntProperty("mesnum",i);

A subscriber, such as the PubSubConsumer application in Listing 15.6, that only wanted messages 0 and 1 would need to define a selector like “mesnum=0” or “mesnum=1”, which would be a command-line parameter to the application (argv[0]):

// Create a TopicSubscriber bound to the Topic
TopicSubscriber tSubscriber = tSession.createSubscriber(topic,argv[0],true);

The createSubscriber method takes a topic, selector, and a flag. The flag indicates whether to ignore messages that may be sent by this client.

Figure 15.26 shows the output of the modified PTPProducer and PTPConsumer applications, PubSubPFilter (Producer) and PubSubCFilter (Consumer). PubSubPFilter creates five text messages with the property mesnum set to 0 through 4. PubSubCFilter has been started two times with different message selectors.

Figure 15.26. Running the PubSubPFilter and PubSubCFilter applications.