Building a Lightweight JMS Provider

Here's a JMS provider that works in a single Java Virtual Machine

Eric is a software-development consultant and can be contacted at ericericbruno.com.

When writing Java applications, I often need a messaging system, even if the application runs within a single Java Virtual Machine (JVM). Almost any application can improve its design by taking advantage of named message queues or the anonymity of publish/ subscribe messaging, both of which are supported by the Java Message Service (JMS). Unfortunately, most JMS providers are bundled as part of a J2EE application server, making them big, difficult to setup and administer, and potentially expensive. This is neither practical nor efficient for use in standalone Java applications, never mind applets.

In this article, I present a JMS provider that is both lightweight in size and overhead—but not in features. Because of its small size (41 KB), it is useful within both Java applets and applications.

Full-featured JMS Providers are implemented as enterprise servers. This means they support reliable, guaranteed, message delivery across LANs, WANs, and even the Internet. Most JMS providers support two-phase commit transactions, meaning they can take part in distributed ACID transactions involving databases and other remote transaction systems.

This industrial strength messaging comes with a price in the form of cost, size, and complexity, all of which are reasonable for projects with enterprise requirements. However, any one of these characteristics makes JMS infeasible for more straightforward standalone Java applications and applets. Unfortunately, this means that your application or applet misses out on the object-oriented characteristics of communication that JMS offers.

The Lightweight JMS Provider

Most applications can make use of JMS's publish/subscribe and point-to-point messaging without requiring support for persistent messages and transactions. For instance, JMS-Light, the lightweight JMS provider I present here, supports JMS 1.1 features such as publish/subscribe, point-to-point message queues, synchronous and asynchronous messaging, request reply, durable subscriptions, queue browsers, all JMS message types (TextMessage, ObjectMessage, MapMessage, ByteMessage, StreamMessage), and multithreaded clients.

JMS-Light doesn't support enterprise features, such as persistent messages, transactions, message selectors, client-based message acknowledgment, inter-JVM messaging, or JNDI.

The most convenient feature of JMS-Light is that it requires no administration. Your application simply imports the 41-KB JAR file and creates topic and queue destinations on the fly. No installation or prior destination setup is required.

A true JMS provider is implemented as a process accessible through a JNDI server either built-in or separate. JNDI lets applications connect to the JMS provider at runtime, eliminating compile-time dependencies. Listing One illustrates the use of JNDI to connect to a JMS provider.

JMS-Light does not support JNDI, as it would add too much overhead. Instead, the singleton class JMSLight simulates JNDI to make the connection code similar. Being a singleton, it provides a central point for the creation and lookup for destinations within a process. Listing Two shows how to connect to JMS-Light.

The code to get the ConnectionFactory object is similar in both listings. For JMS-Light, the lookup method is called on the JMSLight Singleton object instead of the JNDI Context object of a true JMS provider. This design is intentional as it is easier to migrate the application code to a true JMS provider in the future.

JMS Destinations

With full-featured JMS providers, destinations can be created in advance, or on-the-fly at runtime. For destinations created in advance, JMS applications use JNDI to locate them. Since JMS-Light doesn't support JNDI, destinations must be created on-the-fly, like this:

import javax.jms.*;

import com.LJMS.*;

JMSLight ljms = JMSLight.getInstance();

// ...

// Create the "Orders" Queue

Queue q1 = session.createQueue("Orders");

// ...

// Lookup the "Orders" Queue

Queue q2 = (Queue)ljms.lookup("Orders");

JMS-Light stores the destinations in two HashMaps—one for queues and another for topics. Since JMSLight is a Singleton, the lookup method returns the same destination objects to all classes that call it. When the lookup method is called, both HashMaps are checked for the existence of the requested resource. Therefore, the name must be unique for both destination types; there cannot be both a queue and topic of the same name.

Domain-Specific Versus Generic JMS Classes

JMS defines generic interfaces and those specific to the message domain being used. Table 1 lists the generic and domain-specific JMS interfaces.

The use of the generic interfaces was introduced with JMS 1.1. Before that, these interfaces existed, but JMS apps had to use the domain-specific interfaces. This change results in application code that's more object oriented, as it hides the implementation details of the message domain.

JMS 1.1 specifies that providers must support generic interfaces and domain-specific interfaces equally, providing backward compatibility. JMS-Light uses a combination of inheritance and static methods to provide this support.

The classes that implement the domain-specific interfaces extend the classes that implement the generic interfaces. When domain-specific methods need to be called, the base classes call static methods on the domain-specific subclasses. These methods are safely accessible from the base class and no down-casting needs to be done. Figure 1 is a class diagram for JMS-Light's Session and TopicSession implementation classes. Figure 2 is a sequence diagram that contrasts the use of the generic JMS interfaces versus the domain-specific JMS interfaces.

