In order to send a JMS message we need a connection to the JMS provider and a destination address for the message. The connection to the JMS provider is made possible by a JMS connection factory; the destination address of the message is identified by a Topic object. Both the connection factory and the Topic object are obtained from the TravelAgent EJB’s JNDI ENC:

TopicConnectionFactory factory = (TopicConnectionFactory)
    jndiContext.lookup("java:comp/env/jms/TopicFactory");
        
Topic topic = (Topic)
    jndiContext.lookup("java:comp/env/jms/TicketTopic");

The TopicConnectionFactory in JMS is similar in function to the DataSource in JDBC. Just as the DataSource provides a JDBC connection to a database, the TopicConnectionFactory provides a JMS connection to a message router.^[47]

The Topic object itself represents a network-independent destination to which the message will be addressed. In JMS, messages are sent to destinations—either topics or queues—instead of directly to other applications. A topic is analogous to an email list or newsgroup; any application with the proper credentials can receive messages from and send messages to a topic. When a JMS client receives messages from a topic, the client is said to subscribe to that topic. JMS decouples applications by allowing them to send messages to each other through a destination, which serves as virtual channel.

JMS also supports another destination type, called a Queue. The difference between topics and queues is explained in more detail later.

The TopicConnectionFactory is used to create a TopicConnection , which is an actual connection to the JMS provider:

TopicConnection connect = factory.createTopicConnection();

TopicSession session = connect.createTopicSession(true,0);

Once a TopicConnection is obtained, it can be used to create a TopicSession. A TopicSession allows the Java developer to group the actions of sending and receiving messages. In this case you will need only a single TopicSession. However, having more than one TopicSession object is frequently helpful: if you wish to produce and consume messages using multithreading, a different Session needs to be created by each thread accessing that thread. This is because JMS Session objects use a single-threaded model, which prohibits concurrent accessing of a single Session from multiple threads. The thread that creates a TopicSession is usually the thread that uses that Session’s producers and consumers (i.e., TopicPublisher and TopicSubscriber objects). If you wish to produce and consume messages using multithreading, a different Session should be created and used by each thread.

The createTopicSession() method has two parameters:

createTopicSession(boolean transacted, int acknowledgeMode)

According to the EJB 2.0 specification, these arguments are ignored at runtime because the EJB container manages the transaction and acknowledgment mode of any JMS resource obtained from the JNDI ENC. The specification recommends that developers use the arguments true for transacted and 0 for acknowlegeMode, but since they are supposed to be ignored, it should not matter what you use. Unfortunately, not all vendors adhere to this part of the specification. Some vendors ignore these parameters while some do not. Consult your vendor’s documentation to determine the proper values for these parameters in both container-managed and bean-managed transactions.

It’s good programming practice to close a TopicConnection after its been used, in order to conserve resources:

TopicConnection connect = factory.createTopicConnection();
...

connect.close();

The TopicSession is used to create a TopicPublisher, which sends messages from the TravelAgent EJB to the destination specified by the Topic object. Any JMS clients that subscribe to that topic will receive a copy of the message:

TopicPublisher publisher = session.createPublisher(topic);

TextMessage textMsg = session.createTextMessage();
textMsg.setText(ticketDescription);
publisher.publish(textMsg);

In JMS, a message is a Java object with two parts: a header and a message body. The header is composed of delivery information and metadata, while the message body carries the application data, which can take several forms: text, serializable objects, byte streams, etc. The JMS API defines several message types (TextMessage , MapMessage , ObjectMessage , and others) and provides methods for delivering messages to and receiving messages from other applications.

For example, we can change the TravelAgent EJB so that it sends a MapMessage instead of a TextMessage:

TicketDO ticket = new TicketDO(customer,cruise,cabin,price);
...
TopicPublisher publisher = session.createPublisher(topic);

MapMessage mapMsg = session.createTextMessage();
mapMsg.setInt("CustomerID", ticket.customerID.intValue());
mapMsg.setInt("CruiseID", ticket.cruiseID.intValue());
mapMsg.setInt("CabinID", ticket.cabinID.intValue());
mapMsg.setDouble("Price", ticket.price);

publisher.publish(mapMsg);

The attributes of the MapMessage (CustomerID, CruiseID, CabinID, and Price) can be accessed by name from those JMS clients that receive it.

As an alternative, the TravelAgent EJB could be modified to use the ObjectMessage type, which would allow us to send the entire TicketDO object as the message using Java serialization:

TicketDO ticket = new TicketDO(customer,cruise,cabin,price);
...
TopicPublisher publisher = session.createPublisher(topic);

ObjectMessage objectMsg = session.createObjectMessage();
ObjectMsg.setObject(ticket);

publisher.publish(objectMsg);

