Oracle8i Application Developer's Guide - Advanced Queuing
Release 2 (8.1.6)

Part Number A76938-01

Library

Product

Contents

Index

Go to previous page Go to beginning of chapter Go to next page

Creating Applications Using JMS , 6 of 8


Message Producer Features

Priority and Ordering of Messages

The message ordering dictates the order in which messages will be received from a queue or topic. The ordering method is specified when the queue table for the queue or topic is created (see "Create a Queue Table" in Chapter 9, "Administrative Interface"). Currently, AQ supports ordering on two of the message attributes:

When combined, they lead to four possible ways of ordering:

FIFO Ordering of Messages

If enqueue time was chosen as the ordering criteria, then messages are received in the order of the enqueue time. The enqueue time is assigned to the message by AQ at message publish/send time. This is also the default ordering.

Priority Ordering of Messages.

If priority ordering is chosen, each message will be assigned a priority. Priority can be specified as a message property at publish/send time by the Message Producer. The messages will be received in the order of the priorities assigned.

First-In, First-Out (FIFO) Priority Ordering.

A FIFO-priority topic/queue can also be created by specifying both the priority and the enqueue time as the sort order of the messages. A FIFO-priority topic/queue behaves like a priority queue, except if two messages are assigned the same priority, they will be received in the order of their enqueue time.

Enqueue Time Followed by Priority.

Messages with the same enqueue time will be received according to their priorities. If the ordering criteria of two message is the same, then the order in which they are received is indeterminate. However, AQ does ensure that messages send/published in the same session with the same ordering criteria will be received in the order in which they were sent.

Example Scenario and Code

Using the BooksOnLine application, a customer can request

Priority can be specified at the Message Producer level using the setPriority call, or during the send or publish call. Teh latter overrireds the former.

The Order Entry application uses a FIFO queue to store new orders. New orders are processed by the order entry application and published to the booked orders topic. The order entry application will retrieve messages from the new orders queue in the order of their enqueue time. It uses a FIFO-priority topic to store booked orders. Booked orders are propagated to the regional booked orders topics. At each region, orders in these regional booked orders topics are processed in the order of the shipping priorities. The following calls create the FIFO-priority topic for the Order Entry application to store booked orders.

public static void createPriorityTopic(TopicSession jms_session)
{
    AQQueueTableProperty       qt_prop;
    AQQueueTable               pr_qtable;   
    AQjmsDestinationProperty   dest_prop;
    Topic                      bookedorders_topic;   

    try
    {

   /* Create a priority queue table for OE */
   qt_prop = new AQQueueTableProperty("SYS.AQ$_JMS_OBJECT_MESSAGE");
   qt_prop.setComment("Order Entry Priority " + 
             "MultiConsumer Orders queue table");
   qt_prop.setCompatible("8.1");
   qt_prop.setMultiConsumer(true);

   /* Set a FIFO-priority order */
   qt_prop.setSortOrder("priority, enq_time");

   pr_qtable = ((AQjmsSession)jms_session).createQueueTable("OE",
                     "OE_orders_pr_mqtab", qt_prop);
     
   /* Create a Queue in this queue table */
   dest_prop = new AQjmsDestinationProperty();
   
   bookedorders_topic =((AQjmsSession)jms_session).createTopic(pr_qtable, 
              "OE_bookedorders_topic", dest_prop);
   
   /* Enable enqueue and dequeue on the topic */
   ((AQjmsDestination)bookedorders_topic).start(jms_session, true, true);

  
} catch (Exception ex) { System.out.println("Exception: " + ex); } } /* When an order arrives, the order entry application can use the following procedure to publish the order into its booked orders topic. A shipping priority is specified for each order: */ public static void order_enqueue(TopicSession jms_session, String book_title, int book_qty, int order_num, int cust_no, String cust_name, int ship_priority, String cust_state, String cust_country, String cust_order_type) { BolOrder order; BolCustomer cust_data; BolBook book_data; BolOrderItem[] item_list; Topic topic; ObjectMessage obj_message; TopicPublisher tpub; try { book_data = new BolBook(book_title); cust_data = new BolCustomer(cust_no, cust_name); order = new BolOrder(order_num, cust_data); item_list = new BolOrderItem[1]; item_list[0] = new BolOrderItem(book_data, book_qty); order.setItemList(item_list); /* get a handle to the OE bookedorders_topic */ topic = ((AQjmsSession)jms_session).getTopic("OE", "OE_bookedorders_topic"); /* Create the topic publisher */ tpub = jms_session.createPublisher(topic); obj_message = jms_session.createObjectMessage(); obj_message.setObject(order); /* Send message - specify priority */ tpub.publish(topic, obj_message, DeliveryMode.PERSISTENT, ship_priority,0); jms_session.commit(); } catch (Exception ex) { System.out.println("Exception ex: " + ex); } }