Implementing Connections

The JMS Connection interface is a thread-safe entry point to the JMS provider used in creating JMS Sessions. The JMSLightConnection class implements the Connection interface in JMS-Light. Each JMSLightConnection object maintains a list of its active Session objects (those created by that Connection) within a Vector. The Session objects are created and added to this list when the createSession method is called.

The class JMSLightTopicConnection extends the JMSLightConnection class and implements the TopicConnection interface. This interface defines two methods—createConnectionConsumer and createTopicSession—in addition to the base Connection interface. The createConnection- Consumer method differs from the base interface's method only in that it takes a Topic as opposed to a generic Destination object as its first parameter. This method calls the base class's implementation via the keyword super. The base class implementation checks the type of the Destination object because it can be called from either a topic or queue connection; see Listing Three.

The createTopicSession method creates a domain-specific TopicSession object and adds it to the list of Sessions. This method cannot defer to the base class implementation of createSession because it must return a TopicSession object.

The implementation of JMSLightQueueConnection is similar to that of JMSLightTopicConnection. The only differences lie in the creation of the domain-specific QueueSession objects, as opposed to TopicSession objects.

Implementing Sessions

The JMSLightSession object implements the single-threaded Session interface. The main purpose of this class is to create MessageProducer, MessageConsumer, and Message objects.

A JMS client calls the createProducer and createConsumer methods with a generic Destination parameter, among others. The destination type is checked using the instanceof keyword, and the correct domain-specific static method is called.

For example, if the createProducer method is called where the destination object is a Topic instance, the static method _createPublisher is called on the JMSLightTopicSession class.

The JMSLightTopicSession class implements the domain-specific TopicSession interface. This class inherits most of its base implementation from the JMSLightSession class, and implements the publish/subscribe-specific methods: createPublisher, createSubscriber, createTopic, createTemporaryTopic, createDurableSubscriber, and unsubscribe.

For each method, except for unsubscribe, the implementation is deferred to the base class. The base class then calls the correct static methods on the domain-specific class, determined by the type of destination passed to it. When the _createPublisher or _createSubsciber static methods are called on JMSLightTopicSession, a JMSLightTopicPublisher or JMSLightTopicSubscriber object is created, respectively.

The createDurableSubscriber and unsubscribe methods work as a pair. A durable subscriber is one that receives all published messages for a given topic, even if that subscriber is not listening when the messages are sent.

In addition to creating a normal TopicSubscriber object, the createDurableSubscriber method sets the object's subscription name to that provided, and the durable property to True. A call is then made to the Topic object's method, addDurableSubscriber, and the Topic is added to a list of durable subscriptions within the TopicSession object.

The durable subscription can be terminated via a call to the JMSLightTopicSession.unsubscribe method, at which time missed messages will no longer be saved for this subscriber. Internally, this method performs these steps:

1. TopicSubscriber is located within the Session's list of consumers.
2. The Topic is located from the Session's list of durable subscriptions.
3. The TopicSubscriber's durable property is set to False.
4. The TopicSubscriber is removed from the Topic's list of durable subscribers.

The JMSLightQueueSession class is similar to JMSLightTopicSession except:

• There is no support for durable subscriptions. Because of the nature of a message queue, where each message is stored until delivered to a consumer, there is no need for a special, durable consumer.
• There is the addition of a browser component. The QueueBrowser is a special consumer that peeks at messages that exist on a queue without removing them.

Implementing Producers

The JMSLightMessageProducer class implements the JMS MessageProducer interface. This class implements the various send methods for sending messages to a topic or a queue. Clients can use this interface to send messages to destinations without needing to know if the destination is a topic or queue.

JMS-Light supports nonpersistent messages, and all messages are sent with the same priority. Messages can be given a "time to live" value that indicates when the message expires. If a message expires before delivery to a consumer, the message is destroyed and not delivered.

The send method is where all the action takes place in the JMSLightMessageProducer class. When called, this method sets the given message's properties, such as a message timestamp, and ensures that stream-based messages (such as a BytesMessage or StreamMessage) are reset.

Once the message is ready to be delivered, its destination type is checked. If the destination is a Topic, the JMSLightTopicPublisher class's static _publish method is called. If the destination is a Queue, the JMSLightQueueSender class's static send method is called. The domain-specific semantics of message delivery are handled in these static methods.

The static _publish method of the JMSLightTopicPublisher class defers all the work of publishing a message to the Topic destination class. The _publish method is called regardless of whether clients call send on the MessageProducer interface, or publish on the TopicPublisher interface.

