Chapter 29 Using the Message Service

Developing EAServer messaging service applications

To develop an EAServer messaging service application, use the EAServer CORBA APIs, which enable you to configure and use the message service within a client application or EAServer component. See "EAServer message service CORBA API" for more information. Creating message service applications can involve these steps:

Obtaining CtsComponents::MessageService object references

Before a CORBA client can send, publish, or receive messages, it must obtain a MessageService object reference. This code sample performs the setup required for a message service client application:

org.omg.CORBA.*;
import java.util.*;
import SessionManager.*;
import CtsComponents.*;
import java.lang.Object;

public class ReceiveTest
{
   public static void main(String[] args)
   {
      new ReceiveTest().test(args);
   }

   public void test(String[] args)
   {
      Properties props = new Properties();
      props.put("org.omg.CORBA.ORBClass",
                "com.sybase.CORBA.ORB");

      ORB orb = ORB.init((String[])null, props);

      Manager manager =
         ManagerHelper.narrow(orb.string_to_object(
                            "iiop://localhost:9000"));

      Session session =
         manager.createSession("jagadmin", "");

      MessageService cms =
         MessageServiceHelper.narrow(session1.create(
                     "CtsComponents/MessageService"));

      MessageQueue mq = 
         cms.getMessageQueue("test", "",
                          REQUIRES_ACKNOWLEDGE.value);
...

Creating message consumers

Message consumers can be either a message queue or topic. You can create message consumers two ways. The recommended way is to use Jaguar Manager to create and configure message queues and topics, so their configuration properties are stored in a database. See Chapter 8, "Setting up the Message Service," in the EAServer System Administration Guide for information on how to do this. You can also create message queues and topics programmatically; for example:

void create(MESSAGE_QUEUE, "QueueName");
void create(MESSAGE_TOPIC, "TopicName");

See $JAGUAR/html/ir/CtsComponents__MessageService.html for more information on configuring queues and topics within your application.

Creating message selectors

You can use selectors to specify which messages you want delivered to a message queue. Once you add a selector to a queue, the message service delivers only those messages whose message topic matches the selector. You can create message selectors using Jaguar Manager--see Chapter 8, "Setting up the Message Service," in the EAServer System Administration Guide. You can also create message selectors programmatically. This example illustrates how to add a message selector to MyQueue to request notification of a new stock value:

MessageService cms = getMessageService();
cms.addSelector("MyQueue", 
   "topic = 'stock.SY' AND value > 50");

Creating thread pools

The threads in a thread pool provide asynchronous client and component notification. You can create thread pools using Jaguar Manager--see Chapter 8, "Setting up the Message Service," in the EAServer System Administration Guide. You can also create a thread pool within your application. This example creates a thread pool and sets the thread values of readers, writers, and workers to "0". If the thread pool already exists, its configuration is unchanged.

void create (THREAD_POOL, "Thread_Pool_Name" );

To use a thread pool for client notification, set the thread values of readers to "3", writers to "2", and workers to "0". To use a thread pool for component notification, set the thread values of both readers and writers to "0", and set the value of workers to "1". If you want to enable parallel message processing for component notification, set workers to a value greater than "1".

This code fragment sets the value of workers in the system.tp1 thread pool to "1":

props = _cms.getProperties(THREAD_POOL.value,
                           "system.tp1");

for (int i = 0; i < props.length; i++)
{
   Property p = props[i];
   if (p.name.equals("workers"))
   {
      if (p.value.longValue() != 1)
      {
         p.value.longValue(1);
         _cms.setProperties(THREAD_POOL.value, 
                            "system.tp1", props);
         break;
      }
   }
}

Implementing and installing message listeners

Message listeners allow you to receive messages asynchronously. Once you have implemented a listener, install it on a message consumer. When a message is delivered to the message consumer, the listener can send the message to other consumers or notify one or more components.

Message listeners are EAServer components that implement the CtsComponents::MessageListener interface, which contains only this onMessage method:

void onMessage(in CtsComponents::Message msg);

You can install message listeners two ways. The recommended way is to use Jaguar Manager to install a message listener on a message queue or topic--see Chapter 8, "Setting Up the Message Service," in the EAServer System Administration Guide. You can also install a message listener within your application.

This example installs a message listener implemented by the EAServer component "MyPackage/MailService" on the message queue "MyClient:email":

MessageService cms = getMessageService();
cms.addListener("MyClient:email",
                "MyPackage/MailService");

When you create a message listener, you can optionally provide the name of a thread pool. The thread pool must have one or more worker threads. A thread pool with multiple worker threads enables the message listener to process multiple messages at the same time. If you do not specify the name of a thread pool, the message listener uses the system default thread pool, which has a single worker thread.

This code sample adds a message listener and specifies "MailPool" as the name of the thread pool:

cms.addListener("MyClient:email",
                "MyPackage/MailService[MailPool]");

Sending messages

To send a message, you must specify the destination message queue. The message service notifies listeners that are registered for the queue and the message remains in the queue until it is received and acknowledged.

Figure 29-1 illustrates the message flow that occurs when a client or component sends a message.

In this example, we notify a client of a completed order by creating a new message, constructing the message text, and sending the message to the client's queue:

public void notifyOrder(MessageService cms, 
                        String queue, 
                        int orderNo, 
                        String product)
{
  String time = new java.util.Date().toString();
  String text = "Order " + orderNo + " for product " 
                + product + " was completed at " + time;

  Message msg = new Message();
  msg.key = cms.getMessageKey();
  msg.props = new Property[2];
  msg.props[0] = new Property("orderNo", 
                              new PropertyValue());
  msg.props[0].value.longValue(orderNo);
  msg.props[1] = new Property("product", 
                              new PropertyValue());
  msg.props[1].value.stringValue(product);
  msg.replyTo = "";
  msg.text = text;
  cms.send(queue, msg, PERSISTENT.value);
}

Publishing messages

When you publish a message, a copy is sent to all topic subscribers that have a message selector registered with the specified topic. Figure 29-2 illustrates the message flow when a client or component publishes a message.

This example illustrates how to publish a message that notifies clients of changes in a stock value. Set the message topic, define the message text, set the message key, define and set message properties, and call publish:

public void notifyStockValue(MessageService cms,
                             String stock,
                             double value)
{
   String topic = "StockValue." + stock;
   String time = new java.util.Date().toString();
   String text = time + ": The stock " + stock + 
                        " has value " + value;

   Message msg = new Message();
   msg.key = cms.getMessageKey();
   msg.props = new Property[2];
   msg.props[0] = new Property("stock", 
                               new PropertyValue());
   msg.props[0].value.stringValue(stock);
   msg.props[1] = new Property("value", 
                               new PropertyValue());
   msg.props[1].value.doubleValue(value);
   msg.replyTo = "";
   msg.text = text;
   cms.publish(topic, msg, 0);
}

To publish a persistent message using the default priority (4) and timeout (never expires) values, use this syntax:

publisher.publish(textMsg);

Receiving messages

You can receive messages either synchronously or asynchronously. To receive messages synchronously (get all the messages at one time), call the receive method for the message consumer. This code sample gets all the messages from the queue, then, for each message, it prints a message receipt and acknowledges the message:

Message[] seq = mq.receive(0, DEFAULT_TIMEOUT.value);

for (int m = 0; m < seq.lengeth; m++;)
{
   Message msg = seq[m];
   System.out.println("Received message: " + msg.text);
   mq.acknowledge(msg.key);
}

To receive messages asynchronously, implement a message listener and install it on the message consumer, either a topic or a queue. See "Implementing and installing message listeners".

Subscribing to scheduled messages

The message service can generate and send regularly scheduled messages to message queues. You can subscribe to scheduled messages by installing a listener on a queue and subscribing to a topic that defines the times you want to be notified. In this example, we add a listener to "MyQueue" and subscribe to the topic "second:30", which causes the message service to send a message to MyQueue at 30 seconds past each minute:

cms.addListener("MyQueue", "MyPackage/MyComp");
cms.addSelector("MyQueue", "topic = '<second:30>'");

To request a message be sent to MyQueue at 15 and 45 minutes past each hour, use this syntax:

cms.addSelector("MyQueue", "topic = '<minute:15>' and '<minute:45>'");

For information on how to add selectors using Jaguar Manager, see Chapter 8, "Setting up the Message Service," in the EAServer System Administration Guide.

Scheduled messages are delivered to queues with the appropriate selectors within milliseconds of the specified time. The time at which a component receives a message from the queue, however, depends on the number of unprocessed messages in the queue.

Scheduling variables

Scheduled message topic names can be either '<minute:NN>' or '<second:NN>'. Additional constraints must include a variable name and value. You can use these variables to define the message topic subscriptions:

Variable	Definition
MINUTE	Number of minutes past the hour
SECOND	Number of seconds past the minute
HOUR_OF_DAY	To specify 5 PM, "HOUR_OF_DAY = 17"
HOUR	Twice a day, at 6 AM and 6 PM, "HOUR = 6"
YEAR	Four-digit year, for example, 2000
MONTH	The name of the month, for example, January, February, and so forth
DATE	Date of the month, 1-31
DAY_OF_MONTH	Same as DATE
DAY_OF_WEEK	The name of the day of the week, for example, Monday, Tuesday, and so forth.
DAY_OF_WEEK_IN_MONTH	To specify the first Sunday in October, MONTH = October and DAY_OF_WEEK = Sunday and DAY_OF_WEEK_IN_MONTH = 1
DAY_OF_YEAR	To specify February 1, "DAY_OF_YEAR = 32"
WEEK_OF_MONTH	The week number within the current month
WEEK_OF_YEAR	The week number within the current year

The variable names are not case sensitive; minute and MINUTE are equivalent. You can find documentation for the variables, whose names correspond to the constants in the Java class java.util.Calendar, in the Java API Specification .

Scheduled messages are not saved to persistent storage and they are not replicated.

A scheduled message includes two properties that indicate the message creation time, which can be accessed by the component:

Property name	Datatype	Format
@t	double	Number of milliseconds since 1 January 1970
ts	string	YYYY-MM-DD HH:MM:SS

You can find the message structure definition in $JAGUAR/html/ir/CtsComponents.html.

EAServer message service CORBA API

The EAServer CORBA API includes:

MessageService The CtsComponents::MessageService interface allows EAServer clients and components to send and publish messages, register for message topic notification, and manage message queue and message topic properties. See $JAGUAR/html/ir/CtsComponents__MessageService.html for the API definitions and examples of how to use the MessageService interface.
MessageQueue The CtsComponents::MessageQueue interface allows clients to receive messages from a queue, get a list of messages in a queue, acknowledge the receipt of messages, close a message queue object, and recover messages that have been received but not acknowledged. See $JAGUAR/html/ir/CtsComponents__MessageQueue.html for the API definitions and examples of how to use the MessageQueue interface.
MessageListener The CtsComponents::MessageListener interface allows an application component to be notified when new messages are sent or published to its message queue or topic. See $JAGUAR/html/ir/CtsComponents__MessageListener.html for the API definition and an example of how to use the MessageListener interface.