11.6 Message Client Patterns

Once a message has been created and dispatched to an appropriate delivery channel, the next step is for the client to retrieve and process it. Retrieval mechanisms for messages need to balance speed of delivery, reliability, scalability, and ease of use. In this section, we'll look at some patterns for handling messages in client applications. The Message Handle pattern shows how to decouple message processing code from message retrieval code. The Polling Consumer and Event-Driven Consumer patterns explain how to get messages into your application. Message Façade demonstrates using messaging to isolate your business logic. Competing Consumers shows how to speed up message processing in high volume applications, and Message Selector will help you build more efficient client-side message processors.

11.6.1 Message Handler Pattern

All message clients perform two functions: they retrieve and process a message. It's possible to weave these two activities tightly together, but in more complex systems it can result in some pretty complex code. We want to isolate the latter function in a way that lets us share code across different message types. The Message Handler pattern abstracts the code responsible for receiving the actual message away from the code responsible for processing the message content.

This pattern allows us to change the method by which we receive a message without changing the code for processing the message. Changing message handling itself becomes easier, since we can drop message handlers in and out without major changes to the rest of the application (we can even do it dynamically). There's also a testing benefit, since you can write standalone test cases for an isolated handler method more easily than for an integrated component.

Example 11-1 shows a simple mail handler interface for incoming JavaMail messages. The code responsible for retrieving the message instantiates the appropriate handler and calls the handleMessage( ) method, passing in the new message. We'll see the message client code itself when we talk about the Polling Consumer pattern.

Example 11-1. Mail handler interface

public interface BlockingMailHandler 
{
  /** Process a message, returning true on success, false on failure.
        * Does not return until message is processed. */ 
  public boolean handleMessage(javax.mail.Message message);
}

A more complex application than this simple blocking handler might define a variety of message handler types, such as a threaded handler and a single thread model handler. A JMS example would look similar, but it would accept a javax.jms.Message object instead.

Most applications can get away with a transport-specific handler, since the mechanics of retrieving content from JMS and JavaMail message objects differ widely; most applications only use one transport mechanism for any particular use case. But occasionally, applications need to accept the same kind of message from a variety of different sources. One example of this situation is when some of the applications you're integrated with are separated by a firewall: remote applications can send their messages via email (probably encrypting the content) and local applications can use the MOM backbone. In these cases, you'll either want to provide two handleMessage( ) methods, or descend two different handler classes (one per message type) from the same base class (which contains the logic required for processing the message once it has been extracted from its transport container).

11.6.2 Polling Consumer Pattern

An important reason to use a messaging system is in order to allow one component of an integrated system to go down without immediately affecting all the others. The Polling Consumer pattern allows a client to periodically check for messages, rather than maintaining a constant connection to the message server. If the server is unavailable, the client can simply try again later, rather than ceasing all operation. Conversely, if the client goes down, the server will maintain the messages until the client starts up again. In this kind of client crash we still lose time, but we don't lose information.

A polling consumer periodically checks a message channel for new messages, reads them, and processes them. Generally, the consumer is implemented as a thread, which sleeps for a set amount of time before polling for new messages. Polling consumers are the most reliable way to receive Internet mail and are useful for JMS applications where an EJB container isn't used.

Example 11-2 shows the MailMonitor component. MailMonitor is a Java class that can be dropped into a standalone application (by instantiating and running the thread in the main( ) method) or into a web application (started, perhaps, by a ServletContextListener). The mail monitor checks a POP3 mailbox, reads every message, and invokes a BlockingMailHandler to process them. If a message is processed successfully, the client deletes it from the server. Once all messages are processed, the client waits 10 seconds and starts again.

Example 11-2. Standalone MailMonitor

import javax.mail.*;
import javax.mail.internet.*;
import java.io.*;

public class MailMonitor extends Thread {
  boolean interupted = false;
  BlockingMailHandler handler = null;
  private static String username = "contentmail";
  private static String password = "poppasswd";
  private static String mailServer = "mail.company.com";
  
  public MailMonitor(BlockingMailHandler mh) {
    handler = mh;
  }

  public void stopRunning(  ) {
    interupted = true;
  }
  
  public void run(  ) {
    Session session = Session.getDefaultInstance(System.getProperties(  ));
    Store store = null;
    try         {
      store = session.getStore("pop3");
    } catch (NoSuchProviderException nspe) {
      nspe.printStackTrace(  );
      return;
    }

    while(!interupted) {
      System.out.println("Looping to collect mail");
      try {
        if (!store.isConnected(  )) // should always be true
          store.connect(mailServer, -1, username, password);
        System.out.println("Connected");  
        Folder folder = store.getDefaultFolder(  );
        folder = folder.getFolder("INBOX");
        if (folder == null) {
          System.out.println("Unable to open INBOX");
          return;
        }

      System.out.println("Opening folders");
      // Try to open read/write. Open read-only if that fails.
      try {
        folder.open(Folder.READ_WRITE);
      } catch (MessagingException ex) {
        folder.open(Folder.READ_ONLY);
      }

      int totalMessages = folder.getMessageCount(  );
      int newMessages = folder.getNewMessageCount(  );
      System.out.println("Total Messages: " + totalMessages);
      try {
        Message messages[] = null;
        messages = folder.getMessages(1, totalMessages);

        for (int i = 0, i < messages.length; i++) {
          boolean mbDelete = handler.handleMessage(messages[i]);                                                                        
          // Delete the message
          if (mbDelete) {
            messages[i].setFlag(Flags.Flag.DELETED, true);
          }
        } // end for
      } catch (MessagingException e) {
        System.out.println("Unable to get Messages");
      }

      // Close the folder and store
      folder.close(true);
      store.close(  );

      } catch (MessagingException e) {
        System.out.println(e.toString(  ));
        return;
      } 
      
      try {
        this.sleep(10000);
      } catch (InterruptedException e) {
        System.out.println("Exiting");
        return;
      }
    }
  }
}

