8.2. Introducing JMS

If you are already familiar with the Java Message Service (JMS), you can skip this section.

JMS is a vendor-independent API that can be used on the J2EE platform for enterprise messaging. JMS abstracts access to message servers much like JDBC and JNDI abstract access to databases and naming services, respectively. When you use JMS for enterprise messaging, your application is portable and vendor independent with different message service providers.

Before we show you JMS programming, let's define some JMS terms. A Java application that uses JMS is a JMS client. The messaging system that handles the routing and delivery of messages is a JMS provider. A JMS client that generates a message is a producer and a JMS client that receives a message is a consumer. A JMS client can be both a provider (sender) and a consumer (receiver).

With JMS, a client sends a message asynchronously to a destination. JMS has two types of destinations: topic and queue. When a JMS client sends a message to a topic or queue, the client does not wait for a reply. The JMS provider is responsible for routing, and delivering the messages to the appropriate consumer. Providers are not dependent on consumers, and clients that send messages are decoupled from the clients that receive them.

Messaging Domains

How does the JMS provider know where to send messages? In JMS, a message domain defines how the JMS provider routes and delivers messages. There are two ways to set up message routing in a JMS message domain: publish/subscribe and point-to-point. You'll often hear the shorter terms “pub/sub” and “PTP” used for these message domains. Since choosing the right message domain is an important step in designing robust JMS applications and message-driven beans, let's discuss each one separately and show you how to apply them.

Publish/Subscribe

Figure 8-3 shows the two JMS messaging domains. The top half illustrates pub/sub messaging. This model applies to a one-to-many broadcast of messages. Think of it like a magazine publisher with a list of subscribers. A single publisher sends one message to all subscribers who register to be notified when a message is sent. Every subscriber receives a copy of the message.

Pub/sub messaging is a push model, which means that the JMS provider automatically broadcasts messages to consumers. This is an important aspect of publish/subscribe messaging, because it means that subscribers do not have to poll for messages or ask to receive them. Note that pub/sub messaging uses a topic destination for sending and receiving messages.

Point-to-Point

The lower half of Figure 8-3 illustrates PTP messaging. This model applies to a one-to-one conversation between a sender and a receiver. Think of it like a telephone call between two people. A sender transmits a message to one receiver who wants to listen. If there are multiple receivers, only one receiver gets the message (typically the first receiver that reads it).

Point-to-point messaging is a pull model, meaning that consumers must request messages from the JMS provider. This implies that PTP messaging guarantees that only one consumer will process each message. Note that PTP messaging uses a queue destination for sending and receiving messages.

JMS Building Blocks

JMS requires quite a few objects to make everything work. There are administrative objects (such as topic or queue destinations and connection factories), message producers and consumers, connections, and the messages themselves. Figure 8-4 shows these building blocks and how they relate to each other. This diagram augments the following sections as we explain in more detail the responsibilities of each object. Destinations (topic or queue) and the Connection Factory objects are created and maintained administratively by the J2EE application server. JMS clients access them through the JNDI Initial Context lookup. We shade them to show this difference. The other objects are created as indicated in the diagram.

Topic Messaging

With pub/sub messaging, the publisher and subscriber send and receive messages through a topic destination managed by the JMS provider. Producers use a topic to publish messages, and subscribers use the topic to consume them. Remember that subscribers receive messages automatically after registering to listen, and each subscriber to a topic receives its own copy of the message that was published to that topic. You may add or remove subscribers and publishers dynamically at runtime, and a single message from one publisher can be sent to thousands of subscribers.

Let's go through the steps to create a topic destination and publish a message. We'll also show you how to receive messages from the topic destination. Table 8.1 shows the JMS objects required to publish and subscribe to topic destinations.

TopicConnectionFactory and Topic

The first step is to create a JMS connection factory object. After obtaining the context, a successful lookup of the JNDI name returns a TopicConnectionFactory object, as follows.

InitialContext jndi = new InitialContext();

TopicConnectionFactory factory =
    (TopicConnectionFactory)jndi.lookup
      ("java:comp/env/jms/MyTopicConnectionFactory");

This is the object we will use to make the actual connection to the JMS provider. Note that TopicConnectionFactory is similar to the JDBC DataSource object, which is used to connect to a database.

A topic destination is necessary for the message we intend to publish. This is also done with a JNDI lookup, which returns a Topic object.

Topic topic = (Topic)jndi.lookup
       ("java:comp/env/jms/MyTopic");

Design Guideline