Time Specification - Delay

Messages can be sent/published to a queue/topic with Delay. The delay represents a time interval after which the message becomes available to the Message Consumer. A message specified with a delay is in a waiting state until the delay expires and the message becomes available. Delay for a message is specified as message property (JMS_OracleDelay). This property is not specified in the JMS standard. It is an AQ extension to JMS message properties.

Delay processing requires the AQ background process, the queue monitor to be started. Note also that receiving by msgid overrides the delay specification.

Example Scenario and Code

In the BooksOnLine application, delay can be used to implement deferred billing. The billing application defines a queue in which shipped orders that are not billed immediately are placed with a delay. For example, a certain class of customer accounts, such as corporate customers, may not be billed for 15 days. The billing application deliquesce incoming shipped order messages (from the shipped orders queue) and if the order is for a corporate customer, this order is enqueued into a deferred billing queue with a delay. Delay works similarly for publish, though a scenario has not been provided.

public static void defer_billing(QueueSession jms_session, 
             BolOrder deferred_order)
{
    Queue             def_bill_q;   
    ObjectMessage     obj_message;
    QueueSender       qsender;

    try
    {
   /* get a handle to the deferred billing queue */
   def_bill_q = ((AQjmsSession)jms_session).getQueue("CBADM", 
                   "deferbilling_que");

   /* Create the QueueSender */
   qsender = jms_session.createSender(def_bill_q);

   obj_message = jms_session.createObjectMessage();
   obj_message.setObject(deferred_order);

   /* Set Delay as 15 days 
    * Delay is specified in seconds
         */
   obj_message.setIntProperty("JMS_OracleDelay", 15*60*60*24);

   qsender.send(obj_message);

   jms_session.commit();

    }
    catch (Exception ex)
    {
   System.out.println("Exception " + ex);
    }
    
}

Time Specification - Expiration

Producers of messages can specify expiration limits, or Time-to-Live (coded as TimeToLive) for messages. This defines the period of time the message is available for a Message Consumer.

Time-to-Live can be specified at send/publish time or usding the set TimeToLive method of a Message Producer, with the former overriding the latter. Note that the AQ background process, the queue monitor must be running to implement Time-to-Live.

Example Scenario

In the BooksOnLine application, TimeToLive can be used to control the amount of time that is allowed to process a back order. The shipping application places orders for books that are not available on a back order topic. If the shipping policy is that all back orders must be shipped within a week, then messages can be published into the back order topic with an expiration of one week. In this case, any back orders that are not processed within one week are moved to the exception topic with the message state set to EXPIRED. This can be used to flag any orders that have not been shipped according to the back order shipping policy.

Example Code

/* Re-enqueue a back order into a back_order Topic and set a timeToLive of 
   7 days; 
   All back orders must be processed in 7 days or they are moved to the 
   exception queue */ 
public static void requeue_back_order(TopicSession jms_session, 
                  String sale_region, BolOrder back_order)
{
    Topic             back_order_topic;   
    ObjectMessage     obj_message;
    TopicPublisher    tpub;
    long              timetolive;

    try
    {
   /* Look up a back order topic based on the region */
   if(sale_region.equals("WEST"))
   {
       back_order_topic = ((AQjmsSession)jms_session).getTopic("WS", 
                     "WS_backorders_topic");
   }
   else if(sale_region.equals("EAST"))
   {
       back_order_topic = ((AQjmsSession)jms_session).getTopic("ES", 
                     "ES_backorders_topic");
   }
   else
   {
       back_order_topic = ((AQjmsSession)jms_session).getTopic("OS", 
                     "OS_backorders_topic");
   }

   obj_message = jms_session.createObjectMessage();
   obj_message.setObject(back_order);

   tpub = jms_session.createPublisher(null);


   /* Set message expiration to 7 days: */ 
   timetolive = 7*60*60*24*1000;          // specified in  milliseconds 


   /* Publish the message */
   tpub.publish(back_order_topic, obj_message, DeliveryMode.PERSISTENT,
           1, timetolive);


   jms_session.commit();
    }
    catch (Exception ex)
    {
   System.out.println("Exception :" + ex); 
    }
}

Message Grouping

Messages belonging to a queue/topic can be grouped to form a set that can only be consumed by one consumer at a time. This requires the queue/topic be created in a queue table that is enabled for transactional message grouping (see "Create a Queue Table", Chapter 12, "Creating Applications Using JMS"). All messages belonging to a group have to be created in the same transaction and all messages created in one transaction belong to the same group. This feature allows you to segment a complex message into simple messages. This is an AQ extension and not part of the JMS specification.

For example, messages directed to a queue containing invoices could be constructed as a group of messages starting with the header message, followed by messages representing details, followed by the trailer message. Message grouping is also very useful if the message payload contains complex large objects such as images and video that can be segmented into smaller objects.