It's worth noting that in this design the mail handler is responsible for processing a message regardless of its validity. The mail handler implementation is also responsible for error reporting. The only time we want the handler to return false is when a time delay would allow the message to be processed later. Messages that cannot be processed now might be processable later, once other systems come back online or other messages arrive. That said, messaging systems make lousy persistent storage: you really should get the message out as quickly as possible, and deal with queuing in some other context, particularly if you don't have personal control over the mail server in question.

Polling consumers is also useful when an application doesn't need to process a message immediately, or when other factors such as high connection costs create economies of scale for processing multiple messages at once.

11.6.3 Event-Driven Consumer Pattern

Polling for messages can be a lot of work: your application is responsible for waking up periodically, checking for messages, and handling them in a safe fashion. Most of us have been spoiled by event-driven models in various areas of Java development. Servlets, for instance, are event-driven: the service( ) method is called in response to a specific request. The Event-Driven Consumer pattern allows us to forget about most of that code we built up in the Polling Consumer pattern section, in favor of a simple method that is invoked by the environment when necessary. The result is decreased code complexity and increased reuse, since more of the messaging retrieval logic is moved to the server.

Most JMS messages are handed in an event-driven manner. EJB 2.0 supports message-driven beans, which include an onMessage( ) method that is invoked whenever the application server detects a new message on a topic or queue. The base JMS API supports event-driven messages through the MessageListener interface. MessageListener defines an onMessage( ) method, which is called by the JMS subsystem whenever a new message is received.

Here's just about the simplest MessageListener class possible:

class MyMessageHandler implements MessageListener {

        public void onMessage(Message message) {
          TextMessage msg = (TextMessage)message;
          try { 
            System.out.println("Message Received:" + msg.getText(  )); 
          } catch (JMSException e) {
            System.out.println(e.getMessage(  ));
          }
        }
}

And here's how we'd associate it with a Topic (assuming we already have a session named session and a topic named topic):

   TopicSubscriber subscriber = session.createSubscriber(topic);
   MyMessageHandler handler = new MyMessageHandler(  );
   subscriber.setMessageListener(handler);

Each new message on the topic now triggers a call to onMessage( ). This process continues until message delivery on the connection is halted, the session is closed, or the program exits.

The underlying implementation depends on the messaging provider and driver. The driver might very well end up implementing a polling consumer behind the scenes, resulting in some overhead you might not ordinarily expect. Some providers deliver messages over RMI; the driver creates an RMI endpoint, which then delegates to the MessageListener. Check your messaging server documentation for more details.

JavaMail also provides an event-driven consumer via the classes in javax.mail.event, which support listening for changes to folders and message stores. However, we don't advocate the use of these classes: the implementations of the various Internet mail protocols don't support this kind of activity all that well, which means you'll have to test your application even more extensively than usual, and with the same exact infrastructure that you'll be using in production. It's better, in our opinion, to implement a polling consumer: it might be a bit more work, but you'll have tighter control and better visibility when things go wrong.

11.6.4 Message Façade Pattern

The Message Façade pattern describes an object that receives and acts on messages from the presentation tier. The message façade receives asynchronous messages from an external source, translates those messages into calls to appropriate business logic methods, executes the calls, and, if appropriate, sends a reply message containing the results. The rationale for isolating business logic behind the message façade is the same as for isolating it behind a business delegate or a session façade: easier maintenance and improved reuse.

Generally, the tradeoff to consider when developing business logic interfaces is whether you want to expose particular functionality in a synchronous or asynchronous manner. Most high-volume applications perform a surprisingly large amount of processing asynchronously; in an application deployed to a large number of users who may or may not have any particular loyalty to your company or institution, any activity that will take more than a second is a good candidate for asynchronous processing.

Message façades, like façades in general, have advantages and drawbacks. Since an application often takes several paths to its business logic (servlets, JMS, web services, and other routes as yet unknown), isolating that business logic behind a façade provides encapsulation and supports reuse. The business logic and domain model components themselves don't need to be aware of the message infrastructure and can be maintained separately. As with any messaging pattern, the business logic can be used asynchronously by the presentation tier. The cost is additional code, more infrastructure, and an additional layer of software between the user and the actual business process.

