Establishing JMS Connections

 

This section explains how to establish JMS connections, including the following topics:

• Overview of JMS

• WebLogic Process Integrator JMS Destinations

• Connecting to JMS

• Receiving Messages Asynchronously

• Generating Message-Driven Beans for Multiple Event Queues

• Guaranteeing Message Delivery

• Guaranteeing Sequential Processing of Messages

• Example of Connecting to a JMS Topic

Note: For more information about JMS, see Programming WebLogic JMS in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html

Or the JavaSoft JMS specification version 1.0.2, from Sun Microsystems, Inc., available at the following URL:

http://www.javasoft.com/products/jms/docs.html

 

---

Overview of JMS

WebLogic Server, on which WebLogic Process Integrator runs, implements the Java Message Service (JMS), which supports the transmission of XML content. WebLogic Process Integrator uses WebLogic JMS for communicating worklist, time, and event notifications; and error and audit messages.

The following figure shows how JMS enables communication between WebLogic Process Integrator running instances and external client applications through the exchange of XML messages.

Figure 6-1 Overview of JMS

 

As shown here, an XML event issued by a JMS application is:

1. Delivered to a predefined event queue (for example, eventQueue, described in the next section)

2. Directed to the appropriate message-driven bean for processing.

3. Delivered to the running instance.

The following sections describe the JMS destinations that are used by WebLogic Process Integrator, and explain how to connect to and use them. An example of connecting to a JMS topic, wlpiNotify, is also provided.

For more information about JMS, see Programming WebLogic JMS in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html

 

---

WebLogic Process Integrator JMS Destinations

The following table summarizes the JMS destinations that are used by WebLogic Process Integrator.

Table 6-1 WebLogic Process Integrator JMS Destinations

This JMS destination . . .	Of JMS type . . .	Is the destination for . . .
wlpiAudit	Topic	Audit messages.
wlpiError	Topic	Error messages. For more information, see Monitoring Workflow Exceptions.
eventQueue	Queue	Events. All incoming messages are processed via message-driven beans. A message-driven bean is an EJB that acts as a message consumer in the WebLogic JMS messaging system. As with standard JMS message consumers, message-driven beans receive messages from a JMS queue or topic, and perform business logic based on the message contents. In addition, message-driven beans support concurrent processing for JMS destinations. The system administrator can define multiple event queues, if required, as described in Connecting to JMS. In this case, the system administrator must also generate the associated message-driven beans using the utility described in Generating Message-Driven Beans for Multiple Event Queues, and the application developer must specify the appropriate destination for a particular event. You can guarantee the delivery and sequential processing of messages using the methods described in Guaranteeing Message Delivery and Guaranteeing Sequential Processing of Messages, respectively. Note: For backward compatibility, the TopicRouter bean automatically reroutes any messages bound for the wlpiEvent topic, which is supported by WebLogic Process Integrator Release 1.2.1 and earlier releases, to the eventQueue queue.
wlpiInit	Topic	WebLogic Process Integrator initialization tasks.
wlpiNotify	Topic	Worklist notifications. Enables a worklist client to refresh its display dynamically and obtain task object data via the delivered com.bea.wlpi.common.TaskInfo object. For messages posted to the wlpiNotify topic, the following message properties must be defined to ensure delivery to appropriate subscribers: orgId: string specifying the ID of the organization assigneeId: string specifying the ID of the assignee (user or role) role: Boolean flag specifying whether or not the assignee is a role (true, false) action: string specifying which of the following actions has been performed on the task: - assigned: task assigned to user - update: task state has changed - remove: task removed from list of user tasks The worklist client can define a selector to filter messages based on these message properties. For an example of using a selector to filter messages, see Example of Connecting to a JMS Topic. For more information about JMS, see Programming WebLogic JMS in the BEA WebLogic Server document set, available at the following URL: http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html
wlpiTimer	Topic	Time processor triggers (internal use only).

 

 

---

Connecting to JMS

To connect to any of the JMS destinations defined in the table WebLogic Process Integrator JMS Destinations (and to be able to post and receive XML messages), the WebLogic Server administrator must perform the following steps for each destination:

1. Look up a JMS connection factory in JNDI.

   A system administrator defines and configures one or more connection factories, and WebLogic Server adds them to the JNDI space during startup. A connection factory encapsulates connection configuration information, and enables JMS applications to create a connection.

2. Create a connection using the connection factory.

   A connection represents an open communication channel between an application and the messaging system. It is used to create a session for producing and consuming messages.

3. Create a session using the connection.

   A session defines a serial order in which messages are produced and consumed, and can create multiple message producers and message consumers.