In addition to the TextMessage, MapMessage, and ObjectMessage, JMS provides two other message types: StreamMessage and BytesMessage. StreamMessage can take the contents of an I/O stream as its payload. BytesMessage can take any array of bytes, which it treats as opaque data.

When a JMS resource is used, it must be declared in the bean’s XML deployment descriptor, in a manner similar to the JDBC resource used by the Ship EJB in Chapter 10:

<enterprise-beans>
    <session>
        <ejb-name>TravelAgentBean</ejb-name>
        ...
        <resource-ref>
            <res-ref-name>jms/TopicFactory</res-ref-name>
            <res-type>javax.jms.TopicConnectionFactory</res-type>
            <res-auth>Container</res-auth>
        </resource-ref>
        <resource-ref>
            <res-ref-name>jdbc/titanDB</res-ref-name>
            <res-type>javax.sql.DataSource</res-type>
            <res-auth>Container</res-auth>
        </resource-ref>
        <resource-env-ref>
            <resource-env-ref-name>jms/TicketTopic</resource-env-ref-name>
            <resource-env-ref-type>javax.jms.Topic</resource-env-ref-type>
        </resource-env-ref>
        ...
    </session>

The <resource-ref> for the JMS TopicConnectionFactory is similar to the <resource-ref> declaration for the JDBC DataSource: it declares the JNDI ENC name, interface type, and authorization protocol. In addition to the <resource-ref>, the TravelAgent EJB must also declare the <resource-env-ref>, which lists any “administered objects” associated with a <resource-ref> entry. In this case, we declare the Topic used for sending a ticket message. At deployment time the deployer will map the JMS TopicConnectionFactory and Topic declared by the <resource-ref> and <resource-env-ref> elements to a JMS factory and topic.

To get a better idea of how JMS is used, we can create a Java application whose sole purpose is receiving and processing reservation messages. We will develop a very simple JMS client that simply prints a description of each ticket as it receives the messages. We’ll assume that the TravelAgent EJB is using the TextMessage to send a description of the ticket to the JMS clients. The following code shows how the JMS application client might look:

import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.Topic;
import javax.jms.Session;
import javax.jms.TopicSubscriber;
import javax.jms.JMSException;
import javax.naming.InitialContext;


public class JmsClient_1 implements javax.jms.MessageListener {

    public static void main(String [] args) throws Exception {
        
        if(args.length != 2)
            throw new Exception("Wrong number of arguments");
        
        new JmsClient_1(args[0], args[1]);
        
        while(true){Thread.sleep(10000);}
        
    }
        
    public JmsClient_1(String factoryName, String topicName) throws Exception {
            
        InitialContext jndiContext = getInitialContext();
        
        TopicConnectionFactory factory = (TopicConnectionFactory)
            jndiContext.lookup("TopicFactoryNameGoesHere");
        
        Topic topic = (Topic)jndiContext.lookup("TopicNameGoesHere");

        TopicConnection connect = factory.createTopicConnection();

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

        TopicSubscriber subscriber = session.createSubscriber(topic);

        subscriber.setMessageListener(this);
        
        connect.start();
    }
    