The Message Façade pattern is the third (and final) pattern in our series of business logic distribution patterns. We used business delegates to provide access to business logic within a JVM, via direct method invocations. We used session façades to provide synchronous access to remote business resources. Now, we'll use message facades to provide asynchronous access to remote resources.

Of course, our simple summary could be a little misleading. There's no reason that different kinds of façades can't, in the course of fulfilling their responsibilities, make calls to each of the others. A business delegate might call out to a remote session façade and fire off a batch of messages to a message façade. In general, creating a business delegate in order to allocate responsibility to the remote façade is a good idea, regardless of the façade type.

Figure 11-6 shows a client accessing a message façade, which in turn unpacks the message and calls a business delegate to do the heavy lifting. The business delegate then calls a DAO to handle persistence.

Figure 11-6. Message façade using business delegates and DAOs

Implementing a message façade is easy. All you really have to do is implement a consumer. In EJB 2.0, you can create a message-driven bean that includes a JMS onMessage( ) method. Within the onMessage( ) method, you can call your business objects and assemble a reply message. Using an MDB rather than a regular JMS listener gives you transactional awareness: the message-driven bean will participate in any currently running transaction; if a database action fails, the message receipt can be rolled back as well.

You probably won't need to implement message façades with JavaMail all that often, but the process is the same: implement a polling consumer (using the MailMonitor.java framework or any other mechanism you wish) and add the business logic functions. When using JavaMail, you'll have to take more care with security, since you won't be able to count on the message server to verify that only authorized parties are sending messages to your application.

11.6.5 Message Selector Pattern

The profusion of message queues and topics is a frequent problem in JMS applications. Designers often establish separate queues or topics for each type of message in the application. Supporting a high number of consumers can eat up a lot of resources, and managing a large number of separate messaging channels can burden IS departments, particularly when each new message type requires the reconfiguration of the message server. Email messaging presents a similar problem, but in the form of multiple target email addresses.

The Message Selector pattern allows us to handle multiple message types on a single queue or topic without excessively complex logic because it requires the client to differentiate between the different types of messages on the same channel.

JMS has a fairly powerful selection criteria API, based on a subset of the SQL selection syntax. The selection criteria works against the properties set on a particular message. A message selector that only retrieves messages in which the MessageType property equals the string "StockCheck" would look like "MessageType = 'StockCheck'".

With the JMS publish-subscribe API, we specify the selector when we create the subscriber—it can't be changed after the fact. The boolean parameter at the end of the createSubscriber( ) call indicates whether messages originating from the local session should be ignored:

   TopicSubscriber subscriber = session.createSubscriber(topic, 
            "MessageType = 'StockCheck'", true);

For point-to-point messaging, we specify the selector in the call to the createConsumer( ) method, which has the same syntax—except that there is no boolean for ignoring local messages, since that doesn't make sense on a queue.

Using a selector has different consequences depending on the type of messaging used. If a message selector is applied to a queue, the remaining messages stay on the queue and can be read by the same session later on, or by other sessions (see the Competing Consumers pattern, below). When a message selector is used with a topic, the messages that aren't selected are lost to that subscriber, regardless of the durability of the subscription (the alternative would be completely unworkable: thousands upon thousands of undelivered messages, clogging up the middleware for subscriptions that have no intention of ever retrieving them).

11.6.6 Competing Consumers Pattern

Connecting front-line systems with back-office systems via messaging allows the front-line systems to keep on interacting with users, even while performing complex activities. This ability to multitask pushes the delay away from the end user, but it doesn't actually eliminate any bottlenecks. The work is spread out, but messages must still be processed.

Thankfully, there's no rule that says everything in life has to be linear. You can set multiple consumers loose on a single message queue and have them compete for the right to process the messages. Hence, the Competing Consumers pattern. Each consumer reads messages off the queue as quickly as it can and processes them in whatever manner is appropriate to the application. If the consumers start to fall behind we can just add more.

Figure 11-7 shows an example of multiple consumers listening to a single JMS queue. The queue is responsible for ensuring that each message is delivered to one and only one consumer. It makes competing consumers easy to implement in a JMS environment. The biggest potential problem is that the messages need to be self-contained: you have no guarantee that the same consumer will retrieve all the messages from a particular client, so using sequenced messages is impossible without additional work to coordinate between the consumers.

Figure 11-7. Competing consumers

Implementing multiple handlers without a MOM system is a bit more complicated. Since multiple applications cannot reliably access a single Internet email address concurrently, you need to divide the messages between handlers at a central point. Assigning an email address to each of the handlers can accomplish this division. Incoming messages can then be spread across the handlers, either by having one handler that receives all the messages and resends them (which can be much faster than the actual message processing, allowing one client to take care of it), or by configuring the mail server, where possible, to distribute messages in a round-robin fashion. The various handlers can implement polling consumers pointed at the individual addresses.

When implementing competing consumers with JavaMail, you should never have multiple applications accessing the same mailbox. Most mail servers won't allow it anyway, and those that do won't provide you with sufficient isolation: messages will be read more than once, or might be missed entirely.