As with the JMSLightTopicPublisher class, the static _send method on the JMSLightQueueSender class defers the message delivery to the Queue destination class itself. The _send method is called regardless of whether clients call send on the MessageProducer interface, or send on the QueueSender interface.

Implementing Consumers

The JMSLightMessageConsumer class implements the generic JMS interface, MessageConsumer. This class implements the receive methods to support synchronous client message delivery. It also stores a client's MessageListener object, if provided, to support asynchronous message delivery. Both domain-specific classes, JMSLightTopicSubscriber and JMSLightQueueReceiver, defer to this base class for client message delivery.

When a message is to be delivered, the domain-specific destination class calls the onMessage method on the consumer class. The method, onMessage (Listing Four) checks to see if the JMS client has provided a MessageListener for asynchronous message delivery. If there is no MessageListener, this method checks for a blocked client thread awaiting synchronous message delivery.

When a client calls the synchronous method, receive, the call blocks until a message arrives or the call times out, as specified by the caller. The calling thread is blocked via the Java Thread synchronization method, System.wait. The thread is signaled via a call to System.notify when a message arrives or a time out occurs. The domain-specific JMSLightTopicSubscriber and JMSLightQueueReceiver classes perform the thread signaling. Figure 3 illustrates the sequence of calls for synchronous message delivery.

Asynchronous message delivery is straightforward. When a message arrives, the JMSLightMessageConsumer class calls the onMessage method on the JMS client's MessageListener object. Figure 4 illustrates the sequence of calls for asynchronous message delivery.

Implementing Destinations

The JMSLightDestination class implements the Destination interface and defines two nested classes, ProducerNode and ConsumerNode. The Vector member variables—producers and consumers—hold references to these classes, which in turn hold references to actual MessageProducers and MessageConsumers for the destination. Figure 5 shows the relationship between the destination, and its producers and consumers.

A reference to the Connection object for each producer and consumer is stored to check if the connection has been started. A consumer's connection must have been started, or no messages are delivered to it. The domain-specific destination classes, JMSLightTopic and JMSLightQueue, handle the actual message delivery to each client of the destination.

The JMSLightTopic class extends JMSLightDestination, and contains code specific to Topic-based message delivery. Durable subscribers are stored within the "consumers" member variable, but are also tracked within a HashMap named durableSubscriptions.

The JMSLightTopic class defines a nested class, Notifier, which extends the Java Thread class. With each message that is published on a topic, an object of type Notifier is created. This object runs in its own thread and attempts delivery of the message to every Topic subscriber via the following steps:

1. Notifier calls the JMSLightTopic.notifySubscribers method, which iterates through all of the destination's subscribers within the vector.
2. Each subscriber's connection is checked to see if it is started. If so, a call is made to JMSLightTopicSubscriber.onMessage. If this call succeeds, processing continues with the next subscriber. If the onMessage call fails due to an exception, redelivery is attempted two more times before moving on to the next subscriber.
3. After all of the current subscribers have received the message, the thread terminates.

Durable Subscriptions

When the Notifier object has completed message delivery, the message must be stored for each durable subscriber that is not currently listening. The JMSLightTopic class defines a nested class, DurableSubscriberNode, to store these messages along with other durable subscription data. The method, saveDurableMessage(), stores the published message for each durable subscription whose subscriber is not currently listening.

Saved messages are delivered to a durable subscriber when it becomes active again, which occurs when the subscriber calls addDurableSubscriber. An object of class DurableNotifier is created to handle the delivery of all missed messages to this subscriber.

The durable subscription is maintained until that subscriber calls the unsubscribe method on the Topic class. When the call is made, the destination clears out any trace of the durable subscriber, including any undelivered messages.

The JMSLightQueue class extends JMSLightDestination, and leverages the storage of producers and consumers from that base class. This domain-specific class stores the actual messages sent to it in an object of class LinkedList, which works like a FIFO queue itself.

JMSLightQueue defines the nested class Notifier, just as the JMSLightTopic class does. However, only one Notifier object is created for the lifetime of the queue. This differs significantly from the JMSLightTopic implementation, where a Notifier object is created with each message that is sent.

The queue's Notifier object is told to sleep by calling the method Object.wait. When a producer sends a message, it's placed within the LinkedList of messages for that queue and the Notifier thread is signaled.

Implementing JMS Messages

JMS defines many message interfaces, all of which extend the base Message interface. The base interface defines a JMS message's header and properties, as well as the acknowledge method for client-based message delivery acknowledgment. In general, the message header properties are used for message routing. The message data or payload, as it is referred to, is defined by the more-derived message interfaces. Message properties and payload data can only be set before the message is sent.

