JMS Point-to-Point Queue Model

Figure 12.13 depicts the basic JMS architecture elements that support point-to-point message queuing. The message queuing architecture is really an extension of the core JMS architecture with features that specifically focus on message queuing behavior. Connection factories, connections, sessions, message producers, message consumers, and endpoint destinations are all extended with point-to-point message queuing model interfaces. Thus, by understanding the core JMS architecture presented earlier, you can most rapidly understand the specific point-to-point message queuing specializations presented here. As we'll also illustrate later in this chapter, new to JMS v1.1, the more generic JMS architecture may also be used to implement point-to-point messaging without much consideration for the messaging domain-specific APIs illustrated in Figure 12.13.

JMS clients may use JNDI to obtain an initial reference to a named QueueConnectionFactory object. One of the QueueConnectionFactory.createQueueConnection() methods is used to create an instance of a QueueConnection object. The createQueueConnection() method can be called with a username and password or by using the parameterless version of the method with a default user identity assumed.

The QueueConnection interface is a type of Connection interface that represents a connection to a JMS point-to-point messaging queue service. The createQueueSession() method is called by JMS clients to create a QueueSession instance. Whether or not transactions are to be implemented by the QueueSession object is designated by a boolean parameter to createQueueSession(). Also, the acknowledgment mode is specified in the call to createQueueSession() using one of the static session identifiers, such as Session.AUTO_ACKNOWLEDGE, Session.CLIENT_ACKNOWLEDGE, or Session.DUPS_OK_ACKNOWLEDGE. The QueueSession interface extended from the Session interface implements various message-queuing–specific entity creation methods.

QueueSession.createQueue() creates an instance of a Queue object given a provider-specific name for that queue. Most service-provider implementations will provide other means for creating named queues, but this interface can be used by JMS clients for creating queues. The Queue interface encapsulates an interface to a queue destination using JMS's point-to-point messaging queue model and has a getQueueName() method to return the queue name. The QueueSession.createTemporaryQueue() method creates a TemporaryQueue object. The TemporaryQueue is deleted when its QueueConnection is closed.

The QueueSession.createBrowser() method creates an instance of a QueueBrowser object associated with a particular message queue. The QueueBrowser can be used to passively observe the live contents of a particular message queue without modifying the queue contents. If the QueueBrowser was created with a message selector, only those messages identified in the message selector expression will be viewable. The getEnumeration() method returns the list of messages associated with the queue and perhaps refined by the message selector.

The QueueSession.createSender() method creates a QueueSender message producer that is used to send messages to a Queue. Messages can be sent to a Queue using the various QueueSender.send() methods. Variations of these methods enable the delivery of messages to the QueueSender object's associated Queue or to a newly specified Queue object passed to the send() method. Delivery modes, priorities, and time to live for a message can also be specified during calls to QueueSender.send(). Messages to send to a Queue can be created using the various message-creation methods defined on the Session interface from which QueueSession extends.

The QueueSession.createReceiver() method creates a QueueReceiver message consumer used to receive messages from a Queue. A variation of the createReceiver() method also permits the specification of a message selector to filter out which messages are received by the QueueReceiver. Above and beyond the built-in message consuming methods provided by the QueueReceiver interface's base MessageConsumer interface, a QueueReceiver.getQueue() method returns a handle to the associated Queue.

A QueueRequestor helper class can also be used to simplify use of the JMS queue model for request and reply behavior. A QueueRequestor object is first created with a handle to a QueueSession and Queue object. The request method may then be used to send a Message to a queue destination and wait for a response in the form of a Message. To release and close any open distributed resources associated with the QueueRequestor and the underlying QueueSession, the JMS client invokes the close() method on the QueueRequestor.

Finally, as depicted in Figure 12.14, a set of interfaces exists to encapsulate transactional support for point-to-point message queuing. As mentioned before, the JMS client should not be coded to talk directly to such interfaces. Rather, the messaging server will use such interfaces to imbue a point-to-point message queue connection factory, connection, and session with JTA XA-compliant transaction management functionality. The interfaces in Figure 12.14 simply extend the interfaces presented in Figure 12.9 and provide method signatures specific to the message queue domain model.

Point-to-Point Message Queuing Sample Source Code

We present an example here to illustrate the use of JMS point-to-point message queuing. Our example implements a QueueSupplier that submits OrderItem objects to Queue objects managed by a WebLogic JMS service. A supplier implements a QueueConsumer that receives OrderItem objects from the Queue.

OrderItem