When the JMS software detects a problem, it throws a JMSException. Therefore, you must either place JMS code inside a try block or within a method that includes JMSException in its throws clause.

TopicConnection and TopicSession

Once you have a TopicConnectionFactory object, you can connect to the JMS provider.

TopicConnection con = factory.createTopicConnection();

Next, you create a TopicSession object if the connection is successful.

TopicSession session = con.createTopicSession(false,
      Session.AUTO_ACKNOWLEDGE);

The createTopicSession() method has two arguments. The first argument is a boolean that determines whether transaction is enabled for the session or not. The second argument specifies the acknowledgement mode. As you will see later, the EJB container manages the transaction and acknowledgment modes for session objects; hence, we show only typical initial values here.

The TopicSession object is important because it's used to create messages, publishers, and subscribers. Typically, you'll have only one object in a single thread program, because sending and receiving messages is usually done with the same thread that creates the TopicSession object. It is possible, however, to create separate TopicSession objects in the same thread or with each thread in a multithreading application.

A TopicConnection object has start(), stop(), and close() methods to help JMS clients manage the connection. Consumers should call start() when ready to receive messages. The stop() method prevents new messages from being received until start() is called again.

Make sure you always close a topic connection when you are finished with the session. This makes the connection resource available to other clients, since the JMS provider maintains a pool of topic connections.

TopicConnection con = factory.createTopicSession();
. . .
con.start();      // start connection
. . .
con.close();      // close connection

TopicPublisher (Producer Side)

Now that we have a JMS connection and a session object, we create a publisher to send messages to a topic destination.

Topic topic = (Topic)jndi.lookup
       ("java:comp/env/jms/MyTopic");
TopicSession session = con.createTopicSession(false,
       Session.AUTO_ACKNOWLEDGE);
TopicPublisher publisher =
       session.createPublisher(topic);

To create a topic publisher with the createPublisher() method, we call it with a TopicSession object (session) and pass a Topic destination (topic) as an argument. The topic is determined by a JNDI lookup.

Message Types

Ok, we have a publisher, now what? It's time for the publisher to send a message to the topic destination. JMS has several different message types, but here's the simplest, a text message.

TextMessage textMsg = session.createTextMessage();
textMsg.setText("World Series Tickets for sale");
publisher.publish(textMsg);

All JMS messages are Java objects with a header and a message body. The header contains metadata, and the message body is data of a specific JMS type. JMS provides a wide variety of different types that apply to most applications. Five message types are available: TextMessage, StreamMessage, MapMessage, ObjectMessage, and BytesMessage. All are interface types extended from a generic Message interface.

We've already seen a TextMessage example, so let's look at the other JMS message types now. The following example is MapMessage, which handles key-value pairs. Here we use it to publish someone's e-mail address.

MapMessage mapMsg = session.createMapMessage();
mapMsg.setString("Robert", "[email protected]");
publisher.publish(mapMsg);

The StreamMessage type is handy for working with different types.

String name = "Bob"; int age = 25; float weight = 165;
StreamMessage streamMsg = session.createStreamMessage();
streamMsg.writeString(name);
streamMsg.writeInt(age);
streamMsg.writeFloat(weight);
publisher.publish(streamMsg);

Consumers need to read the data types from a StreamMessage in the same order they were written.

The ObjectMessage type handles Java serializable objects. Here we use it to publish a RecordingVO object from our Music Collection Database (see “RecordingVO Class” on page 96).

// albums is an ArrayList of RecordingVO objects
RecordingVO rec = (RecordingVO)albums.get(2);
ObjectMessage objectMsg = session.createObjectMessage();
objectMsg.setObject(rec);
publisher.publish(objectMsg);

The BytesMessage type is useful for byte-oriented data.

byte[] data = { 10, 20, 30, 40 };
BytesMessage byteMsg = session.createBytesMessage();
byteMsg.writeBytes(data);
publisher.publish(byteMsg);

JMS messages have acknowledgment modes and can be prioritized and time stamped. Messages do not expire by default, although you can assign an expiration time to a message if you want to. Likewise, a JMS message is persistent by default, which means a message cannot get lost. If the JMS provider fails, the message will be delivered after the server recovers. You can make a message's delivery mode nonpersistent, but there's no guarantee that the message will be delivered if the JMS provider fails.

TopicSubscriber (Consumer Side)

Messages aren't very useful if no one is around to receive them. Here's how a consumer subscribes to a topic destination.

Topic topic = (Topic)jndi.lookup
       ("java:comp/env/jms/MyTopic");