The two most commonly used message interfaces are TextMessage and ObjectMessage, which define a Java String or Object as the message payload, respectively. The remaining three message interfaces are Stream based, meaning their methods are based on the java.io.DataOutputStream and java.io.DataInputStream classes.

The JMSLightMessage class implements the Message interface and defines the message states, types, header, and properties. The JMS-Light message header consists of: Message ID (unique per message); message type (pub/sub or queue); message state (pending, sent, or delivered); message priority (JMS-Light does not support varying message priorities and defaults to the JMS "normal" priority of four); Destination (Queue or Topic object); optional reply-to Destination for request/reply support; message delivery mode (JMS-Light supports only nonpersistent message delivery); optional correlation ID to link one message to another; message timestamp and expiration values, and a flag that indicates whether the message is being redelivered due to a potential failure on a previous delivery attempt

A JMS client can define application-specific message properties, which are name-value pairs. This provides a way for clients to customize a message's header. Each message property can be boolean, byte, short, int, long, float, double, or String.

JMS-Light defines a class named CollectionContainer, which stores the message properties. This class stores all properties in a Java Hashtable, which supports name-value pairs. JMS defines rules for message property retrieval, such as being able to retrieve any property as a String, regardless of what type it was when added to the Hashtable.

The JMSLightTextMessage class extends the JMSLightMessage class, implementing the TextMessage interface. This class leverages all of the base class functionality and adds the ability to set String as the message payload. The JMSLightObjectMessage class lets clients set any Java object that implements the java.io.Serializable interface as the message payload.

The JMSLightBytesMessage class extends JMSLightMessage and implements the BytesMessage interface. Because of the similarities in the interfaces, the JMSLightStreamMessage class extends the JMSLightBytesMessage class. Figure 6 shows this relationship.

The JMSLightBytesMessage class exists mainly for compatibility with external systems such as databases. The data is written and then read as one blob of bytes, all at once. Therefore, this class does not maintain a cursor-like position variable for every read.

For the JMSLightStreamMessage class, a cursor-like position variable is strictly maintained. With the StreamMessage interface, clients typically make multiple calls to write data into the stream, and later read data from the stream. Because of the ability to read and write raw bytes in the middle of the stream, a variable is maintained that tracks the byte position within the stream for each read.

In the end, the JMSLightStreamMessage class overrides all of the readX methods of the JMSLightBytesMessage class, simply to increment the position variable by the number of bytes read. The bulk of the implementation is leveraged from the base class.

The MapMessage class is different from the other Message interfaces in that data is stored as name-value pairs. This should sound familiar; it's identical to the way the Message interface stores application-defined header properties. Because of the similarities, the JMSLightMapMessage class leverages the CollectionContainer class for its payload, as well as its properties.

Using JMS-Light

To use JMS-Light, simply import the file, ljms.jar (available electronically; see "Resource Center," page 3) and call getInstance on the Singleton JMSLight class. Listing Two outlines the code needed to connect to JMS-Light. JMS-Light simulates JNDI to make transitioning your client code to a real JMS provider relatively easy.

The sample JMS order-system application I present here (available electronically) shows how to use JMS-Light. The application consists of a main class, OrderSystem, with two nested classes—Supplier and Customer. Both nested classes extend Thread and run in their own threads when created.

Listing Five shows that the OrderSystem class creates two destinations—a sales notification topic and order queue. These destinations exist within the Singleton JMSLight object and are used by both the Supplier and Customer classes.

Working with Topic destinations in JMS-Light is almost exactly the same as with a real JMS provider. Again, the only difference is the lack of JNDI support. In the order-system application, the Supplier class creates a publisher for the sale notification topic using code similar to this:

salesTopic = (Destination)ljms.lookup ( SALES_TOPIC_NAME );

connFact = (ConnectionFactory)

ljms.lookup("ConnectionFactory");

Connection conn =

connFact.createConnection();

Session session = conn.createSession (false,Session.AUTO_ACKNOWLEDGE);

MessageProducer saleProducer = session.createProducer(salesTopic);

The topic is looked up through the Singleton JMSLight object, and from there, a JMS connection and session is established as with any JMS provider. A publisher component is created for the topic by calling the createProducer method on the JMS session.

In this code, a TextMessage object is created via the JMS session and then published via a call to send on the MessageProducer:

// The sale notification is a JMS TextMessage

TextMessage msg =

session.createTextMessage();

msg.setText( saleItems[ item++ ] );

saleProducer.send( msg );