An OrderItem class implements a Serializable object encapsulating a simple order item request message. The OrderItem has an order identifier, order description, unit price, and an order quantity. Aside from a set of public attributes, an OrderItem.toString() prints the order request contents:

package com.wls8unleashed.jms;
 ...
public class OrderItem implements Serializable{
 private String itemID;
 private String itemDescription;
 private int quantity;
 private double price;

 public OrderItem( String itemID, String itemDescription,
             int quantity, double price){
  this.itemID = itemID;
  this.itemDescription = itemDescription;
  this.quantity = quantity;
  this.price = price;
 }

 public String toString(){
  return "OrderItem--> { "
      + "(ID, " + itemID + "), "
      + "(Description, " + itemDescription + "), "
      + "(Quantity, " + quantity + "), "
      + "(Price, $" + price + ") }";
 }

 public String getID(){ return itemID; }
 public String getDescription(){ return itemDescription;}
 public int getQuantity(){ return quantity;}
 public double getPrice(){ return price;}
}

OrderManager

The simple OrderManager class has a single getOrders() method that simply returns a Collection of OrderItem objects stubbed out for simplicity here as follows:

package com.wls8unleashed.jms;
 ...
public class OrderManager {
 public static Collection getOrders(){
  Vector dummyOrders = new Vector(2);

  OrderItem item1
   = new OrderItem("OID_2000", "Stale Yams", 20, 1.25);
  OrderItem item2
   = new OrderItem("OID_4000", "Elvis Wigs", 2, 19.95);

  dummyOrders.add(item1);
  dummyOrders.add(item2);

  return dummyOrders;
 }
}

Props

A special Props class is also used by these examples to read in information from the build.properties file for easy getter access by property name, as illustrated here:

package com.wls8unleashed.jms;
 ...
public class Props {
 private static String FILE_NAME = "build.properties";
 private static Properties props = null;

 private static void getInstance(){
  props = new Properties();
   ...
  props.load(new FileInputStream(FILE_NAME));
   ...
 }

 public static String get(String name){
  if(props == null){
   getInstance();
  }

  return (String) props.get(name);
 }

 public static Context getInitialContext(){
  Context ctx = null;

  // Get properties for naming service
  String jndiFactory = Props.get("jndi.factory");
  String providerURL = Props.get("jndi.provider.url");

  // Create hashtable of JNDI properties
  Hashtable env = new Hashtable();
  env.put(Context.INITIAL_CONTEXT_FACTORY, jndiFactory);
  env.put(Context.PROVIDER_URL, providerURL);

  // Return handle to JNDI context
   ...
  ctx = new InitialContext(env);
   ...

  return ctx;
 }
}

You'll note that the Props class also handles constructing a JNDI context for the sample code from within the getInitialContext() method by reading in JNDI properties from the build.properties file and constructing a handle to a JNDI InitialContext object. The JNDI properties of interest are the WebLogic JNDI factory class name and the URL of the WebLogic Server instance to which your JMS client examples will connect as exemplified by these entries in build.properties:

# JNDI Factory
jndi.factory=weblogic.jndi.WLInitialContextFactory

# JNDI Provider URL
jndi.provider.url=t3://localhost:7001/

QueueManager

A base QueueManager class is used by the QueueConsumer and QueueSupplier classes to encapsulate common JMS queue initialization functionality. The QueueManager constructor takes a JNDI Context object as an argument, looks up a handle to a JMS factory, creates JMS connections and sessions, looks up a handle to a queue name, and subsequently looks up a handle to a Queue object. As we'll see, the QueueConsumer and QueueSupplier inherit this behavior to initialize their handle to a Queue. The QueueManager class is illustrated here:

package com.wls8unleashed.jms;
 ...
public class QueueManager {

 protected QueueConnectionFactory queueConnectionFactory;
 protected QueueConnection queueConnection;
 protected QueueSession queueSession;
 protected Queue queue;

 public QueueManager(Context context){
    ...
   // Get JMS factory JNDI name
   String jmsFactoryName = Props.get("jms.factory.for.queue");

   // Create queue connection factory
   System.out.println("Looking up factory name: " + jmsFactoryName);
   queueConnectionFactory =
    (QueueConnectionFactory) context.lookup(jmsFactoryName);

   // Create queue connection to the factory
   System.out.println("Creating queue connection...");
   queueConnection = queueConnectionFactory.createQueueConnection();

   // Create session to the connection
   System.out.println("Creating queue session...");
   queueSession = queueConnection.createQueueSession(
                   false, Session.AUTO_ACKNOWLEDGE);

   // Get queue name
   String queueName = Props.get("queue.name");

   // Lookup handle to the Queue
   try {
    System.out.println("Looking up queue name: " + queueName);
    queue = (Queue) context.lookup(queueName);
   } catch (NamingException namingException) {
    // If not created, create new queue, and bind queue to name
    System.out.println("Didn't find the queue...so creating: " + queueName);
    queue = queueSession.createQueue(queueName);
    System.out.println("Binding queue: " + queueName);
    context.bind(queueName, queue);
   }
  }
  ...
 }
}