    public void onMessage(Message message) {
        try {
        
            TextMessage textMsg = (TextMessage)message;
            String text = textMsg.getText();
            System.out.println("
 RESERVATION RECIEVED:
"+text);
        
        } catch(JMSException jmsE) {
            jmsE.printStackTrace();
        }
    }
    
    public static InitialContext getInitialContext() {
        // create vendor-specific JNDI context here
    }
}

The constructor of JmsClient_1 obtains the TopicConnectionFactory and Topic from the JNDI InitialContext. This context is created with vendor-specific properties so that the client can connect to the same JMS provider as the one used by the TravelAgent EJB. For example, the getInitialContext() method for the WebLogic application server would be coded as follows:^[48]

public static InitialContext getInitialContext() {
    Properties env = new Properties();
    env.put(Context.SECURITY_PRINCIPAL, "guest");
    env.put(Context.SECURITY_CREDENTIALS, "guest");
    env.put(Context.INITIAL_CONTEXT_FACTORY,
       "weblogic.jndi.WLInitialContextFactory");
    env.put(Context.PROVIDER_URL, "t3://localhost:7001");
    return new InitialContext(env);
}

Once the client has the TopicConnectionFactory and Topic, it creates a TopicConnection and a TopicSession in the same way as the TravelAgent EJB. The main difference is that the TopicSession object is used to create a TopicSubscriber instead of a TopicPublisher. The TopicSubscriber is designed specifically to process incoming messages that are published to its specified Topic:

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

TopicSubscriber subscriber = session.createSubscriber(topic);
subscriber.setMessageListener(this);


connect.start();

The TopicSubscriber can receive messages directly, or it can delegate the processing of the messages to a javax.jms.MessageListener. We chose to have JmsClient_1 implement the MessageListener interface so that it can process the messages itself. MessageListener objects implement a single method, onMessage() , which is invoked every time a new message is sent to the subscriber’s topic. In this case, every time the TravelAgent EJB sends a reservation message to the topic, the JMS client will have its onMessage() method invoked so that it can receive a copy of the message and process it:

public void onMessage(Message message) {
    try {
        TextMessage textMsg = (TextMessage)message;
        String text = textMsg.getText();
        System.out.println("
 RESERVATION RECIEVED:
"+text);
        
    } catch(JMSException jmsE) {
        jmsE.printStackTrace();

    }
}

Please refer to Workbook Exercise 13.1, JMS as a Resource. This workbook is available free, in PDF format, at http://www.oreilly.com/catalog/entjbeans3/workbooks.

In publish-and-subscribe messaging, one producer can send a message to many consumers through a virtual channel called a topic. Consumers can choose to subscribe to a topic. Any messages addressed to a topic are delivered to all the topic’s consumers. Every consumer receives a copy of each message. The pub/sub messaging model is by and large a push-based model, where messages are automatically broadcast to consumers without them having to request or poll the topic for new messages.

In the pub/sub messaging model, the producer sending the message is not dependent on the consumers receiving the message. Optionally, JMS clients that use pub/sub can establish durable subscriptions that allow consumers to disconnect and later reconnect and collect messages that were published while they were disconnected.

The TravelAgent EJB in this chapter uses the pub/sub programming model with a Topic object as a destination.

The point-to-point messaging model allows JMS clients to send and receive messages both synchronously and asynchronously via virtual channels known as queues . The p2p messaging model has traditionally been a pull- or polling-based model, where messages are requested from the queue instead of being pushed to the client automatically.^[49]

A queue may have multiple receivers, but only one receiver may receive each message. As shown earlier in Figure 13-2, the JMS provider will take care of doling out the messages among JMS clients, ensuring that each message is consumed by only one JMS client. The JMS specification does not dictate the rules for distributing messages among multiple receivers.

The messaging API for p2p is similar to that used for pub/sub. The following code listing shows how the TravelAgent EJB could be modified to use the queue-based p2p API instead of the topic-based pub/sub model used in the earlier example:

public TicketDO bookPassage(CreditCardDO card, double price)
    throws IncompleteConversationalState {
        ...

        TicketDO ticket = new TicketDO(customer,cruise,cabin,price);

        String ticketDescription = ticket.toString();
        QueueConnectionFactory factory = (QueueConnectionFactory)
            jndiContext.lookup("java:comp/env/jms/QueueFactory");

        Queue queue = (Queue)
            jndiContext.lookup("java:comp/env/jms/TicketQueue");

        QueueConnection connect = factory.createQueueConneciton();

        QueueSession session = connect.createQueueSession(true,0);

        QueueSender sender = session.createSender(queue);

        TextMessage textMsg = session.createTextMessage();
        textMsg.setText(ticketDescription);
        sender.send(textMsg);
        connect.close();

        return ticket;
    } catch(Exception e) {
        throw new EJBException(e);
    }
}

The rationale behind the two models lies in the origin of the JMS specification. JMS started out as a way of providing a common API for accessing existing messaging systems. At the time of its conception, some messaging vendors had a p2p model and some had a pub/sub model. Hence JMS needed to provide an API for both models to gain wide industry support. The JMS 1.0.2 specification does not require a JMS provider to support both models. EJB 2.0 vendors, however, are required to support both messaging models.

Almost anything that can be done with the pub/sub model can be done with point-to-point, and vice versa. An analogy can be drawn to developers’ programming language preferences. In theory, any application that can be written with Pascal can also be written with C. Anything that can be written in C++ can also be written in Java. In some cases it comes down to a matter of preference, or which model you are already familiar with.

In most cases, the decision about which model to use depends on the distinct merits of each model. With pub/sub, any number of subscribers can be listening on a topic, and they will all receive copies of the same message. The publisher may not care if everybody is listening, or even if nobody is listening. For example, consider a publisher that broadcasts stock quotes. If any particular subscriber is not currently connected and misses out on a great quote, the publisher is not concerned. In contrast, a point-to-point session is likely to be intended for a one-on-one conversation with a specific application at the other end. In this scenario, every message really matters.

The range and variety of the data the messages represent can be a factor as well. Using pub/sub, messages are dispatched to the consumers based on filtering that is provided through the use of specific topics. Even when messaging is being used to establish a one-on-one conversation with another known application, it can be advantageous to use pub/sub with multiple topics to segregate different kinds of messages. Each kind of message can be dealt with separately through its own unique consumer and onMessage() listener.

Point-to-point is more convenient when you want only a particular receiver to process a given message once. This is perhaps the most critical difference between the two models: p2p guarantees that only one consumer processes each message. This is extremely important when messages need to be processed separately but in tandem.