In the order-system app, the Customer class subscribes to the sale notification topic for asynchronous message delivery. As seen in Listing Six, the Customer class implements the MessageListener interface by defining the onMessage method. A call is then made to setMessageListener on the MessageConsumer object.

The order-system application uses a JMS Queue to handle new order requests. I chose a queue for two main reasons: First, the order requests are stored until received; second, one (and only one) receiver processes each order. If I opted for a pub/sub topic and multiple subscriber threads were created for performance reasons, a single order would be processed multiple times. A queue ensures that this behavior will not occur.

In the sample code, the Supplier class creates a queue consumer and the Customer class creates a queue producer. The code for both the order producer and consumer is almost identical to the code for the sale notification topic. Both classes create producers and consumers using the generic JMS interfaces and, therefore, have no hard-coded knowledge of the destination types. This means, for example, the sale notification topic can be changed to a queue in the future without requiring changes to producer or consumer code.

Conclusion

Most Java applications will benefit from the lightweight-messaging infrastructure that JMS-Light provides. Using the messaging features of JMS-Light can improve the design of your applications.

DDJ

Listing One

import javax.naming.*;
import javax.jms.*;

// Set the JMS Provider specific properties here.
// This can be read from a file at startup so that it's not hard coded
Hashtable props = new Hashtable();
props.put( Context.PROVIDER_URL, "rmi://localhost:1099/JndiServer" );
props.put( Context.INITIAL_CONTEXT_FACTORY, 
           "jms.jndi.rmi.RmiJndiInitialContextFactory");
Context jmsContext = new InitialContext(props);
// Get a ConnectionFactory object
ConnectionFactory connFactory 
jmsContext.lookup("ConnectionFactory");

import javax.jms.*;
import com.LJMS.*;
// Get the JMS-Light singleton instance
JMSLight ljms = JMSLight.getInstance();
// Get a ConnectionFactory object
ConnectionFactory connFactory = (ConnectionFactory)
ljms.lookup("ConnectionFactory");
// or...
QueueConnectionFactory queueConnFact = (QueueConnectionFactory)
ljms.lookup("QueueConnectionFactory");
TopicConnectionFactory topicConnFact = (TopicConnectionFactory)
ljms.lookup("TopicConnectionFactory");

public class JMSLightTopicConnection extends JMSLightConnection 
                                     implements TopicConnection
{
    public ConnectionConsumer createConnectionConsumer(Topic topic, ...)
    {
        // Call the base class' implementation. Topic extends
        // Destination so no cast is required
        return super.createConnectionConsumer(topic, ...);
    }
    public static ConnectionConsumer _createConnectionConsumer(...)
    {
        // this is where the real work is done...
    }
    // ...
}
// The base class:
public class JMSLightConnection implements Connection
{
    public ConnectionConsumer createConnectionConsumer(Destination dest,...)
    {
        // Because we're using 'instanceof', the downcast is safe
        if ( destination instanceof Queue ) {
            return JMSLightQueueConnection._createConnectionConsumer(
                (Queue)dest, ... );
        }
        else if ( destination instanceof Topic ) {
            return JMSLightTopicConnection._createConnectionConsumer(
                (Topic)dest, ... );
        }
    }
    // ...
}

public void onMessage(Message message) throws Exception
{
    if ( messageListener != null ) {
        // Asynchronous listener
        logger.log("JMSLightMessageConsumer: calling client onMessage()");
        messageListener.onMessage(message);
    }
    else {
        // Synchronous listener
        if ( ! clientBlocking )
            throw new Exception("Synchronous listener in wrong state");
        // Blocking listener, store message and set flag
        logger.log(this + " Storing message for queue " + destination );
        this.message = message;
        // Notify the blocking receiver thread and yield to it
        try {
            notifyAll();
            Thread.yield();
        } 
        catch ( Exception e ) {
            e.printStackTrace();
        }    
    }
}

public class OrderSystem
{
    private JMSLight ljms = JMSLight.getInstance();
    // ...
    public void createDestinations() throws Exception 
    {
        salesTopic = (Destination)ljms.createTopic("SaleItemTopic");
        orderQueue = (Destination)ljms.createQueue("OrderQueue");
    }
}

class Customer extends Thread implements MessageListener {
    // ...
    public void run() {
        salesTopic = (Destination)ljms.lookup( SALES_TOPIC_NAME );
        connFact   = (ConnectionFactory)ljms.lookup("ConnectionFactory");
        Connection conn = connFact.createConnection();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageConsumer saleConsumer = session.createConsumer(salesTopic);
        saleConsumer.setMessageListener(this);
    }
    public void onMessage( Message msg ) {
        // process the JMS message...
    }
}