Note that the QueueManager reads the JMS factory and queue names passed into the JNDI lookup from the following properties set in the build.properties file:

# JMS Connection Factory for Queues
jms.factory.for.queue=UnleashedJMSFactory

# JMS Queue Name
queue.name=UnleashedQueue

We'll see shortly how such resources are established from within BEA WebLogic Server.

QueueSupplier

The QueueSupplier class extends the QueueManager class and implements a producer of a message to a queue. The QueueSupplier.main() method obtains a Collection of new BeeShirts.com order requests using the OrderManager.getOrders() call. For order item in the Hashtable of orders, an order request is sent within the QueueSupplier.sendOrder() method. The QueueSupplier.main() method is shown here:

public static void main(String[] args) {
    ...
   // Get some orders from the OrderManager class
   System.out.println("Getting a bunch of orders to send...");
   Collection orders = OrderManager.getOrders();

   // For each order, get order request and send to consumer
   Iterator it = orders.iterator();
   while (it.hasNext()) {
   // Get an OrderItem object to send
   OrderItem item = (OrderItem) it.next();

   // Create JNDI context
   Context context = Props.getInitialContext();

   // Create new QueueSupplier
   System.out.println("Creating new QueueSupplier...");
    QueueSupplier queueSupplier = new QueueSupplier(context);

    // Send the order
    System.out.println("Initiating order to send...");
    queueSupplier.sendOrder(item);

    // Close QueueSupplier
    System.out.println("Closing QueueSupplier...");
    queueSupplier.close();
   }
   ...
}

The QueueSupplier constructor is where all messaging queue initialization is performed on the supplier side. The QueueManager superclass constructor is first invoked to perform basic JMS queue object initialization. A QueueSender is then created from within the QueueSupplier. The QueueConnectionFactory, QueueConnection, QueueSession, Queue, and QueueSender objects are all saved as variables in either the QueueSupplier or base QueueManager class. The QueueSupplier constructor is shown here:

public QueueSupplier(Context context){
 // Call superclass QueueManager constructor
 super(context);
  ...
 // Create queue sender
 System.out.println("Creating queue sender...");
 queueSender = queueSession.createSender(queue);
   ...
}

The sendOrder() method called from main() first sends an object to a JMS queue. In our sample case, the object is always an OrderItem object. The sendOrder() method creates an ObjectMessage with the OrderItem object and sends it using a QueueSender:

public void sendOrder(OrderItem message) throws JMSException {
   System.out.println("Sending order message:" + message);

   // Create empty ObjectMessage on session
   System.out.println("Creating empty object message...");
   ObjectMessage sendingMessage = queueSession.createObjectMessage();

   // Start the queue connection
   System.out.println("Starting queue connection...");
   queueConnection.start();

   // Set the order object onto the message carrier
   System.out.println("Setting order item into message...");
   sendingMessage.setObject((Serializable) message);

   // Send the message
   System.out.println("Sending the order message...");
   queueSender.send(sendingMessage);
}

Finally, our QueueSupplier.close() method simply cleans up QueueSender, QueueSession, and QueueConnection resources:

public void close() throws JMSException{
 queueSender.close();
 queueSession.close();
 queueConnection.close();
}

QueueConsumer

Our sample QueueConsumer represents a vendor that the order producer application, represented by the QueueSupplier, talks to. The QueueConsumer has a queue that the QueueSupplier uses to place an order request with the vendor. The QueueConsumer.main() method first creates a JNDI context via a call to Props.getInitialContext(). The QueueConsumer is then constructed with the JNDI context as a parameter. The QueueConsumer then receives messages until the user exits the program. Finally, QueueConsumer.close resources are cleaned up, along with the JNDI context. The QueueConsumer.main() method is shown here:

public static void main(String[] args){
    ...
  // Create JNDI context
  Context context = Props.getInitialContext();

  // Create new QueueConsumer
  System.out.println("Creating new QueueConsumer...");
  QueueConsumer queueConsumer = new QueueConsumer(context);

  System.out.println("QueueConsumer is ready to receive messages...");

  // Receive messages until user quits from program or quit flags true.
  synchronized (queueConsumer) {
   while (!queueConsumer.quitFromReceiving) {
    try {
     // Wait for messages
     queueConsumer.wait();
    } catch (InterruptedException interruptedException) {
       System.out.println("Interruption!");
       interruptedException.printStackTrace();
    }
    }
   }

   // Close up resources when done and exit
   queueConsumer.close();
   context.close();
   System.exit(0);
   ...
}

The QueueConsumer constructor invokes its base class QueueManager constructor to perform the bulk of JMS queue object initialization. The QueueConsumer then creates a QueueReceiver and registers itself as a MessageListener with the QueueReceiver. The QueueConnection thread is then initialized to start receiving messages. The QueueConsumer constructor initialization is shown here:

public QueueConsumer(Context context)
  throws NamingException, JMSException {

 // Call superclass QueueManager constructor
 super(context);

 // Create queue receiver
 System.out.println("Creating queue receiver...");
 queueReceiver = queueSession.createReceiver(queue);

 // Register QueueConsumer as MessageListener
 System.out.println("Setting message listener...");
 queueReceiver.setMessageListener(this);

 // Start receiving messages
 System.out.println("Starting queue connection...");
 queueConnection.start();
}

Because the QueueConsumer is a message listener by virtue of implementing the MessageListener interface, it must implement the onMessage(Message) method. The QueueConsumer.onMessage() is called when the WebLogic JMS service provider receives an OrderItem from the QueueSupplier. The QueueConsumer object's onMessage() method delegates the handling of received messages to a MessageHandler object as illustrated here:

public void onMessage(Message message) {
 MessageHandler handler = new MessageHandler();
 handler.onMessage(message);
}

MessageHandler

The MessageHandler class implements one method: onMessage(Message). The onMessage() method will first display information from the received Message header and then display the message body information, depending on whether it is a TextMessage or an ObjectMessage as illustrated here:

public void onMessage(Message message) {
  ...
 // Print information about the message
 System.out.println("\nReceived message...");
 System.out.println("MessageID :"
           + message.getJMSMessageID()
           + " for "
           + message.getJMSDestination());

 System.out.print("Message expiration info: ");
 if (message.getJMSExpiration() > 0) {
  System.out.println(new Date(message.getJMSExpiration()));
 } else {
  System.out.println(" Never expires");
 }

 System.out.println("Priority :" + message.getJMSPriority());

 System.out.println("Mode  : "
      + (message.getJMSDeliveryMode() == DeliveryMode.PERSISTENT
      ? "PERSISTENT"
      : "NON_PERSISTENT"));

 System.out.println("Reply to :   " + message.getJMSReplyTo());

 System.out.println("Message type : " + message.getJMSType());

 if (message instanceof TextMessage) {
  String receivedMessage = ((TextMessage) message).getText();
  System.out.println("Received text message:" + receivedMessage);
 } else if (message instanceof ObjectMessage) {
  String receivedMessage = message.toString();
  System.out.println("Received object message:" + receivedMessage);
 }
  ...
}

WebLogic JMS Server Configuration

Now that we've seen the JMS source code that produces messages to a queue and consumes messages from a queue, let's examine how to configure WebLogic Server to manage such queues. The first step is to start the WebLogic Server administration console and step into the JMS servers configuration screen from within ServicesJMSServers as shown in Figure 12.15.

You then must click on Configure a New JMS Server to begin configuring a new JMS server instance to run inside of your WebLogic instance. You then give the server a new name and click the Create button at the bottom of the screen, which results in the creation of a server configuration as illustrated in Figure 12.16. Note that we've created a server named UnleashedJMS for this example. A few other options may also be configured for your server at this stage. We'll discuss some of these options later in this section.

By clicking on the Thresholds and Quotas tab for your JMS server configuration, as illustrated ion Figure 12.17, you can also fine-tune how the server instance handles heavy load or large messages for your JMS server. Message and byte maximums and minimums for the JMS server can all be configured from this screen. In addition to helping with scalability, this also is useful for protecting against denial-of-service attacks.

At this point, you've only configured the JMS server but have yet to deploy it as an active server ready to manage JMS messages. By clicking on the JMSServers browser selection in the left browser pane, you can see those JMS servers that have been configured and are ready to deploy. To deploy your JMS server to the active state, the Target and Deploy screen is used. In Figure 12.18 we can select a particular WebLogic instance to which your JMS server is to be deployed. Upon deployment, you can verify that your JMS server is active by selecting your server's name in the left browser pane beneath the JMSServers browser selection, clicking on the server's Monitoring tab in the center pane, and then selecting Monitor Active JMS Servers to see that your JMS server is active and running.

