JMS and Message-Driven Beans

JMS and Message-Driven Beans

All EJB 3.0 vendors must support a JMS provider. Most vendors have a JMS provider built in and must support other JMS providers through the JCA. However, regardless of whether your vendor has its own JMS provider or allows you to integrate some other provider, a JMS provider is an absolute necessity for supporting message-driven beans. By forcing the adoption of JMS, Sun has guaranteed that EJB developers can expect to have a working JMS provider on which messages can be both sent and received.

JMS as a Resource

JMS is a vendor-neutral API that can be used to access enterprise messaging systems. Enterprise messaging systems (a.k.a. message-oriented middleware) facilitate the exchange of messages between software applications over a network. The role of JMS isn't unlike the role of JDBC: just as JDBC provides a common API for accessing many different relational databases, JMS provides vendor-independent access to enterprise messaging systems. Although messaging products aren't as familiar as database products, there's no shortage of messaging systems that support JMS, including JBossMQ, IBM's MQSeries, BEA's WebLogic JMS service, Sun Microsystems' Sun ONE Message Queue, and Sonic's SonicMQ. Software applications that use the JMS API for sending or receiving messages are portable from one JMS vendor to another.

Applications that use JMS are called JMS clients , and the messaging system that handles routing and delivery of messages is called the JMS provider . A JMS application is a business system composed of many JMS clients and, generally, one JMS provider. A JMS client that sends a message is called a producer , and a JMS client that receives a message is called a consumer . A single JMS client can be both a producer and a consumer.

In EJB, enterprise beans of all types can use JMS to send messages. The messages are consumed by other Java applications or by message-driven beans. JMS facilitates sending messages from enterprise beans using a messaging service , sometimes called a message broker or router. Message brokers have been around for a couple of decadesthe oldest and most established is IBM's MQSeriesbut JMS is fairly new, and it is specifically designed to deliver a variety of message types from one Java application to another.

Reimplementing the TravelAgent EJB with JMS

We can modify the TravelAgent EJB developed in Chapter 11 so that it uses JMS to alert some other Java application that a reservation has been made. The following code shows how to modify the bookPassage( ) method so that the TravelAgent EJB sends a simple text message based on a description obtained from the TicketDO object:

@Resource(mappedName="
ConnectionFactoryNameGoesHere
")
private ConnectionFactory connectionFactory;

@Resource(mappedName="TicketTopic")
private Topic topic;

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

    if (customer == null || cruise == null || cabin == null) {
        throw new IncompleteConversationalState( );
    }
    try {
        Reservation reservation = new Reservation(
                              customer, cruise, cabin, price, new Date( ));
        entityManager.persist(reservation);

        process.byCredit(customer, card, price);
 
        TicketDO ticket = new TicketDO(customer, cruise, cabin, price);

     Connection connect = factory.createConnection( );
        Session session = connect.createSession(true,0);
        MessageProducer producer = session.createProducer(topic);
        TextMessage textMsg = session.createTextMessage( );
        textMsg.setText(ticketDescription);

        producer.send(textMsg);
        connect.close( );

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

While all the code we added might look a little overwhelming, the basics of JMS are not all that complicated.

ConnectionFactory and Topic

In order to send a JMS message, we need a connection to the JMS provider and a destination address for the message. A JMS connection factory makes the connection to the provider possible; the destination address is identified by a Topic object. Both the connection factory and the Topic object are obtained by using @javax.annotation.Resource to inject these objects directly into the fields of the TravelAgent EJB:

@Resource(mappedName="ConnectionFactoryNameGoesHere")
private ConnectionFactory connectionFactory;

@Resource(mappedName="TicketTopic")
private Topic topic;

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

^[*] This analogy is not perfect. One might also say that the Session is analogous to the DataSource, since both represent transaction-resource connections.

The Topic object itself represents a network-independent destination to which the message will be addressed. In JMS, messages aren't sent directly to applications; they're sent to topics or queues. 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 a virtual channel. A queue is another type of destination that we'll discuss in detail later.

Connection and Session

The ConnectionFactory is used to create a Connection, which is an actual connection to the JMS provider:

Connection connect = connectionFactory.createConnection( );
Session session = connect.createSession(true,0);

Once you have a Connection , you can use it to create a Session. A Session allows you to group the actions of sending and receiving messages. In this case, you need only a single Session. Using multiple Sessions is helpful if you wish to produce and consume messages in different threads. Session objects use a single-threaded model, which prohibits concurrent access to a single Session from multiple threads. The thread that creates a Session is usually the thread that uses that Session's producers and consumers (i.e., MessageProducer and MessageConsumer objects). If you wish to produce and consume messages using multithreading, you must create a different Session object for each thread.

The createSession( ) method has two parameters:

createSession(boolean transacted, 
 int acknowledgeMode)
 

According to the EJB specifications, 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 acknowledgeMode, 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; others do not.

It's good programming practice to close a Connection after it has been used:

Connection connect = factory.createConnection( );
...
connect.close( );

MessageProducer

The Session is used to create a MessageProducer, 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:

MessageProducer producer = session.createProducer(topic);

TextMessage 
 textMsg = session.createTextMessage( );
textMsg.setText(ticketDescription);
producer.send(textMsg);

Message types

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, and 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 rather than a TextMessage :

TicketDO 
 ticket = new TicketDO(customer,cruise,cabin,price);
...
MessageProducer producer = session.createProducer(topic);

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

producer.send(mapMsg);

The attributes of 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);
...
MessageProducer producer = session.createProducer(topic);

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

producer.send(mapMsg);

In addition to 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.

JMS Application Client

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. This application is a simple JMS client that prints a description of each ticket as it receives the messages. We'll assume that the TravelAgent EJB is using TextMessage to send a description of the ticket to the JMS clients. Here's how the JMS application client might look:

import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.ConnectionFactory;
import javax.jms.Connection;
import javax.jms.Session;
import javax.jms.Topic;
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( );

        ConnectionFactory factory = (ConnectionFactory)
            jndiContext.lookup("ConnectionFactoryNameGoesHere");
        Topic topic = (Topic)jndiContext.lookup("TopicNameGoesHere");
        Connection connect = factory.createConnection( );
        Session session =
            connect.createSession(false,Session.AUTO_ACKNOWLEDGE);
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(this);

        connect.start( );
    }

    public void onMessage(Message message) {
        try {

            TextMessage textMsg = (TextMessage)message;
            String text = textMsg.getText( );
            System.out.println("\n RESERVATION RECEIVED\n"+text);

        } catch(JMSException jmsE) {
            jmsE.printStackTrace( );
        }
    }

    public static InitialContext getInitialContext( ) {
        // create vendor-specific JNDI context here
    }
}