TopicSession receiver = con.createTopicSession(false,
       Session.AUTO_ACKNOWLEDGE);
TopicSubscriber subscriber =
       receiver.createSubscriber(topic);

To create a topic subscriber with the createSubscriber() method, we call it with a TopicSession object (receiver) and pass a Topic destination (topic) as an argument. Note that a topic subscriber would use the same JNDI lookup name as a topic publisher.

One important aspect of pub/sub messaging is that if a message is sent to a topic and a subscriber has disconnected from the JMS provider, the subscriber will not receive the message unless it has been set up with a durable subscription. If a durable subscriber is unavailable, the JMS provider saves all messages for that topic and redelivers them when the durable subscriber reconnects.

Design Guideline

A persistent message is not enough with topics and pub/sub messaging. To guarantee delivery and implement true store-and-forward messaging, a topic subscriber should be created with a durable subscription.

Here's how to create a durable subscriber.

TopicSubscriber subscriber =
       receiver.createDurableSubscriber(topic, "News");

The createDurableSubscriber() method is called with a TopicSession object (receiver). Its first argument is a topic destination, and the second argument is a subscription name. This string name can be useful in administrative tools that track the status of durable subscriptions.

Durable subscribers are very important to receivers who may not be able to stay connected to JMS servers all the time but still have to receive messages when they reconnect.

The onMessage() Method

All that's left is to show you how to receive messages. Suppose one client publishes a topic text message to another client, as shown in Figure 8-5. Client1, acting as a topic publisher, looks up the JNDI topic name, connects to the JMS server, and sends a message to the topic destination. Client2, acting as a topic subscriber, connects to the JMS server with the same JNDI lookup name as Client1. Client2, however, has to be set up properly to receive the message.

In JMS, both pub/sub and PTP messaging support a Java event model for handling incoming messages. This event mechanism is similar to what is used in Java applets and JavaBeans. Basically, a client class implements the MessageListener interface and registers itself with the TopicSubscriber object. The receiving client class must define an onMessage() method to handle incoming messages from the topic.

When a message is sent to a topic, the TopicSubscriber object calls the onMessage() method in any MessageListener object registered with the subscriber. This registration is typically done in the client constructor after creating the subscriber object.

The following skeletal code shows how all this is done for a text message.

import javax.jms.*;
public class Client2 implements MessageListener {

  public Class2() {
    try {
      . . .
      TopicSubscriber subscriber =
          receiver.createSubscriber(topic);
      subscriber.setMessageListener(this);
    }
    catch (JMSException ex) { . . . }
  }

  public void onMessage(Message message) {
    try {
      TextMessage msg = (TextMessage)message;
      String text = msg.getText();
      System.out.println(text);
    } catch (JMSException ex) { . . . }
  }
  . . .
}

Because Client2 implements the MessageListener interface, we must provide an onMessage() method in the Client2 class. To register with the subscriber, the constructor calls setMessageListener(this) with a TopicSubscriber object. Inside onMessage(), we print the message received as a string.

Queue Messaging

A message queue implements the point-to-point messaging model in JMS. Queues are similar to topics, except that a queue is a one-to-one connection rather than a one-to-many broadcast. Although JMS queues can perform synchronous messaging, we will discuss only how to send and receive messages asynchronously here. Remember that each message is guaranteed to be delivered to only one receiver.

Design Guideline

Messages delivered to a queue remain in the JMS server, even if a consumer is not connected. The JMS server delivers the message to the consumer when they reconnect. Note that this behavior is different from topics, which do not guarantee a message delivery unless the receiver is a durable subscriber.

Working with queues is similar to working with topics. Table 8.2 shows the JMS objects required to send and receive queue messages.

There is a QueueConnectionFactory object for the JNDI lookup name that connects to the JMS provider, and a QueueConnection object to create QueueSession objects. A QueueSession object can create QueueSender and QueueReceiver objects to deliver messages or receive messages from a queue. This is similar in concept to the TopicPublisher and TopicSubscriber objects in pub/sub messaging.

To demonstrate how point-to-point messaging is done with queues, let's examine a program that sends a text message to itself. This program could presumably be used to verify that a JMS server is operating properly. Listing 8.1 shows the program (PingServerApp.java). The example combines the separate steps for defining the producer and consumer that we showed you earlier with topics, only we use a queue for the messaging. (This example uses JMS, but it is not a message-driven bean.)

Listing 8.1. PingServerApp.java