4. Look up destinations in JNDI.

   A destination can be either a queue or a topic, encapsulating the address syntax for a specific provider. An administrator defines and configures the destinations and WebLogic Server adds them to the JNDI space during startup.

   On the client side, destinations are handles to the objects on the server. The methods return only the destination names. To access destinations for messaging, you create message producers and consumers that can attach to destinations.

5. Create message producers and message consumers using session and destinations.

   A message producer sends messages to a queue or topic. A message consumer receives messages from a queue or topic.

6. Perform one of the following steps:

   1. If you are creating message producers, create the message object.

      A message encapsulates the information exchanged by applications.

   2. If you are creating message consumers, you may want to register an asynchronous message listener, as described in Receiving Messages Asynchronously.

7. Start the connection.

For more information about JMS, see Programming WebLogic JMS in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html

 

---

Receiving Messages Asynchronously

To receive messages asynchronously from a destination, you must register an asynchronous message listener by performing the following steps:

1. Implement the javax.jms.MessageListener interface, which includes an onMessage() method.

   Note: This step can also be accomplished using message-driven beans for WebLogic Server 6.0 or greater. For more information about EJBs, see Programming WebLogic Enterprise JavaBeans in the BEA WebLogic Server document set, available at the following URL:
   
   http://download.oracle.com/docs/cd/E13222_01/wls/docs60/ejb/index.html

2. Set the message listener using the following javax.jms.MessageConsumer method and passing the listener information as an argument:

   public void setMessageListener(
     javax.jms.MessageListener listener
   ) throws javax.jms.JMSException

For more information about JMS, see Programming WebLogic JMS in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html

 

---

Generating Message-Driven Beans for Multiple Event Queues

As noted previously, the system administrator can define multiple event queues, if required, as described in Connecting to JMS. In this case, the system administrator must also generate the associated message-driven beans.

To generate message-driven beans for multiple event queues, use the com.bea.wlpi.util.MDBGenerator utility, as follows:

java com.bea.wlpi.util.MDBGenerator 
    -queue qname [-min minimum -max maximum] [-order number] 
    [-validate] [-help]

The following table lists the MDBGenerator utility arguments.

Table 6-2 MDBGenerator Utility Arguments

Argument	Description
-q[ueue] qname	Queue name. qname specifies a valid queue name. This argument is required.
-m[in] minimum	Minimum number of message listeners allocated to handle unordered messages. Unordered messages are those that do not need to be processed in a specific order, as described in Guaranteeing Sequential Processing of Messages. minimum specifies an integer value that must be less than the maximum value. This argument is optional and defaults to 0.
-m[ax] maximum	Maximum number of message listeners allocated to handle unordered messages. maximum specifies an integer value that must be less than or equal to 100, and greater than the minimum value specified. This argument is optional and defaults to 5.
-o[rder] ordernum	Number of message listeners allocated to handle ordered messages. Ordered messages are those that require messages to be processed in a specific order, as described in Guaranteeing Sequential Processing of Messages. ordernum specifies a prime number less than or equal to 31. This argument is optional and defaults to 0.
-validate	Flag specifying that you want to validate all messages received on the queue against the DOC-TYPE tag in the XML JMS message. This argument is optional.
-help	Flag specifying that you want to display command usage syntax. This argument is optional.

 

This utility generates a message-driven bean deployment for a specific queue as a jar file named qname-mdb.jar, where qname specifies the associated queue name.

To deploy the message-driven beans on WebLogic Server, edit the ejb-jar.xml and weblogic-ejb.xml files to associate the EJB with a configured JMS destination. For more information, see Programming WebLogic Enterprise JavaBeans in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/ejb/index.html

 

---

Guaranteeing Message Delivery

You can guarantee message delivery using addressed messages. Addressed messages persist an incoming event message until it is consumed by all recipients or a specified expiration time (time-to-live) elapses, whichever occurs first. You can guarantee message delivery on a workflow instance or template basis.

To guarantee message delivery, the sending and receiving applications must perform the following steps:

1. The sending application (message producer) must define the following information:

   1. The WLPIInstanceIds and/or WLPITemplateNames fields as part of the message header, which defines the message consumer addresses.

      Specifically, these fields specify the IDs of the instances and/or templates that you want to receive the message. You can define the WLPIInstanceIds and WLPITemplateNames JMS header fields using the following javax.jms.Message class method:

      public void setStringProperty(
        java.lang.String name, 
        java.lang.String value
      ) throws javax.jms.JMSException

      For example, to define the JMS header fields for the msg message instance, use the following methods:

      msg.setStringProperty("WLPIInstanceIds", 12345);
      msg.setStringProperty("WLPITemplateNames", 12345);

      These fields must be specified as String values. You can specify multiple ids for each field, separated by a comma. You should target only those instances and/or templates that have already been instantiated.

      Note: If you want to address messages to a template, avoid using commas in the template name.

      JMS header fields are always transmitted with the message, and are available to the message consumers (including the message-driven beans). For more information about defining JMS message header fields, see "Setting and Browsing Message Header and Property Fields" in Programming WebLogic JMS, in the BEA WebLogic Server documentation set, at the following URL:

      http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/implement.html#1095413

   2. The time-to-live value as a parameter when sending the message which specifies the number of seconds to persist an addressed message. The processing engine persists the message until the time-to-live value expires or the message has been consumed by all addressed recipients, whichever occurs first. This parameter must be specified as an integer value. For more information about sending messages, see "Setting and Browsing Message Header and Property Fields" in Programming WebLogic JMS, in the BEA WebLogic Server documentation set, at the following URL:

      http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/implement.html#1022980

      The time-to-live value can also be set via the Studio Post XML Event dialog box, as described in "Integration Actions" in Defining Actions in Using the BEA WebLogic Process Integrator Studio.

      Once set, you can view the resulting expiration time via the JMSExpiration JMS header field using the following javax.jms.Message class method:

      public long getJMSExpiration(
      ) throws javax.jms.JMSException

      For more information about viewing JMS message header fields, see "Setting and Browsing Message Header and Property Fields" in Programming WebLogic JMS, in the BEA WebLogic Server documentation set, at the following URL:

      http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/implement.html#1095413

2. The receiving application (message consumer) defines a JMS message selector to filter the messages that it needs to receive.

   A message selector is a Boolean expression. It consists of a String with a syntax similar to the where clause of an SQL select statement. For example:

   WLPIInstanceId=12345

   For more information about defining JMS message selectors, see "Filtering Messages" in Programming WebLogic JMS, in the BEA WebLogic Server documentation set, at the following URL:

   http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/implement.html#1186875

The following figure illustrates how to guarantee message delivery using addressed messaging.

Figure 6-2 Guaranteed Message Delivery

 

Note the following in this figure:

• Addressed messages are delivered to the instance or template specified by the JMS header field in the sending application (WLPIInstanceIds or WLPITemplateNames).

• Addressed messages that cannot be delivered because a recipient is unavailable are persisted until the recipient becomes available or the time-to-live value (JMSExpiration header field value) is reached, whichever occurs first.

 

---

Guaranteeing Sequential Processing of Messages

One advantage of message-driven beans is that incoming messages can be processed in parallel by random message-driven bean instances. In this case, however, you cannot guarantee the order in which messages are processed.

If the order in which messages are received and processed is significant, you can guarantee that order by directing all messages to the same message-driven bean instance.

To guarantee that messages are delivered to the same message-driven bean instance and processed in a sequential order, the sending and receiving applications must preform the following steps:

1. Before sending the message, the sending application (message producer) must define an order key (WLPIOrderKey) as a field in the JMS message header. The order key value must be an integer, such as, the instance ID.

   You can define the WLPIOrderKey JMS header field using the following javax.jms.Message class method:

   public void setIntProperty(
     java.lang.String name, 
     int value
   ) throws javax.jms.JMSException

   For example, to define the WLPIOrderKey JMS header field for the msg message instance, use the following method:

   msg.setIntProperty("WLPIOrderKey", 1);

   Related messages that require sequential processing should be assigned the same order key value.

   JMS header fields are always transmitted with the message, and are available to the message consumers (including the message-driven beans). For more information about defining JMS message header fields, see "Setting and Browsing Message Header and Property Fields" in Programming WebLogic JMS, in the BEA WebLogic Server documentation set, at the following URL:

   http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/implement.html#1095413

2. The receiving application (message consumer) must define a JMS message selector to filter the messages that it needs to process based on the WLPIOrderKey value.

   A message selector is a Boolean expression. It consists of a String with a syntax similar to the where clause of an SQL select statement.

   For example: WLPIOrderKey=1

   For more information about defining JMS message selectors, see "Filtering Messages" in Programming WebLogic JMS, in the BEA WebLogic Server documentation set, at the following URL:

   http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/implement.html#1186875

Note: All message groups that are to be processed in sequential order must be sent to the same JMS event queue.

All messages with a specific order key are processed by the same message-driven bean instance, in the order received, guaranteeing the sequential processing of the messages.

The following figure illustrates the sequential processing of messages using order keys.

Figure 6-3 Guaranteed Sequential Processing of Messages

 

Note the following in this figure:

• All messages that are to be processed in sequential order are sent to the same JMS event queue, eventQueue, or other user-defined queue.

• All messages with the WLPIOrderKey header field set to 1 are processed by Message-Driven Bean 1; messages with the WLPIOrderKey header field set to 2 are processed by Message-Driven Bean 2; and so on.