The constructor of JmsClient_1 obtains the ConnectionFactory 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, here's how the getInitialContext( ) method for the JBoss application server would be coded:^[*]

^[*] JNDI also allows the properties to be set in a jndi.properties file that contains the property values for the InitialContext and that can be discovered dynamically at runtime. In this book, I chose to set the properties explicitly.

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,
       " org.jboss.security.jndi.JndiLoginInitialContextFactory");
    env.put(Context.PROVIDER_URL, " jnp://hostname:1099");
    return new InitialContext(env);
}

Once the client has the ConnectionFactory and Topic, it creates a Connection and a Session in the same way as the TravelAgent EJB does. The main difference is that the Session object is used to create a MessageConsumer rather than a MessageProducer . The MessageConsumer is designed to process incoming messages that are published to its Topic:

Session session =
    connect.createSession(false,Session.AUTO_ACKNOWLEDGE);
MessageConsumer consumer = session.createConsumer(topic);
consumer.setMessageListener(this);
connect.start( );

The MessageConsumer can receive messages directly or delegate message processing 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's onMessage( ) method is invoked to receive and process a copy of the message:

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

JMS Is Asynchronous

One of the principal advantages of JMS messaging is that it's asynchronous. In other words, a JMS client can send a message without having to wait for a reply. Contrast this flexibility with the synchronous messaging of Java RMI or JAX-RPC. Each time a client invokes a bean's method, it blocks the current thread until the method completes execution. This lock-step processing makes the client dependent on the availability of the EJB server, resulting in a tight coupling between the client and the enterprise bean. JMS clients send messages asynchronously to a destination (topic or queue) from which other JMS clients can also receive messages. When a JMS client sends a message, it doesn't wait for a reply; it sends the message to a router, which is responsible for forwarding the message to other clients. There's no effect on the client if one or more recipients are unavailable; it just goes ahead with its work. It's the router's responsibility to make sure that the message eventually reaches its destination. Clients sending messages are decoupled from the clients receiving them; senders are not dependent on the availability of receivers.

The limitations of RMI make JMS an attractive alternative for communicating with other applications. Using the standard JNDI environment-naming context, an enterprise bean can obtain a JMS connection to a JMS provider and use it to deliver asynchronous messages to other Java applications. For example, a TravelAgent session bean can use JMS to notify other applications that a reservation has been processed, as shown in Figure.

Using JMS with the TravelAgent EJB

In this case, the applications receiving JMS messages from the TravelAgent EJB may be message-driven beans, other Java applications in the enterprise, or applications in other organizations that benefit from being notified that a reservation has been processed. Examples might include business partners who share customer information or an internal marketing application that adds customers to a catalog mailing list.

Because messaging is inherently decoupled and asynchronous, the transactions and security contexts of the sender are not propagated to the receiver. For example, when the TravelAgent EJB sends the ticket message, the JMS provider may authenticate it, but the message's security context won't be propagated to the JMS client that received the message. When a JMS client receives the message from the TravelAgent EJB, the client has no idea about the security context under which the message was sent. This is how it should be, because the sender and receiver often operate in environments with different security domains.