WebLogic JMS Connection Factory Configuration

Recall from our JMS queue code example that we set the following factory name in our build.properties file used by the code to look up a handle to a QueueConnectionFactory object:

# JMS Connection Factory for Queues
jms.factory.for.queue=UnleashedJMSFactory

By default, WebLogic provides two default connection factories that can be used to look up connections: one for standard connections and one for XA-compliant connections. The weblogic.jms.ConnectionFactory and weblogic.jms.XAConnectionFactory JNDI names may be used to look up handles to these default connection factories. You may also configure your own connection factory, as we will do for our example. You begin this configuration process by clicking on the JMSConnection Factories option in the left browser pane as illustrated in Figure 12.19.

You then select the Configure a New JMS Connection Factory option in the center pane, which yields the JMS connection factory configuration screen shown in Figure 12.20. Here we establish the name of our connection factory known to WebLogic Server and the JNDI name of the connection factory. The Transactions and Flow Control tabs associated with the JMS factory configuration screen also enable you to respectively specify transactional behavior and maximum rate at which messages can be produced via the connections created over the connection factory.

After configuring a JMS connection factory, you must generally then target a particular server or cluster to deploy the factory. The Target and Deploy tab shown in Figure 12.21 can be used to deploy active instances of the connection factories that you create. Note that whereas the JMS server can be targeted to a single WebLogic server instance, the connection factories can be targeted to many.

WebLogic JMS Queue Configuration

After configuring a JMS server and connection factory, the next step in creating a concrete JMS application is to configure and deploy a JMS destination associated with a JMS server. In our first example, we use JMS queues to communicate between a consumer and producer. By clicking on the Destinations link in the left browser pane beneath a particular JMS server name in the JMSServers list, you yield the screen displayed in Figure 12.22.

For our first example, we want to click on Create a New JMS Queue in the center pane of the screen shown in Figure 12.22. The screen shown in Figure 12.23 is then used to perform the basic configuration for a queue. The queue name known to the JMS server and the JNDI name for the queue must be established along with other basic configuration information, such as the cluster-ability of the queue's JNDI name and the persistence of the queue. Note that we use the JNDI name UnleashedQueue in Figure 12.23 to be onsistent with the name we assume for our queue sample code as read in via the build.properties file:

# JMS Queue Name
queue.name=UnleashedQueue

A set of additional configuration screens also exists as tabs beneath the queue's configuration tab in the center pane. Configuration screens exist to define thresholds and quotas for message storage, override values for message priorities and delivery modes, message expiration policies, and message redelivery policies. The center pane also has a tab named Monitoring that enables you to monitor all active destinations, including the queue that we just created. Figure 12.24 demonstrates some of the useful information fields that can be observed for an active queue.

JMS Queue Sample Execution

Now that you have source code demonstrating a queue consumer and producer as well as a WebLogic Server instance hosting a queue, you're ready to execute the sample code presented earlier. The first step is to run the Ant script for this chapter's code by typing ant at the command line in the directory where you have deposited the source code. The default all Ant target will compile all the source code examples and generate two scripts: runqueue-consumer and runqueue-supplier. The scripts will have either a .bat or .sh extension depending on whether you are running the examples on a Windows or Unix platform, respectively.

Be sure to check the example's build.properties file as well. Aside from ensuring that the JMS libraries are properly referenced inside the CLASSPATH for the execution scripts, the JMS server URL, connection factory name, and queue name must all be set as you've configured for your WebLogic Server environment.

Then, in one command-line window, execute the runqueue-consumer script to start the sample QueueConsumer. In a separate window, you need to execute the runqueue-supplier script to start the QueueSupplier. The QueueSupplier publishes messages to the JMS queue that we configured. The messages are then asynchronously delivered to the QueueConsumer by the WebLogic JMS server.

Although this example illustrates how standalone J2SE clients using the JMS libraries can send and receive messages, the JMS producer code can also run inside of a J2EE container called from either a Java servlet, JavaServer Page (JSP), or Enterprise JavaBean (EJB). When calling JMS producer code from within a J2EE component, JNDI is used to look up handles to connection factories and destinations as with JMS producer code operating outside of a J2EE container. However, a call to the parameter-less form of the InitialContext constructor may be used to connect with the container's JMS managed resources. Furthermore, as we'll also see later in this book, message-driven EJBs can act as messaging consumers while leveraging the advantages of EJB life cycle management.