• All messages in which the WLPIOrderKey header field information is not defined are processed by a pool of message-driven beans.

• Each XML message may affect one or more workflow templates and/or instances.

 

---

Example of Connecting to a JMS Topic

This section provides an example showing how to connect to a JMS destination, in this case, the Worklist notification topic, wlpiNotify. The wlpiNotify topic is used by a Worklist client to refresh its display dynamically upon receipt of a message. In this example, a message selector (filter) is used to receive notification. Subsequently, if the properties of the received message match the current values of the organization, role, user, and action, the message selector is used to refresh the display.

Each section of code is described in detail. For more information about JMS, see Programming WebLogic JMS in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html

Define the required variables, including the JNDI context, JMS connection factory, and topic static variables:

protected static final String JNDI_FACTORY= "weblogic.jndi.WLInitialContextFactory";
protected static final String JMS_FACTORY= "javax.jms.TopicConnectionFactory";
protected static final String NOTIFY_TOPIC="com.bea.wlpi.wlpiNotify";

Create all the objects necessary for sending messages to a JMS topic:

private TopicConnectionFactory tconFactory;
private TopicConnection tcon;
private TopicSession tsession;
private Topic topic;
private TopicSubscriber tsubscriber;

Set up the JNDI initial context, as follows:

try {
   Hashtable env = new Hashtable();
   env.put(Context.PROVIDER_URL, wlpi.getUrl());
   env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
   env.put("weblogic.jndi.createIntermediateContexts", "true");

   Context ctx = new InitialContext(env);

Step 1

Look up a connection factory using JNDI.

    tconFactory = (TopicConnectionFactory) ctx.lookup(JMS_FACTORY);

Step 2

Create a connection using the connection factory.

    tcon = tconFactory.createTopicConnection();

Step 3

Create a session using the connection. The following method defines the session as non-transacted and specifies that messages will be acknowledged automatically.

For more information about setting session transaction and acknowledge modes, see Programming WebLogic JMS, in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html

tsession = tcon.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

Step 4

Look up the destination (topic) using JNDI.

    topic = (Topic)ctx.lookup(NOTIFY_TOPIC);

Step 5

Create a reference to a message producer (topic publisher) using the session and destination (topic). If a subscriber already exists, it is closed. Information about the organization and participant is used as the selector. For more information about defining selectors, see Programming WebLogic JMS, in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html

In the following example, wlpi represents an instance of the com.bea.wlpi.client.common.WLPI class.

if (tsubscriber != null) tsubscriber.close();
String selector = "orgId = '" +
wlpi.getWorklist().getActiveOrganization() + "' AND assigneeId = '" +  
   wlpi.getUserId() + "' AND NOT role" + " AND action = 'assigned'";
tsubscriber = tsession.createSubscriber(topic, selector, false);

Step 6

Register an asynchronous message listener.

    tsubscriber.setMessageListener(this);

Step 7

Start the connection.

    tcon.start();
  }

When a message is delivered to the topic session, it is passed to the onMessage() method, which you must implement as described in Receiving Messages Asynchronously.

The following provides an example of a message listener client that uses the notification to refresh its tasklist displays.

Note: For AWT/Swing applications, you must not update the user interface on the same thread as the call to the onMessage() method call, or you will encounter deadlocks and/or exceptions. You should marshall the call onto the AWT Event Dispatcher Thread. The Swing utilities provide the invokeLater() method to enqueue a runnable object on the AWT Event Dispatcher Thread, and you can include the update logic in the runnable object run() method.

import javax.swing.SwingUtilities;
public void onMessage(Message msg) {
   String action;
   TaskInfo task;
   try {
   // We're only interested in ObjectMessages with a TaskInfo
   // payload.
      if (!(msg instanceof ObjectMessage))
         return;
      Object data = ((ObjectMessage)msg).getObject();
      if (!(data instanceof TaskInfo))
         return;
      action = msg.getStringProperty("action");
      task = (TaskInfo)data;
   } catch (JMSException e) {
      e.printStackTrace();
      return;
   }

   if (action.equals("update"))
      taskUpdated(task);
   else if (action.equals("remove"))
      taskDeleted(task);
}

private void taskDeleted(final TaskInfo task) {
   SwingUtilities.invokeLater(new Runnable() {
      public void run() {
         vTasks.remove(task);
         int row = vDisplayedTasks.indexOf(task);
         if (row != -1) {
            vDisplayedTasks.remove(row);
            model.fireTableRowsDeleted(row, row);
         }
      }
   });
}

For more information about JMS, see Programming WebLogic JMS, in the BEA WebLogic Server document set, available at the following URL:

http://download.oracle.com/docs/cd/E13222_01/wls/docs60/jms/index.html