Similarly, transactions are never propagated from the sender to the receiver. For one thing, the sender has no idea who the receivers of the message will be. If the message is sent to a topic, there could be one receiver or thousands; managing a distributed transaction under such ambiguous circumstances is not tenable. In addition, the clients receiving the message may not get it for a long time after it is sent; there may be a network problem, the client may be down, or there may be some other problem. Transactions are designed to be executed quickly because they lock up resources, and applications can't tolerate the possibility of a long transaction with an unpredictable end.

A JMS client can, however, have a distributed transaction with the JMS provider so that it manages the send or receive operation in the context of a transaction. For example, if the TravelAgent EJB's transaction fails for any reason, the JMS provider discards the ticket message sent by the TravelAgent EJB. Transactions and JMS are covered in more detail in Chapter 16.

JMS Messaging Models

JMS provides two types of messaging models: publish-and-subscribe and point-to-point . The JMS specification refers to these as messaging domains. In JMS terminology, publish-and-subscribe and point-to-point are frequently shortened to pub/sub and p2p (or PTP), respectively. This chapter uses both the long and short forms throughout.

In the simplest sense, publish-and-subscribe is intended for a one-to-many broadcast of messages, and point-to-point is intended for one-to-one delivery of messages (see Figure).

JMS messaging domains

Each messaging domain (i.e., pub/sub and p2p) has its own set of interfaces and classes for sending and receiving messages. This results in two different APIs, which share some common types. JMS 1.1 introduced a Unified API that allows developers to use a single set of interfaces and classes for both messaging domains. Only the Unified API is used in this chapter.

Publish-and-subscribe

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. The pub/sub messaging model is largely a push-based model , in which messages are automatically broadcast to consumers without the consumers 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. 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.

Point-to-point

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, in which messages are requested from the queue instead of being pushed to the client automatically.^[*] A queue may have multiple receivers, but only one receiver may receive each message. As shown earlier in Figure, the JMS provider takes 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 JMS specification does not specifically state how the p2p and pub/sub models must be implemented. Either model can use push or pullbut conceptually, pub/sub is push and p2p is pull.

Which messaging model should you use?

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.

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 which model is a better fit for the application. With pub/sub, any number of subscribers can be listening on a topic, and all of them will 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 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 ability is extremely important when messages need to be processed separately but in tandem.

Session Beans Should Not Receive Messages

JmsClient_1 was designed to consume messages produced by the TravelAgent EJB. Can another session bean receive those messages also? The answer is yes, but it's a really bad idea.

Session beans respond to calls from EJB clients and they cannot be programmed to respond to JMS messages, as can message-driven beans. It's impossible to write a session or entity bean that is driven by incoming messages. It is possible to develop a session bean that can consume a JMS message from a business method, but an EJB client must call the method first. For example, when the business method on the Hypothetical EJB is called, it sets up a JMS session and then attempts to read a message from a queue:

@Stateless
public class HypotheticalBean implements HypotheticalRemote {
    @Resource(mappedName="ConnectionFactory");
    private ConnectionFactory factory;

    @Resource(mappedName="MyQueue")
    private Queue queue;

    public String businessMethod( ) {

        try{
            Connection connect = factory.createConnection( );
            Session session = connect.createSession(true,0);
            MessageConsumer receiver = session.createConsumer(queue);
            TextMessage textMsg = (TextMessage)receiver.receive( );

            connect.close( );

            return textMsg.getText( );
        } catch(Exception e) {
            throws new EJBException(e);
        }

    }
    ...
}

The message consumer is used to proactively fetch a message from the queue. While this operation has been programmed correctly, it is dangerous because a call to the MessageConsumer.receive( ) method blocks the thread until a message becomes available. If a message is never delivered, the thread is blocked indefinitely! If no one ever sends a message to the queue, the MessageConsumer just sits there waiting, forever.

To be fair, there are other receive( ) methods that are less dangerous. For example, receive(long timeout) allows you to specify a time after which the MessageConsumer should stop blocking the thread and give up waiting for a message. There is also receiveNoWait( ), which checks for a message and returns null if none is waiting, thus avoiding a prolonged thread block. However, this operation is still dangerous. There is no guarantee that the less risky receive( ) methods will perform as expected, and the risk of programmer error (e.g., using the wrong receive( ) method) is too great.

The moral of the story is simple: don't write convoluted code trying to force session beans to receive messages. If you need to receive messages, use a message-driven bean; MDBs are specially designed to consume JMS messages.

Learning More About JMS

JMS (and enterprise messaging in general) represents a powerful paradigm in distributed computing. While this chapter has provided a brief overview of JMS, it has presented only enough material to prepare you for the discussion of message-driven beans in the next section. To understand JMS and how it is used, you will need to study it independently.^[*] Taking the time to learn JMS is well worth the effort.

^[*] For a detailed treatment of JMS, see Java Message Service (O'Reilly).