// PingServerApp.java - JMS Message Queues
import java.io.*;
import javax.jms.*;
import javax.naming.*;
public class PingServerApp implements MessageListener {
  private QueueConnection connect = null;
  private QueueSession sender, receiver;
  private QueueSender qsender;
  private QueueReceiver qreceiver;

  public PingServerApp() {
    try {
      // Setup required by both JMS consumers & producers
      InitialContext jndi = new InitialContext();

      QueueConnectionFactory factory =
        (QueueConnectionFactory)jndi.lookup
        ("java:comp/env/jms/MyQueueConnectionFactory");
      connect = factory.createQueueConnection();
      Queue queue = (Queue)jndi.lookup
        ("java:comp/env/jms/MyQueue");

      // JMS producer setup
      sender = connect.createQueueSession(false,
          Session.AUTO_ACKNOWLEDGE);
      qsender = sender.createSender(queue);

      // JMS consumer setup
      receiver = connect.createQueueSession(false,
          Session.AUTO_ACKNOWLEDGE);
      qreceiver = receiver.createReceiver(queue);
      qreceiver.setMessageListener(this);
      connect.start();

    } catch (Exception ex) {
      ex.printStackTrace();
      System.exit(1);
    }
  }

  public void sendMessage(String message) {
    try {
      TextMessage msg = sender.createTextMessage();
      msg.setText(message);
      qsender.send(msg);
      System.out.println("Sent message: " + message);
    } catch (Exception ex) {
      ex.printStackTrace();
    }
  }

  public void onMessage(Message message) {
    try {
      TextMessage msg = (TextMessage)message;
      String text = msg.getText();
      System.out.println("Received message: " + text);
    } catch (Exception ex) {
      ex.printStackTrace();
    }
  }

  public void exit() {
    try {
      if (connect != null)
        connect.close();
    } catch (Exception ex) {
      ex.printStackTrace();
      System.exit(1);
    }
    System.exit(0);
  }

  public static void main(String args[]) {
    PingServerApp ping = new PingServerApp();
    try {
      BufferedReader stdin = new
        BufferedReader(new InputStreamReader(System.in));

      while (true) {
        System.out.println("type quit to exit");
        ping.sendMessage("Test Message");
        String cmd = stdin.readLine();
        if (cmd.equals("quit"))
          ping.exit();
      }
    } catch (Exception ex) {
       ex.printStackTrace();
    }
    }
}

In the constructor, JNDI lookups yield the QueueConnectionFactory and Queue objects for connecting to the JMS provider with a queue destination. After creating separate QueueSession objects (sender and receiver), the constructor attaches the same queue to the sender and receiver objects (qsender and qreceiver). Note that PingServerApp implements the MessageListener interface and registers itself as a listener with the qreceiver object. This means we'll need an onMessage() method in PingServerApp to receive messages.

The main() method creates an instance of PingServerApp to send and receive messages. Inside the while loop, we send a text string message to ourself. The readLine() method blocks program execution until the user types input followed by a return. When a messages arrives, the event handler invokes method onMessage(). The exit() method terminates the program when the user types “quit”. Otherwise, the message is sent again in the loop.

The sendMessage() method creates a TextMessage from its string argument and sends the message via the qsender object. When it's available by the JMS provider, the onMessage() method receives the message and converts it to a String for display. The exit() method closes the queue connection.

JMS Deployment Descriptors

Listing 8.2 shows the deployment descriptor for our application client (PingServerApp), which uses JMS resources.

Listing 8.2. Deployment Descriptor for PingServerApp

<application-client>
 <display-name>PingServerApp</display-name>
 <resource-ref>
    <res-ref-name>jms/MyQueueConnectionFactory
    </res-ref-name>
    <res-type>javax.jms.QueueConnectionFactory
    </res-type>
    <res-auth>Container</res-auth>
    <res-sharing-scope>Shareable</res-sharing-scope>
 </resource-ref>
 <resource-env-ref>
    <resource-env-ref-name>jms/MyQueue
    </resource-env-ref-name>
    <resource-env-ref-type>javax.jms.Queue
    </resource-env-ref-type>
 </resource-env-ref>
</application-client>

The <resource-ref> for the JMS QueueConnectionFactory specifies a JNDI name, interface type, and container authorization for the resource. This format makes the descriptor similar to a JDBC DataSource. Likewise, the <resource-env-ref> specifies a JNDI name and type for the Queue used by the program. The deployer maps the JMS QueueConnectionFactory and Queue elements to JMS factory and queue objects at deployment time.