The general message properties (priority, delay, expiration) for the messages in a group are determined solely by the message properties specified for the first message (head) of the group irrespective of which properties are specified for subsequent messages in the group.

The message grouping property is preserved across propagation. However, it is important to note that the destination topic to which messages have to be propagated must also be enabled for transactional grouping. There are also some restrictions you need to keep in mind if the message grouping property is to be preserved while dequeuing messages from a queue enabled for transactional grouping (see "Dequeue Methods" and "Modes of Dequeuing" for additional information).

Example Scenario

In the BooksOnLine application, message grouping can be used to handle new orders. Each order contains a number of books ordered one by one in succession. Items ordered over the Web exhibit similar behavior.

In the example given below, each send corresponds to an individual book that is part of an order, and the group/transaction represents a complete order. Only the first message contains customer information. Note that the OE_neworders_que is defined in the queue table OE_orders_sqtab which has been enabled for transactional grouping.

Example Code

public static void createMsgGroupQueueTable(QueueSession jms_session)
{
    AQQueueTableProperty       sqt_prop;
    AQQueueTable               sq_table;   
    AQjmsDestinationProperty   dest_prop;
    Queue                      neworders_q;   

    try
    {
   /* Create a single-consumer orders queue table 
         * with message grouping = TRANSACTIONAL 
         */
   sqt_prop = new AQQueueTableProperty("BOLADM.order_typ");
   sqt_prop.setComment("Order Entry Single-Consumer Orders queue table");
   sqt_prop.setCompatible("8.1");
   sqt_prop.setMessageGrouping(AQQueueTableProperty.TRANSACTIONAL);

   sq_table = ((AQjmsSession)jms_session).createQueueTable("OE", 
                  "OE_orders_sqtab", sqt_prop);

    
   /* Create new orders queue for OE */
   dest_prop = new AQjmsDestinationProperty();
   neworders_q = ((AQjmsSession)jms_session).createQueue(sq_table, 
                            "OE_neworders_que", 
                        dest_prop);   
     
    }
    catch (Exception ex)
    {
   System.out.println("Exception: " + ex); 
    }
}


/* This method send an order to the specified queue */
public static void enqueue_order(QueueSession jms_session, Queue queue,
                  int order_num, String cust_name, int cust_id, 
             int book_qty, String book_title)
{
   QueueSender       sender;
   ObjectMessage     obj_message;   
   BolOrder          order;
   BolCustomer       cust_data=null;
   BolBook           book_data;
   BolOrderItem[]    item_list;

   try
   {
     book_data = new BolBook(book_title);
     
     if(cust_name != null)
     {    
       cust_data = new BolCustomer(cust_id, cust_name);
     }

     order = new BolOrder(order_num, cust_data);
     
     item_list = new BolOrderItem[1];
     item_list[0] = new BolOrderItem(book_data, book_qty);
     
     order.setItemList(item_list);
     
     
     sender = jms_session.createSender(queue);
     
     obj_message = jms_session.createObjectMessage();
     
     obj_message.setObject(order);
     
     sender.send(obj_message);
   }
   catch (Exception ex)
   {
     System.out.println("Exception ex: " + ex); 
   }
}   


/* Enqueue groups of orders */
public static void enqueue_order_groups(QueueSession jms_session)
{
  Queue neworders_q;

  try
  {
    neworders_q = ((AQjmsSession)jms_session).getQueue("OE",
                         "OE_neworders_que"); 

    /* Enqueue first group */   
    enqueue_order(jms_session, neworders_q, 1, "John", 1000, 2,
        "John's first book");
    
    enqueue_order(jms_session, neworders_q, 1, null, 0, 1, 
        "John's second book");
    
    jms_session.commit();
    
    /* Enqueue second group */   
    enqueue_order(jms_session, neworders_q, 2, "Mary", 1001, 1,
        "Mary's first book");
    
    enqueue_order(jms_session, neworders_q, 2, null, 0, 1, 
        "Mary's second book");
    
    enqueue_order(jms_session, neworders_q, 2, null, 0, 1, 
        "Mary's third book");
    
    
    jms_session.commit();
    
    
    /* Enqueue third group */   
    enqueue_order(jms_session, neworders_q, 3, "Scott", 1002, 1,
        "Scott's first book");
    
    enqueue_order(jms_session, neworders_q, 3, null, 0, 2, 
        "Scott's second book");
    
    enqueue_order(jms_session, neworders_q, 3, null, 0, 2,
        "Scott's third book");
    
    
    jms_session.commit();
  }
  catch (Exception ex)
  {
    System.out.println("Exception ex: " + ex);
  }


}


Go to previous page Go to beginning of chapter Go to next page
Oracle
Copyright © 1996-2000, Oracle Corporation.

All Rights Reserved.

Library

Product

Contents

Index