Creating Applications Using JMS , 5 of 8

Publish-Subscribe Model Features

Topic

JMS has various features that allow you to develop an application based on a publish-subscribe model. The aim of this application model is to enable flexible and dynamic communication between applications functioning as publishers and applications playing the role of subscribers. The specific design point is that the applications playing these different roles should be decoupled in their communication, that they should interact based on messages and message content.

In distributing messages, publisher applications do not have to explicitly handle or manage message recipients. This allows for the dynamic addition of new subscriber applications to receive messages without changing any publisher application logic. Subscriber applications receive messages based on message content without regard to which publisher applications are sending messages. This allows the dynamic addition of subscriber applications without changing any subscriber application logic. Subscriber applications specify interest by defining a rule-based subscription on message properties and/or the message content of a topic. The system automatically routes messages by computing recipients for published messages using the rule-based subscriptions.

In the Publish-Subscribe model, merssages are published to and received from topics. A topic is created using the CreateTopic method in an AQjmsSession. A client may obtain a handle to a previously-created Topic using the getTopic methodin AQjmsSession.

You use the publish-subscribe model of communication in JMS by taking the following steps:

Enable enqueue/dequeue on the Topic using the start call in AQjmsDestination.
Set up one or more topics to hold messages. These topics should represent an area or subject of interest. For example, a topic can be used to represent billed orders.
Create a set of Durable Subscribers. Each subscriber may specify a selector that represents a specification (selects) for the messages that the subscriber wishes to receive. A null selector indicates that the subscriber wishes to receive all messages published on the topic
Subscribers may be local or remote. Local subscribers are durable subscribers defined on the same topic on which the message is published. Remote subscribers are other topics, or recipients on other topics that are defined as subscribers to a particular queue. In order to use remote subscribers, you must set up propagation between the two local and remote topic. For details on propagation, see: Chapter 9, "Administrative Interface".S
Create TopicPublishers using the session's createPublisher method Messages are published using the publish call. Messages may be published to all subscribers to the topic or to a specified subset of recipients on the topic
Subscribers may receive messages on the topic by using the receive method
Subscribers may also receive messages asynchronously by using Message Listeners. The concepts of Remote Subscribers and Propagation are Oracle extensions to JMS.

Example Scenario

In the BooksOnline application all booked orders are published to the OE_bookedorders_topic. Orders for customers in the eastern region are routed to the ES.ES_bookedorders_topic and those for the western region are routed to the WS.WS_bookedorders_topic. There is also another application that subscribes to the OE_bookedorders_topic to track messages for some important customers. Refer to the code examples in teh following sections.

Durable Subscriber

Durable Subscribers are instituted in either of the following ways:

A client uses the session's createDurableSubscriber method to create durable subscribers.
A DurableSubscriber is be created with a message selector. This allows the client to restrict messages delivered to the subscriber to those that match the selector.

The selector for topics containing payloads of type TextMessage, StreamMessage, BytesMessage, ObjectMessage, MapMessage can contain any expression which has a combination of one or more of the following:

JMS Message header fields or properties:

JMSPriority < 3 AND JMSCorrelationID = 'Fiction'

User defined message properties:

color IN ('RED', BLUE', 'GREEN') AND price < 30000

For topics containing AdtMessages the selector must be a SQL expression on the message payload contents or priority or correlationID.

Selector on priority or correlation is specified as follows
```
priority < 3 AND corr_id = 'Fiction'
```

Selector on message payload is specified as follows

tab.user_data.color = 'GREEN' AND tab.user_data.price < 30000

The syntax for the selector is described in detail in the Oracle8i Supplied Java Packages Reference (createDurableSubscriber use case).

Remote subscribers are defined using the createRemoteSubscriber call.The remote subscriber may be a specific consumer at the remote topic or all subscribers at the remote topic

A remote subscriber is defined using the AQjmsAgent structure. An AQjmsAgent consists of a name and address. The name refers to the consumer_name at the remote topic. The address refers to the remote topic:

<schema>.<topic_name>[@dblink]

To publish messages to a particular consumer at the remote topic, the subscription_name of the recipient at the remote topic must be specified in the name field of AQjmsAgent. The remote topic must be specified in the address field of AQjmsAgent.
To publish messages to all subscribers of the remote topic, the name field of AQjmsAgent must be set to null. The remote topic must be specified in the address field of AQjmsAgent.

In the BooksOnline application there is one local subscriber SUBS1 and two remote subscribers -

West_Shipping at the remote topic WS.WS_bookedorders_topic
East_Shipping at ES.ES_booked_orders_topic

Example Code

 public void create_booked_orders_subscribers(TopicSession jms_session)
{
   Topic            topic;
   TopicSubscriber  tsubs;
   AQjmsAgent       agt_east;
   AQjmsAgent       agt_west;

   try
   {
   
     /* get a handle to the OE_bookedorders_topic */
     topic = ((AQjmsSession)jms_session).getTopic("OE", 
                    "OR_bookedorders_topic");
     
     
     /* Create local subscriber - to track messages for some customers */
     tsubs = jms_session.createDurableSubscriber(topic, "SUBS1",
             "JMSPriority < 3 AND Customer = 'MARTIN'", 
                   false);
     
     /* Create remote subscribers in the western and eastern region */   
     agt_west = new AQjmsAgent("West_Shipping", "WS.WS_bookedorders_topic");
     
     ((AQjmsSession)jms_session).createRemoteSubscriber(topic, agt_west, 
                     "Region = 'WESTERN'");

     agt_east = new AQjmsAgent("East_Shipping", "ES.ES_bookedorders_topic");
     
     ((AQjmsSession)jms_session).createRemoteSubscriber(topic, agt_east, 
                     "Region = 'EASTERN'");

     /* schedule propagation between bookedorders_topic and 
   WS_bookedorders_topic, ES.ES_bookedorders_topic */   
     ((AQjmsDestination)topic).schedulePropagation(jms_session, 
                      "WS.WS_bookedorders_topic", 
                         null, null, null, null);

     ((AQjmsDestination)topic).schedulePropagation(jms_session, 
                   "ES.ES_bookedorders_topic", 
                      null, null, null, null);
   }   
   catch (Exception ex)
   {
     System.out.println("Exception " + ex);
   }
      
}

Topic Publisher

Messages are published using TopicPublisher:

A TopicPublisher is created by passing a Topic to a session's createPublisher method. A client also has the option of creating a TopicPublisher without supplying a Topic. In this case, a Topic must be specified on every publish operation. A client can specify a default delivery mode, priority and time-to-live for all messages sent by the TopicPublisher. It can also specify these options on a per message basis.

Example Scenario and Code

In the BooksOnline application, booked orders are published to the OE.OE_bookedorders_topic

public void book_new_order(TopicSession jms_session, ObjectMessage obj_message)
{
   TopicPublisher  publisher;   
   Topic           topic;

   try
   {
     /* get a handle to the booked_orders topic */   
     topic = ((AQjmsSession) jms_session).getTopic("OE",
                     "OE_bookedorders_topic");
     
     publisher = jms_session.createPublisher(topic);   
     
     publisher.publish(topic, obj_message);
     
     jms_session.commit();   
   }
   catch (JMSException ex)
   {
     System.out.println("Exception: " + ex);
   }

}

In the BooksOnline application, each shipping region receives messages from the corresponding booked orders topic (WS_bookedorder_topic or ES_bookedorder_topic). The local subscriber SUBS1 receives messages from the OE_booked_orders_topic.


public void get_martins_orders(TopicSession jms_session)
{
   Topic            topic;
   TopicSubscriber  tsubs;
   ObjectMessage    obj_message;
   BolCustomer      customer;
   BolOrder         new_order;
   String           state;
   int              i = 0;
   
   try
   {
     /* get a handle to the OE_bookedorders_topic */
     topic = ((AQjmsSession)jms_session).getTopic("OE", 
                  "OE_bookedorders_topic");
   

     /* Create local subscriber - to track messages for some customers */
     tsubs = jms_session.createDurableSubscriber(topic, "SUBS1",
                "JMSPriority < 3 AND Customer = 'MARTIN'",
                   false);

     /* process 10 messages */
     for(i=0; i<10; i++)
     {
       /* wait for a message to show up in the topic */
       obj_message = (ObjectMessage)tsubs.receive(10);
       
       new_order = (BolOrder)obj_message.getObject();
       
       customer = new_order.getCustomer();
       state    = customer.getState();
       
       System.out.println("Order: " + i + " for customer " + 
           customer.getName()); 
       jms_session.commit();      
     }    
   }
   catch (Exception ex)
   {
     System.out.println("Exception " + ex); 
   }
}

Recipient Lists

In the JMS publish-subscribe model, clients can specify explicit recipient lists instead of having messages sent to all the subscribers of the topic. These recipients may or may not be existing subscribers of the topic. The recipient list overrides the subscription list on the topic for this message. The concept of recipient lists is an Oracle extension to JMS.

Example Scenario and Code

Suppose we want to send high priority messages only to SUBS1 and Fedex_Shipping in the Eastern region instead of publishing them to all the subscribers of the OE_bookedorders_topic:

public void book_rush_order(TopicSession jms_session, 
             ObjectMessage obj_message)
{
     
   TopicPublisher  publisher;   
   Topic           topic;
   AQjmsAgent[]    recp_list = new AQjmsAgent[2];   

   try
   {
     /* get a handle to the booked_orders topic */   
     topic = ((AQjmsSession) jms_session).getTopic("OE",
                     "OE_bookedorders_topic");
     
     publisher = jms_session.createPublisher(null);   
     
     recp_list[0] = new AQjmsAgent("SUBS1", null);
     recp_list[1] = new AQjmsAgent("Fedex_Shipping", 
               "ES.ES_bookedorders_topic");
     
     publisher.setPriority (1);
     ((AQjmsTopicPublisher)publisher).publish(topic, obj_message, recp_list);
     
     jms_session.commit();   

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

TopicReceiver

If the recipient name is explicitly specified in the recipient list, but that recipient is not a subscriber to the queue, then messages sent to it can be received by creating a TopicReceiver.TopicReceiver is an Oracle extension to JMS.

Example Scenario and Code

public void ship_rush_orders(TopicSession jms_session)
{
   Topic            topic;
   TopicReceiver    trec;
   ObjectMessage    obj_message;
   BolCustomer      customer;
   BolOrder         new_order;
   String           state;
   int              i = 0;   

   try
   {
     /* get a handle to the OE_bookedorders_topic */
     topic = ((AQjmsSession)jms_session).getTopic("ES", 
                    "ES_bookedorders_topic");
   

     /* Create local subscriber - to track messages for some customers */
     trec = ((AQjmsSession)jms_session).createTopicReceiver(topic, 
                         "Fedex_Shipping", 
                         null);
     
     /* process 10 messages */
     for(i = 0; i < 10; i++)
     {
       /* wait for a message to show up in the topic */
       obj_message = (ObjectMessage)trec.receive(10);
       
       new_order = (BolOrder)obj_message.getObject();
       
       customer = new_order.getCustomer();
       state    = customer.getState();
       
       System.out.println("Rush Order for customer " + 
           customer.getName()); 
       jms_session.commit();      
     }    
   }
   catch (Exception ex)
   {
     System.out.println("Exception ex: " + ex);
   }

}

For remote subscribers - if the subscriber name at the remote topic has explicitly been specified in the createRemoteSubscriber call, then to receive a message, we can use TopicReceivers

public void get_westernregion_booked_orders(TopicSession jms_session)
{
   Topic            topic;
   TopicReceiver    trec;
   ObjectMessage    obj_message;
   BolCustomer      customer;
   BolOrder         new_order;
   String           state;
   int              i = 0;   

   try
   {
     /* get a handle to the WS_bookedorders_topic */
     topic = ((AQjmsSession)jms_session).getTopic("WS", 
                    "WS_bookedorders_topic");
   

     /* Create local subscriber - to track messages for some customers */
     trec = ((AQjmsSession)jms_session).createTopicReceiver(topic, 
                         "West_Shipping", 
                         null);

     /* process 10 messages */
     for(i = 0; i < 10; i++)
     {
       /* wait for a message to show up in the topic */
       obj_message = (ObjectMessage)trec.receive(10);
       
       new_order = (BolOrder)obj_message.getObject();
       
       customer = new_order.getCustomer();
       state    = customer.getState();
       
       System.out.println("Received Order for customer " + 
           customer.getName()); 
       jms_session.commit();      
     }    
   }
   catch (Exception ex)
   {
     System.out.println("Exception ex: " + ex); 
   }

}

If the subscriber name is not specified in the createRemoteSubscriber call, clients have to use durable subscribers at the remote site to receive messages